This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 20d733e735b feat: allow non-sealed schemas for MSQ when using base
table spec (#19845)
20d733e735b is described below
commit 20d733e735b7b4691624b93e4efba760ac6525ee
Author: Clint Wylie <[email protected]>
AuthorDate: Mon Aug 24 14:49:32 2026 -0700
feat: allow non-sealed schemas for MSQ when using base table spec (#19845)
---
.../destination/SegmentGenerationUtils.java | 66 +++++++++-
.../org/apache/druid/msq/exec/MSQInsertTest.java | 130 +++++++++++++++++++
.../input/impl/AdaptedBaseTableProjectionSpec.java | 12 ++
.../data/input/impl/BaseTableProjectionSpec.java | 5 +
...lusteredValueGroupsBaseTableProjectionSpec.java | 37 ++++++
...eredValueGroupsBaseTableProjectionSpecTest.java | 141 +++++++++++++++++++++
.../druid/catalog/model/table/DatasourceDefn.java | 13 --
.../catalog/model/table/DatasourceTableTest.java | 8 +-
.../druid/segment/indexing/DataSchemaTest.java | 29 +++++
9 files changed, 422 insertions(+), 19 deletions(-)
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/destination/SegmentGenerationUtils.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/destination/SegmentGenerationUtils.java
index f0b9ca01c25..87f3065afae 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/destination/SegmentGenerationUtils.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/destination/SegmentGenerationUtils.java
@@ -28,6 +28,7 @@ import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.LongDimensionSchema;
import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.frame.key.ClusterBy;
import org.apache.druid.frame.key.KeyColumn;
import org.apache.druid.indexer.granularity.ArbitraryGranularitySpec;
@@ -66,6 +67,7 @@ import org.apache.druid.utils.CollectionUtils;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -104,8 +106,14 @@ public final class SegmentGenerationUtils
if (destination.getBaseTable() != null) {
final Granularity queryGranularity =
query.context().getGranularity(DruidSqlInsert.SQL_INSERT_QUERY_GRANULARITY,
jsonMapper);
- final BaseTableProjectionSpec baseTable =
+ final BaseTableProjectionSpec declared =
destination.getBaseTable().withQueryGranularity(queryGranularity);
+ // The query may produce columns the base table does not declare, which
happens when the target table does not
+ // require a strict schema. Store them rather than drop them; they are
appended after the declared columns, so
+ // the shape the operator asked for is unchanged.
+ final BaseTableProjectionSpec baseTable = declared.withAdditionalColumns(
+ undeclaredColumns(declared, querySignature, columnMappings, query,
destination.getDimensionSchemas())
+ );
return DataSchema.builder()
.withDataSource(destination.getDataSource())
.withTimestamp(new
TimestampSpec(ColumnHolder.TIME_COLUMN_NAME, "millis", null))
@@ -212,6 +220,62 @@ public final class SegmentGenerationUtils
&&
!query.context().getBoolean(GroupByQueryConfig.CTX_KEY_ENABLE_MULTI_VALUE_UNNESTING,
true);
}
+ /**
+ * The columns a query produces that a base table does not declare, in query
output order, as the
+ * {@link DimensionSchema}s they should be stored with.
+ * <p>
+ * A base table declares the columns an operator asked for. When the target
table does not require a strict schema,
+ * a query may produce others; they are stored so that ingesting a column is
never silently a no-op.
+ */
+ private static List<DimensionSchema> undeclaredColumns(
+ final BaseTableProjectionSpec baseTable,
+ final RowSignature querySignature,
+ final ColumnMappings columnMappings,
+ final Query<?> query,
+ @Nullable final Map<String, DimensionSchema> dimensionSchemas
+ )
+ {
+ final Set<String> declared = new HashSet<>();
+ for (DimensionSchema dimension :
baseTable.getDimensionsSpec().getDimensions()) {
+ declared.add(dimension.getName());
+ }
+ for (AggregatorFactory metric : baseTable.getMetrics() == null ? new
AggregatorFactory[0] : baseTable.getMetrics()) {
+ declared.add(metric.getName());
+ }
+ // The time column is positional in a base table, never appended.
+ declared.add(ColumnHolder.TIME_COLUMN_NAME);
+
+ final List<DimensionSchema> undeclared = new ArrayList<>();
+ for (final String outputColumnName :
columnMappings.getOutputColumnNames()) {
+ if (!declared.add(outputColumnName)) {
+ continue;
+ }
+ final int outputColumn = CollectionUtils.getOnlyElement(
+ columnMappings.getOutputColumnsByName(outputColumnName),
+ xs -> new ISE("Expected single output column for name [%s], but got
[%s]", outputColumnName, xs)
+ );
+ final String queryColumn =
columnMappings.getQueryColumnName(outputColumn);
+ final ColumnType type =
+ querySignature.getColumnType(queryColumn)
+ .orElseThrow(() -> new ISE("No type for column [%s]",
outputColumnName));
+
+ if (type.is(ValueType.COMPLEX)) {
+ final String typeName = type.getComplexTypeName();
+ if (typeName == null ||
!DimensionHandlerUtils.DIMENSION_HANDLER_PROVIDERS.containsKey(typeName)) {
+ // A base table stores columns as dimensions, so a complex type with
no dimension handler has nowhere to go
+ throw InvalidInput.exception(
+ "Column [%s] has type [%s], which cannot be stored in a base
table that does not declare it. Declare the"
+ + " column in the table, or cast it to a type that can be stored
as a dimension",
+ outputColumnName,
+ type
+ );
+ }
+ }
+ undeclared.add(getDimensionSchema(outputColumnName, type,
query.context(), dimensionSchemas));
+ }
+ return undeclared;
+ }
+
private static DimensionSchema getDimensionSchema(
final String outputColumnName,
@Nullable final ColumnType queryType,
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java
index c4708d0904b..28ff630363f 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQInsertTest.java
@@ -280,6 +280,17 @@ public class MSQInsertTest extends MSQTestBase
)
.buildSpec()
);
+ metadataCatalog.addSpec(
+ TableId.datasource("fooClusteredUnsealed"),
+ // Not sealed: the table declares the layout it cares about
(clustering column, time position, delta) and lets
+ // a query bring whatever else it produces.
+ TableBuilder.datasource("fooClusteredUnsealed",
Granularities.DAY.toString())
+ .column("channel", Columns.SQL_VARCHAR)
+ .timeColumn()
+ .column("delta", Columns.SQL_BIGINT)
+ .baseTable(new
ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("channel"), null, null))
+ .buildSpec()
+ );
return new LiveCatalogResolver(metadataCatalog);
}
@@ -875,6 +886,125 @@ public class MSQInsertTest extends MSQTestBase
.verifyResults();
}
+ @MethodSource("data")
+ @ParameterizedTest(name = "{index}:with context {0}")
+ public void
testInsertOnExternalDataSourceWithUnsealedCatalogClusteredBaseTable(
+ String contextName,
+ Map<String, Object> context
+ ) throws IOException
+ {
+ final File toRead = getResourceAsTemporaryFile("/wikipedia-sampled.json");
+ final String toReadFileNameAsJson =
queryFramework().queryJsonMapper().writeValueAsString(toRead.getAbsolutePath());
+
+ // The table declares only channel, __time and delta. The other four
columns the query produces are appended after
+ // them in query output order, so the declared layout (clustering prefix,
time position) is unchanged and nothing
+ // the query selected is dropped.
+ RowSignature rowSignature = RowSignature.builder()
+ .add("channel", ColumnType.STRING)
+ .add("__time", ColumnType.LONG)
+ .add("delta", ColumnType.LONG)
+ .add("page", ColumnType.STRING)
+ .add("user", ColumnType.STRING)
+ .add("added", ColumnType.LONG)
+ .add("deleted", ColumnType.LONG)
+ .build();
+
+ testIngestQuery().setSql(" insert into fooClusteredUnsealed SELECT\n"
+ + " floor(TIME_PARSE(\"timestamp\") to minute)
AS __time,\n"
+ + " channel,\n"
+ + " page,\n"
+ + " user,\n"
+ + " added,\n"
+ + " deleted,\n"
+ + " delta\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{ \"files\": [" + toReadFileNameAsJson +
"],\"type\":\"local\"}',\n"
+ + " '{\"type\": \"json\"}',\n"
+ + " '[{\"name\": \"timestamp\", \"type\":
\"string\"}, {\"name\": \"channel\", \"type\": \"string\"}, {\"name\":
\"page\", \"type\": \"string\"}, {\"name\": \"user\", \"type\": \"string\"},
{\"name\": \"added\", \"type\": \"long\"}, {\"name\": \"deleted\", \"type\":
\"long\"}, {\"name\": \"delta\", \"type\": \"long\"}]'\n"
+ + " )\n"
+ + ") PARTITIONED by day ")
+ .setExpectedDataSource("fooClusteredUnsealed")
+ .setExpectedRowSignature(rowSignature)
+ .setQueryContext(context)
+ .setExpectedSegments(ImmutableSet.of(SegmentId.of(
+ "fooClusteredUnsealed",
+ Intervals.of("2016-06-27/P1D"),
+ "test",
+ 0
+ )))
+ // The appended columns are not clustered on; clustering
is what the table declared.
+ .setExpectedClusterGroups(
+ new ClusterGroupTuples(
+ RowSignature.builder().add("channel",
ColumnType.STRING).build(),
+ ImmutableList.of(
+ ImmutableList.of("#ceb.wikipedia"),
+ ImmutableList.of("#de.wikipedia"),
+ ImmutableList.of("#en.wikipedia"),
+ ImmutableList.of("#es.wikipedia"),
+ ImmutableList.of("#id.wikipedia"),
+ ImmutableList.of("#pl.wikipedia"),
+ ImmutableList.of("#pt.wikipedia"),
+ ImmutableList.of("#ru.wikipedia"),
+ ImmutableList.of("#sh.wikipedia"),
+ ImmutableList.of("#sv.wikipedia"),
+ ImmutableList.of("#zh.wikipedia")
+ )
+ )
+ )
+ // Rows are read back in segment order, which now sorts
by delta ahead of the appended columns.
+ .setExpectedResultRows(
+ ImmutableList.of(
+ new Object[]{"#ceb.wikipedia", 1466985660000L,
4150L, "Neqerssuaq", "Lsjbot", 4150L, 0L},
+ new Object[]{"#de.wikipedia", 1466992920000L,
2560L, "Benutzer Diskussion:Squasher/Archiv/2016", "TaxonBot", 2560L, 0L},
+ new Object[]{"#de.wikipedia", 1466992980000L,
364L, "Benutzer Diskussion:HerrSonderbar", "GiftBot", 364L, 0L},
+ new Object[]{"#en.wikipedia", 1466985600000L,
-2L, "Richie Rich's Christmas Wish", "JasonAQuest", 0L, 2L},
+ new Object[]{"#en.wikipedia", 1466985600000L, 2L,
"Bailando 2015", "181.230.118.178", 2L, 0L},
+ new Object[]{"#en.wikipedia", 1466985660000L,
496L, "Panama Canal", "Mariordo", 496L, 0L},
+ new Object[]{"#en.wikipedia", 1466992980000L,
-463L, "File:Paint.net 4.0.6 screenshot.png", "Calvin Hogg", 0L, 463L},
+ new Object[]{"#es.wikipedia", 1466985660000L,
-173L, "Sumo (banda)", "181.110.165.189", 0L, 173L},
+ new Object[]{"#es.wikipedia", 1466989320000L, 4L,
"Clasificación para la Eurocopa Sub-21 de 2017", "Guly600", 4L, 0L},
+ new Object[]{"#id.wikipedia", 1466989320000L,
106L, "Ibnu Sina", "Ftihikam", 106L, 0L},
+ new Object[]{"#pl.wikipedia", 1466985600000L,
270L, "Kategoria:Dyskusje nad usunięciem artykułu zakończone bez konsensusu −
lipiec 2016", "Beau.bot", 270L, 0L},
+ new Object[]{"#pt.wikipedia", 1466992920000L,
1926L, "Dobromir Zhechev", "Ceresta", 1926L, 0L},
+ new Object[]{"#ru.wikipedia", 1466985720000L,
196L, "Википедия:Опросы/Унификация шаблонов «Не переведено»", "Wanderer777",
196L, 0L},
+ new Object[]{"#sh.wikipedia", 1466985660000L,
-1L, "El Terco, Bachíniva", "Kolega2357", 0L, 1L},
+ new Object[]{"#sh.wikipedia", 1466985720000L,
-1L, "Hermanos Díaz, Ascensión", "Kolega2357", 0L, 1L},
+ new Object[]{"#sh.wikipedia", 1466989320000L,
-1L, "El Sicomoro, Ascensión", "Kolega2357", 0L, 1L},
+ new Object[]{"#sh.wikipedia", 1466992920000L,
-1L, "Trinidad Jiménez G., Benemérito de las Américas", "Kolega2357", 0L, 1L},
+ new Object[]{"#sv.wikipedia", 1466985600000L,
31L, "Salo Toraut", "Lsjbot", 31L, 0L},
+ new Object[]{"#zh.wikipedia", 1466989320000L,
18L, "中共十八大以来的反腐败工作", "2001:DA8:207:E132:94DC:BA03:DFDF:8F9F", 18L, 0L},
+ new Object[]{"#zh.wikipedia", 1466992920000L,
1986L, "Wikipedia:頁面存廢討論/記錄/2016/06/27", "Tigerzeng", 1986L, 0L}
+ )
+ )
+ .verifyResults();
+ }
+
+ @MethodSource("data")
+ @ParameterizedTest(name = "{index}:with context {0}")
+ public void testInsertOnUnsealedCatalogClusteredBaseTableUnstorableColumn(
+ String contextName,
+ Map<String, Object> context
+ )
+ {
+ // A base table stores columns as dimensions, so an undeclared sketch
column has nowhere to go: it could only be
+ // stored as a metric, and a base table declares its own metrics.
+ testIngestQuery().setSql(
+ "insert into fooClusteredUnsealed "
+ + "select __time, dim1 as channel, cnt as delta,
unique_dim1 as unique_users "
+ + "from foo partitioned by day"
+ )
+ .setQueryContext(context)
+ .setExpectedExecutionErrorMatcher(
+ ThrowableMatcher.of(ISE.class).expectMessageContains(
+ "Column [unique_users] has type
[COMPLEX<hyperUnique>], which cannot be stored in a"
+ + " base table that does not declare it. Declare
the column in the table, or cast it"
+ + " to a type that can be stored as a dimension"
+ )
+ )
+ .verifyExecutionError();
+ }
+
@MethodSource("data")
@ParameterizedTest(name = "{index}:with context {0}")
public void testInsertOnFoo1WithGroupByLimitWithoutClusterBy(String
contextName, Map<String, Object> context)
diff --git
a/processing/src/main/java/org/apache/druid/data/input/impl/AdaptedBaseTableProjectionSpec.java
b/processing/src/main/java/org/apache/druid/data/input/impl/AdaptedBaseTableProjectionSpec.java
index ac59162a514..2828d2cda53 100644
---
a/processing/src/main/java/org/apache/druid/data/input/impl/AdaptedBaseTableProjectionSpec.java
+++
b/processing/src/main/java/org/apache/druid/data/input/impl/AdaptedBaseTableProjectionSpec.java
@@ -26,6 +26,7 @@ import org.apache.druid.query.OrderBy;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.segment.VirtualColumns;
import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.utils.CollectionUtils;
import javax.annotation.Nullable;
import java.util.ArrayList;
@@ -70,6 +71,17 @@ public final class AdaptedBaseTableProjectionSpec implements
BaseTableProjection
return granularitySpec;
}
+ @Override
+ public AdaptedBaseTableProjectionSpec withAdditionalColumns(@Nullable
List<DimensionSchema> additionalColumns)
+ {
+ if (CollectionUtils.isNullOrEmpty(additionalColumns)) {
+ return this;
+ }
+ final List<DimensionSchema> revised = new
ArrayList<>(dimensionsSpec.getDimensions());
+ revised.addAll(additionalColumns);
+ return new AdaptedBaseTableProjectionSpec(granularitySpec,
dimensionsSpec.withDimensions(revised), metrics);
+ }
+
@Override
public VirtualColumns getVirtualColumns()
{
diff --git
a/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java
b/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java
index c236a427210..292a26fe68c 100644
---
a/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java
+++
b/processing/src/main/java/org/apache/druid/data/input/impl/BaseTableProjectionSpec.java
@@ -85,6 +85,11 @@ public interface BaseTableProjectionSpec
*/
BaseTableProjectionSpec withQueryGranularity(@Nullable Granularity
queryGranularity);
+ /**
+ * Returns a copy of this spec with the given columns appended to those it
already declares.
+ */
+ BaseTableProjectionSpec withAdditionalColumns(@Nullable
List<DimensionSchema> additionalColumns);
+
/**
* Returns true if this spec is equivalent to {@code other} for the purpose
of deciding whether a segment is already
* compacted. Segment granularity, query granularity, and rollup are each
compared by their own compaction check
diff --git
a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java
b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java
index 401816421dc..4a2855ba9b3 100644
---
a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java
+++
b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java
@@ -229,6 +229,43 @@ public final class
ClusteredValueGroupsBaseTableProjectionSpec implements BaseTa
.equals(((ClusteredValueGroupsBaseTableProjectionSpec)
other).withoutQueryGranularity());
}
+ /**
+ * Appends the given columns after the declared ones, which keeps the
clustering columns the leading prefix of
+ * {@link #getColumns()}.
+ */
+ @Override
+ public ClusteredValueGroupsBaseTableProjectionSpec withAdditionalColumns(
+ @Nullable List<DimensionSchema> additionalColumns
+ )
+ {
+ if (CollectionUtils.isNullOrEmpty(additionalColumns)) {
+ return this;
+ }
+ final List<DimensionSchema> revised = new ArrayList<>(columns.size() +
additionalColumns.size());
+ revised.addAll(columns);
+ for (DimensionSchema additionalColumn : additionalColumns) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(additionalColumn.getName())) {
+ throw InvalidInput.exception(
+ "Cannot append column [%s] to a [%s] base table; it must be
declared at its position in the column list",
+ ColumnHolder.TIME_COLUMN_NAME,
+ TYPE_NAME
+ );
+ }
+ if (virtualColumns.getVirtualColumn(additionalColumn.getName()) != null)
{
+ throw InvalidInput.exception(
+ "Cannot append column [%s] to a [%s] base table; it is computed by
a virtual column, so"
+ + " the arriving values for this column would be ignored",
+ additionalColumn.getName(),
+ TYPE_NAME
+ );
+ }
+ revised.add(additionalColumn);
+ }
+ // Duplicates of a declared column, and of a column materialized by a
virtual column, are rejected by the
+ // constructor's validation.
+ return new ClusteredValueGroupsBaseTableProjectionSpec(virtualColumns,
revised, clusteringColumns);
+ }
+
/**
* Returns a copy of this spec with the {@link
Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} virtual column removed,
* the inverse of {@link #withQueryGranularity(Granularity)}. If no such
virtual column is present this returns
diff --git
a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java
b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java
index 603919e7112..a09fdfc42de 100644
---
a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java
+++
b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java
@@ -19,8 +19,10 @@
package org.apache.druid.data.input.impl;
+import com.google.common.collect.ImmutableList;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.OrderBy;
import org.apache.druid.query.dimension.DimensionSpec;
import org.apache.druid.query.expression.TestExprMacroTable;
import org.apache.druid.segment.ColumnSelectorFactory;
@@ -37,6 +39,7 @@ import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
class ClusteredValueGroupsBaseTableProjectionSpecTest extends
InitializedNullHandlingTest
{
@@ -294,6 +297,144 @@ class ClusteredValueGroupsBaseTableProjectionSpecTest
extends InitializedNullHan
Assertions.assertTrue(e.getMessage().contains("[dotty]"));
}
+ @Test
+ void testWithAdditionalColumnsAppendsAfterDeclaredColumns()
+ {
+ final ClusteredValueGroupsBaseTableProjectionSpec spec =
tenantSpec().withAdditionalColumns(
+ ImmutableList.of(new LongDimensionSchema("cnt"), new
StringDimensionSchema("city"))
+ );
+
+ Assertions.assertEquals(
+ ImmutableList.of("tenant", "region", "__time", "cnt", "city"),
+
spec.getColumns().stream().map(DimensionSchema::getName).collect(Collectors.toList())
+ );
+ // The clustering prefix is untouched: the appended columns are stored and
sorted by, but not clustered on.
+ Assertions.assertEquals(ImmutableList.of("tenant"),
spec.getClusteringColumnNames());
+ Assertions.assertEquals(
+ ImmutableList.of("tenant"),
+
spec.getClusteringColumns().stream().map(DimensionSchema::getName).collect(Collectors.toList())
+ );
+ Assertions.assertEquals(
+ ImmutableList.of("region", "__time", "cnt", "city"),
+
spec.getNonClusteringColumns().stream().map(DimensionSchema::getName).collect(Collectors.toList())
+ );
+ // Rows are physically sorted by every column present, so the appended
columns join the ordering at the end.
+ Assertions.assertEquals(
+ ImmutableList.of(
+ OrderBy.ascending("tenant"),
+ OrderBy.ascending("region"),
+ OrderBy.ascending("__time"),
+ OrderBy.ascending("cnt"),
+ OrderBy.ascending("city")
+ ),
+ spec.getOrdering()
+ );
+ Assertions.assertEquals(spec.getColumns(),
spec.getDimensionsSpec().getDimensions());
+ }
+
+ @Test
+ void testWithAdditionalColumnsNullAndEmptyAreNoOps()
+ {
+ final ClusteredValueGroupsBaseTableProjectionSpec spec = tenantSpec();
+ Assertions.assertSame(spec, spec.withAdditionalColumns(null));
+ Assertions.assertSame(spec,
spec.withAdditionalColumns(Collections.emptyList()));
+ }
+
+ @Test
+ void testWithAdditionalColumnsKeepsVirtualColumnsAndQueryGranularity()
+ {
+ final ClusteredValueGroupsBaseTableProjectionSpec spec =
ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .virtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("region_upper", "upper(region)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new StringDimensionSchema("region"),
+ new StringDimensionSchema("region_upper"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant")
+ .build()
+ .withQueryGranularity(Granularities.HOUR)
+ .withAdditionalColumns(ImmutableList.of(new
LongDimensionSchema("cnt")));
+
+
Assertions.assertNotNull(spec.getVirtualColumns().getVirtualColumn("region_upper"));
+ Assertions.assertEquals(Granularities.HOUR, spec.getQueryGranularity());
+ Assertions.assertEquals("cnt",
spec.getColumns().get(spec.getColumns().size() - 1).getName());
+ }
+
+ @Test
+ void testWithAdditionalColumnsRejectsTimeColumn()
+ {
+ // __time marks a position in the column list, so it can never arrive as
an appended extra.
+ final DruidException e = Assertions.assertThrows(
+ DruidException.class,
+ () -> tenantSpec().withAdditionalColumns(ImmutableList.of(new
LongDimensionSchema("__time")))
+ );
+ Assertions.assertTrue(e.getMessage().contains("[__time]"));
+ }
+
+ @Test
+ void testWithAdditionalColumnsRejectsDuplicateOfDeclaredColumn()
+ {
+ final DruidException e = Assertions.assertThrows(
+ DruidException.class,
+ () -> tenantSpec().withAdditionalColumns(ImmutableList.of(new
StringDimensionSchema("region")))
+ );
+ Assertions.assertTrue(e.getMessage().contains("duplicate name [region]"));
+ }
+
+ @Test
+ void testWithAdditionalColumnsRejectsDuplicateOfVirtualColumnOutput()
+ {
+ // region_upper is materialized by a virtual column, so an incoming column
of the same name would be computed by
+ // the virtual column rather than read; the arriving values would be
ignored.
+ final DruidException e = Assertions.assertThrows(
+ DruidException.class,
+ () -> ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .virtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("region_upper", "upper(region)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new StringDimensionSchema("region"),
+ new StringDimensionSchema("region_upper"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant")
+ .build()
+ .withAdditionalColumns(ImmutableList.of(new
StringDimensionSchema("region_upper")))
+ );
+ Assertions.assertTrue(e.getMessage().contains("[region_upper]"));
+ Assertions.assertTrue(e.getMessage().contains("computed by a virtual
column"));
+ }
+
+ @Test
+ void testWithAdditionalColumnsRejectsUnmaterializedVirtualColumnName()
+ {
+ // Chain: tenant_key := upper(v0), v0 := lower(tenant); the intermediate
v0 is not a stored column. Appending a
+ // column named v0 would pass the constructor's rules (its output would
simply become stored), but ingest reads
+ // virtual columns first, so the arriving v0 values would be silently
discarded.
+ final ClusteredValueGroupsBaseTableProjectionSpec spec =
ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .virtualColumns(VirtualColumns.create(
+ new ExpressionVirtualColumn("v0", "lower(tenant)",
ColumnType.STRING, TestExprMacroTable.INSTANCE),
+ new ExpressionVirtualColumn("tenant_key", "upper(v0)",
ColumnType.STRING, TestExprMacroTable.INSTANCE)
+ ))
+ .columns(
+ new StringDimensionSchema("tenant_key"),
+ new StringDimensionSchema("tenant"),
+ new LongDimensionSchema("__time")
+ )
+ .clusteringColumns("tenant_key")
+ .build();
+ final DruidException e = Assertions.assertThrows(
+ DruidException.class,
+ () -> spec.withAdditionalColumns(ImmutableList.of(new
StringDimensionSchema("v0")))
+ );
+ Assertions.assertTrue(e.getMessage().contains("[v0]"));
+ Assertions.assertTrue(e.getMessage().contains("computed by a virtual
column"));
+ }
+
/**
* Minimal test-only virtual column whose only meaningful behavior is {@link
#usesDotNotation()} returning true; the
* selector/capability methods are never reached by spec validation. (No
core virtual column uses dot notation.)
diff --git
a/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
b/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
index 89982a94883..5b2e163eb94 100644
---
a/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
+++
b/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
@@ -30,7 +30,6 @@ import
org.apache.druid.catalog.model.ModelProperties.StringListPropertyDefn;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.TableDefn;
import org.apache.druid.catalog.model.TableSpec;
-import org.apache.druid.error.InvalidInput;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.StringUtils;
@@ -107,18 +106,6 @@ public class DatasourceDefn extends TableDefn
super.validate(table);
final DatasourceBaseTableMetadata baseTable =
table.decodeProperty(BASE_TABLE_PROPERTY);
if (baseTable != null) {
- // A base table layout derives the physical segment schema from the
declared columns, so a column the query
- // produces but the table does not declare cannot be stored; require
'sealed' so ingestion rejects such columns
- // instead of silently dropping them. Requiring the flag allows us to
someday support non-sealed definitions,
- // which could work by appending undeclared columns to the derived
schema.
- if (!table.booleanProperty(SEALED_PROPERTY)) {
- throw InvalidInput.exception(
- "Datasource with a [%s] layout must also set [%s] to true; the
declared columns define the physical"
- + " segment schema, so columns not declared in the table cannot be
ingested",
- BASE_TABLE_PROPERTY,
- SEALED_PROPERTY
- );
- }
// Cross-validate the layout against the declared columns by deriving
the physical spec, so that catalog writes
// fail fast instead of surfacing layout problems at ingest time.
baseTable.createSpec(table.spec().columns());
diff --git
a/server/src/test/java/org/apache/druid/catalog/model/table/DatasourceTableTest.java
b/server/src/test/java/org/apache/druid/catalog/model/table/DatasourceTableTest.java
index dc40a131578..657f74b21ee 100644
---
a/server/src/test/java/org/apache/druid/catalog/model/table/DatasourceTableTest.java
+++
b/server/src/test/java/org/apache/druid/catalog/model/table/DatasourceTableTest.java
@@ -163,8 +163,8 @@ public class DatasourceTableTest extends
InitializedNullHandlingTest
}
{
- // A base table layout requires 'sealed': the declared columns define
the physical segment schema, so
- // undeclared columns cannot be ingested.
+ // A base table layout does not require 'sealed': columns the table does
not declare are appended after the
+ // declared ones at ingest time rather than dropped.
TableSpec spec = new TableSpec(
DatasourceDefn.TABLE_TYPE,
ImmutableMap.of(
@@ -173,9 +173,7 @@ public class DatasourceTableTest extends
InitializedNullHandlingTest
),
columns
);
- ResolvedTable table = registry.resolve(spec);
- DruidException e = Assertions.assertThrows(DruidException.class,
table::validate);
- Assertions.assertTrue(e.getMessage().contains("must also set [sealed] to
true"));
+ expectValidationSucceeds(spec);
}
{
diff --git
a/server/src/test/java/org/apache/druid/segment/indexing/DataSchemaTest.java
b/server/src/test/java/org/apache/druid/segment/indexing/DataSchemaTest.java
index d030618ea59..0bf965fe5b0 100644
--- a/server/src/test/java/org/apache/druid/segment/indexing/DataSchemaTest.java
+++ b/server/src/test/java/org/apache/druid/segment/indexing/DataSchemaTest.java
@@ -63,6 +63,7 @@ import org.mockito.Mockito;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -841,6 +842,34 @@ class DataSchemaTest extends InitializedNullHandlingTest
Assertions.assertArrayEquals(schema.getAggregators(),
effective.getMetrics());
}
+ @Test
+ void testLegacyModeEffectiveBaseTableSpecAppendsAdditionalColumns()
+ {
+ final BaseTableProjectionSpec effective = DataSchema.builder()
+
.withDataSource("datasource")
+
.withTimestamp(TIMESTAMP_SPEC)
+ .withDimensions(new
StringDimensionSchema("tenant"))
+ .withAggregators(new
CountAggregatorFactory("rows"))
+
.withGranularity(ARBITRARY_GRANULARITY)
+ .build()
+
.getEffectiveBaseTableSpec();
+
+ Assertions.assertSame(effective, effective.withAdditionalColumns(null));
+ Assertions.assertSame(effective,
effective.withAdditionalColumns(Collections.emptyList()));
+
+ final BaseTableProjectionSpec appended =
+ effective.withAdditionalColumns(ImmutableList.of(new
StringDimensionSchema("region")));
+ Assertions.assertEquals(
+ ImmutableList.of(new StringDimensionSchema("tenant"), new
StringDimensionSchema("region")),
+ appended.getDimensionsSpec().getDimensions()
+ );
+ Assertions.assertArrayEquals(effective.getMetrics(),
appended.getMetrics());
+ Assertions.assertEquals(
+ ARBITRARY_GRANULARITY,
+ ((AdaptedBaseTableProjectionSpec) appended).getGranularitySpec()
+ );
+ }
+
@Test
void testLegacyModeJsonRoundTripOmitsBaseTable() throws IOException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
