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 f0b9ca01c25d..0c2efdc6fab8 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.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 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 static DataSchema makeDataSchemaForIngestion( 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,59 @@ private static boolean isRollupQuery(Query query) && !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. + *

+ * 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 undeclaredColumns( + final BaseTableProjectionSpec baseTable, + final RowSignature querySignature, + final ColumnMappings columnMappings, + final Query query, + @Nullable final Map dimensionSchemas + ) + { + final Set 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 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) && !DimensionHandlerUtils.DIMENSION_HANDLER_PROVIDERS.containsKey(type.getComplexTypeName())) { + // 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 f425fdfbd0ef..42627247f833 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 @@ -281,6 +281,17 @@ protected CatalogResolver createMockCatalogResolver() ) .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); } @@ -879,6 +890,128 @@ public void testInsertOnExternalDataSourceWithCatalogClusteredBaseTable( .verifyResults(); } + @MethodSource("data") + @ParameterizedTest(name = "{index}:with context {0}") + public void testInsertOnExternalDataSourceWithUnsealedCatalogClusteredBaseTable( + String contextName, + Map 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 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( + CoreMatchers.allOf( + CoreMatchers.instanceOf(ISE.class), + ThrowableMessageMatcher.hasMessage(CoreMatchers.containsString( + "Column [unique_users] has type [COMPLEX], 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 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 ac59162a514c..2828d2cda533 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.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 GranularitySpec getGranularitySpec() return granularitySpec; } + @Override + public AdaptedBaseTableProjectionSpec withAdditionalColumns(@Nullable List additionalColumns) + { + if (CollectionUtils.isNullOrEmpty(additionalColumns)) { + return this; + } + final List 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 c236a427210a..292a26fe68cc 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 @@ default Granularity getQueryGranularity() */ 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 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 401816421dc5..f98fd40766f0 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,35 @@ public boolean hasEqualCompactionState(BaseTableProjectionSpec other) .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 additionalColumns + ) + { + if (CollectionUtils.isNullOrEmpty(additionalColumns)) { + return this; + } + final List 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 + ); + } + 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 603919e71128..4d0ee6c93269 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 java.util.Collections; import java.util.List; +import java.util.stream.Collectors; class ClusteredValueGroupsBaseTableProjectionSpecTest extends InitializedNullHandlingTest { @@ -294,6 +297,117 @@ void testDotNotationVirtualColumnIsRejected() 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 is not an extra; it is a + // collision with a column the spec already declares. + 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("duplicate name [region_upper]")); + } + /** * 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 89982a94883f..5b2e163eb94f 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.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 void validate(ResolvedTable table) 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 ba32f61de7fc..de85be1e110b 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 @@ -171,8 +171,8 @@ public void testSpecWithBaseTableProp() } { - // 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( @@ -181,9 +181,7 @@ public void testSpecWithBaseTableProp() ), columns ); - ResolvedTable table = registry.resolve(spec); - DruidException e = assertThrows(DruidException.class, table::validate); - 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 ddc8168e2f97..5b1bcc4748f2 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 @@ -64,6 +64,7 @@ import java.io.IOException; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Set; @@ -842,6 +843,34 @@ void testLegacyModeEffectiveBaseTableSpecSynthesizedFromLegacyFields() 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 {