This is an automated email from the ASF dual-hosted git repository.
gianm 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 1e510ab555e feat: catalog + clustered segments (#19711)
1e510ab555e is described below
commit 1e510ab555e1292530625ee02995bbd693d85ee5
Author: Clint Wylie <[email protected]>
AuthorDate: Fri Jul 24 15:42:20 2026 -0700
feat: catalog + clustered segments (#19711)
---
.../catalog/CatalogIngestAndQueryTest.java | 137 ++++++++-
.../testing/embedded/catalog/CatalogTestBase.java | 5 +-
.../druid/catalog/sql/CatalogInsertTest.java | 4 +
.../druid/catalog/sql/CatalogReplaceTest.java | 4 +
.../apache/druid/catalog/storage/CatalogTests.java | 6 +-
.../sampler/InputSourceSamplerDiscoveryTest.java | 2 +-
.../apache/druid/msq/sql/MSQTaskQueryMaker.java | 42 ++-
.../org/apache/druid/msq/exec/MSQInsertTest.java | 163 ++++++++++
.../msq/indexing/MSQCompactionRunnerTest.java | 2 +-
.../org/apache/druid/msq/test/MSQTestBase.java | 21 +-
.../druid/msq/test/MSQTestWorkerContext.java | 7 +-
.../druid/data/input/impl/DimensionSchema.java | 7 +-
.../druid/segment/DefaultColumnFormatConfig.java | 2 +-
.../druid/segment/NestedDataColumnSchema.java | 11 +-
.../segment/nested/NestedDataComplexTypeSerde.java | 2 +-
.../druid/data/input/impl/DimensionSchemaTest.java | 47 +++
.../druid/segment/NestedDataColumnSchemaTest.java | 2 +-
.../ClusteredValueGroupsBaseTableMetadata.java | 178 +++++++++++
.../catalog/model/DatasourceBaseTableMetadata.java | 65 ++++
.../catalog/model/facade/DatasourceFacade.java | 80 ++++-
.../druid/catalog/model/table/DatasourceDefn.java | 47 ++-
.../druid/catalog/model/table/TableBuilder.java | 6 +
.../ClusteredValueGroupsBaseTableMetadataTest.java | 340 +++++++++++++++++++++
.../catalog/model/table/DatasourceTableTest.java | 143 +++++++++
.../sql/calcite/planner/DruidSqlValidator.java | 13 +-
.../calcite/CalciteCatalogIngestionDmlTest.java | 187 +++++++++++-
26 files changed, 1478 insertions(+), 45 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogIngestAndQueryTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogIngestAndQueryTest.java
index c9407a46472..1dcca2c4a8f 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogIngestAndQueryTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogIngestAndQueryTest.java
@@ -20,15 +20,23 @@
package org.apache.druid.testing.embedded.catalog;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata;
import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
import org.apache.druid.catalog.model.TableMetadata;
import org.apache.druid.catalog.model.table.ClusterKeySpec;
import org.apache.druid.catalog.model.table.DatasourceDefn;
import org.apache.druid.catalog.model.table.TableBuilder;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
import org.apache.druid.query.http.ClientSqlQuery;
import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.server.metrics.LatchableEmitter;
import org.apache.druid.testing.embedded.msq.EmbeddedMSQApis;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -74,7 +82,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* 2022-12-26T12:34:56,extra,8,"40",2.0,foq
* 2022-12-26T12:34:56,extra,8,"50",2.0,fop
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', the
BIGINT type column should
+ * When querying the table with query: {@code SELECT * FROM table}, the
BIGINT type column should
* be implicitly coherced into type DOUBLE when inserted into the table,
since the column being
* written into is type DOUBLE.
* <p>
@@ -141,7 +149,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* 2022-12-26T12:34:56,extra,8,"40",2.0,foq
* 2022-12-26T12:34:56,extra,8,"50",2.0,fop
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', because
of the clustering
+ * When querying the table with query: {@code SELECT * FROM table}, because
of the clustering
* defined on the table, the data should be reordered to:
* <p>
* __time, bigint_col1
@@ -211,7 +219,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* 2022-12-26T12:34:56,extra,8,"40",2.0,foq
* 2022-12-26T12:34:56,extra,8,"50",2.0,fop
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', because
of the clustering
+ * When querying the table with query: {@code SELECT * FROM table}, because
of the clustering
* defined on the table, the data should be reordered to:
* <p>
* __time, bigint_col1
@@ -281,7 +289,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* 2022-12-26T12:34:56,extra,8,"40",2.0,foq
* 2022-12-26T12:34:56,extra,8,"50",2.0,fop
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', because
of the clustering
+ * When querying the table with query: {@code SELECT * FROM table}, because
of the clustering
* defined on the table, the data should be reordered to:
* <p>
* __time, varchar_col1, bigint_col1, float_col1, varchar_col2
@@ -360,7 +368,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* 2022-12-26T12:34:56,extra,8,"40",2.0,foq
* 2022-12-26T12:34:56,extra,8,"50",2.0,fop
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', because
of the clustering
+ * When querying the table with query: {@code SELECT * FROM table}, because
of the clustering
* defined on the query, the data should be reordered to:
* <p>
* __time, varchar_col1, bigint_col1, float_col1, varchar_col2
@@ -415,6 +423,123 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
);
}
+ /**
+ * Create a table with columns (in declared order, which is the physical
segment order; the clustering column
+ * must be declared first):
+ * <p>
+ * varchar_col2 VARCHAR
+ * __time LONG
+ * varchar_col1 VARCHAR
+ * bigint_col1 BIGINT
+ * <p>
+ * and a clustered base table layout with clustering column {@code
varchar_col2}, plus an aggregate projection
+ * ({@code varchar_col2_daily}: group by varchar_col2, sum bigint_col1), so
ingestion produces clustered (V10)
+ * segments that also carry a segment-wide projection.
+ * <p>
+ * Insert the following data:
+ * <p>
+ * __time, varchar_col1, bigint_col1, float_col1, varchar_col2
+ * 2022-12-26T12:34:56,extra,10,"20",2.0,foo
+ * 2022-12-26T12:34:56,extra,9,"30",2.0,foo
+ * 2022-12-26T12:34:56,extra,8,"40",2.0,foq
+ * 2022-12-26T12:34:56,extra,8,"50",2.0,fop
+ * <p>
+ * {@code SELECT * FROM table} returns columns in catalog order and rows in
segment order: cluster groups in
+ * clustering-value order (foo, fop, foq), each group internally ordered by
the declared column order (within
+ * 'foo', by bigint_col1).
+ */
+ @Test
+ public void testInsertClusteredBaseTableFromCatalog()
+ {
+ String tableName = dataSource;
+ TableMetadata table = TableBuilder.datasource(tableName, "P1D")
+ .column("varchar_col2", "VARCHAR")
+ .column(Columns.TIME_COLUMN, Columns.LONG)
+ .column("varchar_col1", "VARCHAR")
+ .column("bigint_col1", "BIGINT")
+ .sealed(true)
+ .property(
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("varchar_col2"), null)
+ )
+ .property(
+ DatasourceDefn.PROJECTIONS_KEYS_PROPERTY,
+ ImmutableList.of(
+ new DatasourceProjectionMetadata(
+ AggregateProjectionSpec.builder("varchar_col2_daily")
+ .virtualColumns(
+ Granularities.toVirtualColumn(
+ Granularities.DAY,
+
Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+ )
+ )
+ .groupingColumns(
+ new
LongDimensionSchema(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME),
+ new
StringDimensionSchema("varchar_col2")
+ )
+ .aggregators(new
LongSumAggregatorFactory("sum_bigint_col1", "bigint_col1"))
+ .build()
+ )
+ )
+ )
+ .build();
+
+ client.createTable(table, true);
+ String queryInline =
+ StringUtils.format(dmlPrefixPattern, tableName) + "\n"
+ + "SELECT\n"
+ + " TIME_PARSE(a) AS __time,\n"
+ + " b AS varchar_col1,\n"
+ + " c AS bigint_col1,\n"
+ + " f AS varchar_col2\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + "
'{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,extra,10,\\\"20\\\",2.0,foo\\n2022-12-26T12:34:56,extra,9,\\\"30\\\",2.0,foo\\n2022-12-26T12:34:56,extra,8,\\\"40\\\",2.0,foq\\n2022-12-26T12:34:56,extra,8,\\\"50\\\",2.0,fop\"}',\n"
+ + "
'{\"type\":\"csv\",\"findColumnsFromHeader\":false,\"columns\":[\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"]}'\n"
+ + " )\n"
+ + ") "
+ + " EXTEND (a VARCHAR, b VARCHAR, c BIGINT, d VARCHAR, e FLOAT, f
VARCHAR)\n";
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqApis.submitTaskSql(queryInline);
+ cluster.callApi().waitForTaskToSucceed(sqlTaskStatus.getTaskId(),
overlord);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+
+ cluster.callApi().verifySqlQuery(
+ "SELECT * FROM %s",
+ dataSource,
+ "foo,2022-12-26T12:34:56.000Z,extra,9\n"
+ + "foo,2022-12-26T12:34:56.000Z,extra,10\n"
+ + "fop,2022-12-26T12:34:56.000Z,extra,8\n"
+ + "foq,2022-12-26T12:34:56.000Z,extra,8"
+ );
+
+ final LatchableEmitter emitter = historical.latchableEmitter();
+ emitter.flush();
+
+ // This aggregation matches the varchar_col2_daily projection (group by
varchar_col2, sum bigint_col1), proving
+ // projections coexist with (and are chosen over) the clustered base table.
+ cluster.callApi().verifySqlQuery(
+ "SELECT varchar_col2, SUM(bigint_col1) FROM %s GROUP BY 1 ORDER BY 1",
+ dataSource,
+ "foo,19\nfop,8\nfoq,8"
+ );
+
+ // When the projection is used, the segment-scan query metrics carry the
projection name as a dimension.
+ emitter.waitForEvent(
+ event -> event.hasMetricName("query/segment/time")
+ .hasDimension("projection", "varchar_col2_daily")
+ );
+
+ // Grouping on a column the projection does not carry cannot be served by
it: this cross-group aggregation
+ // exercises the clustered read path (per-group local dictionaries).
+ cluster.callApi().verifySqlQuery(
+ "SELECT varchar_col1, SUM(bigint_col1) FROM %s GROUP BY 1",
+ dataSource,
+ "extra,35"
+ );
+ }
+
/**
* Adding a new column during ingestion that is not defined in a sealed
table, should fail with
* proper validation error. Disabling catalog validation, through context
parameter, and issuing ingest
@@ -487,7 +612,7 @@ public abstract class CatalogIngestAndQueryTest extends
CatalogTestBase
* <p>
* 2022-12-26T12:34:56,extra
* <p>
- * When querying the table with query: 'SELECT * from ##tableName', the data
is returned as:
+ * When querying the table with query: {@code SELECT * FROM table}, the data
is returned as:
* <p>
* __time, double_col
* 2022-12-26T12:34:56,0.0
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogTestBase.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogTestBase.java
index 325859a0c7a..de034eb46e1 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogTestBase.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogTestBase.java
@@ -48,6 +48,7 @@ public abstract class CatalogTestBase extends
EmbeddedClusterTestBase
private final EmbeddedIndexer indexer = new EmbeddedIndexer()
.setServerMemory(1_000_000_000)
.addProperty("druid.worker.capacity", "2");
+ protected final EmbeddedHistorical historical = new EmbeddedHistorical();
@Override
protected EmbeddedDruidCluster createCluster()
@@ -58,11 +59,13 @@ public abstract class CatalogTestBase extends
EmbeddedClusterTestBase
CatalogClientModule.class,
CatalogCoordinatorModule.class
)
+ // clustered base table segments require the
V10 format
+
.addCommonProperty("druid.indexer.task.buildV10", "true")
.addServer(overlord)
.addServer(coordinator)
.addServer(broker)
.addServer(indexer)
- .addServer(new EmbeddedHistorical());
+ .addServer(historical);
}
void verifySubmitSqlTaskFailsWith400BadRequest(String sql, String
expectedMessageSubstring)
diff --git
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogInsertTest.java
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogInsertTest.java
index 82076c89604..c07af9db584 100644
---
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogInsertTest.java
+++
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogInsertTest.java
@@ -100,6 +100,10 @@ public class CatalogInsertTest extends
CalciteCatalogInsertTest
tableBuilder.clusterColumns(catalogMetadata.clusterKeys().toArray(new
ClusterKeySpec[0]));
}
+ if (catalogMetadata.baseTableMetadata() != null) {
+ tableBuilder.baseTable(catalogMetadata.baseTableMetadata());
+ }
+
createTableMetadata(tableBuilder.build());
});
}
diff --git
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogReplaceTest.java
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogReplaceTest.java
index caff8a01f4a..55ab1a06d6b 100644
---
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogReplaceTest.java
+++
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/CatalogReplaceTest.java
@@ -99,6 +99,10 @@ public class CatalogReplaceTest extends
CalciteCatalogReplaceTest
tableBuilder.clusterColumns(catalogMetadata.clusterKeys().toArray(new
ClusterKeySpec[0]));
}
+ if (catalogMetadata.baseTableMetadata() != null) {
+ tableBuilder.baseTable(catalogMetadata.baseTableMetadata());
+ }
+
createTableMetadata(tableBuilder.build());
});
DatasourceFacade catalogMetadata =
diff --git
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/storage/CatalogTests.java
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/storage/CatalogTests.java
index caab4a3dad8..0c30c073686 100644
---
a/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/storage/CatalogTests.java
+++
b/extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/storage/CatalogTests.java
@@ -19,11 +19,13 @@
package org.apache.druid.catalog.storage;
+import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import org.apache.druid.catalog.storage.sql.CatalogManager;
import org.apache.druid.catalog.storage.sql.SQLCatalogManager;
import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.metadata.TestDerbyConnector.DerbyConnectorRule;
import org.apache.druid.server.security.Access;
import org.apache.druid.server.security.Action;
@@ -70,7 +72,9 @@ public class CatalogTests
}
}
- public static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper();
+ public static final ObjectMapper JSON_MAPPER = new
DefaultObjectMapper().setInjectableValues(
+ new InjectableValues.Std().addValue(ExprMacroTable.class,
ExprMacroTable.nil())
+ );
public static class DbFixture
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/sampler/InputSourceSamplerDiscoveryTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/sampler/InputSourceSamplerDiscoveryTest.java
index ffc20ee2052..70f58e795cc 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/sampler/InputSourceSamplerDiscoveryTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/sampler/InputSourceSamplerDiscoveryTest.java
@@ -80,7 +80,7 @@ public class InputSourceSamplerDiscoveryTest extends
InitializedNullHandlingTest
new DoubleDimensionSchema("double"),
new LongDimensionSchema("bool"),
new StringDimensionSchema("variant"),
- AutoTypeColumnSchema.of("array"),
+ new AutoTypeColumnSchema("array", ColumnType.LONG_ARRAY, null),
AutoTypeColumnSchema.of("nested")
),
response.getLogicalDimensions()
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java
index 517731a2089..6e01450ee0d 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java
@@ -26,12 +26,14 @@ import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.runtime.Hook;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.druid.catalog.MetadataCatalog;
+import org.apache.druid.catalog.model.DatasourceBaseTableMetadata;
import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.TableId;
import org.apache.druid.catalog.model.table.DatasourceDefn;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.BaseTableProjectionSpec;
import org.apache.druid.error.DruidException;
import org.apache.druid.error.InvalidInput;
import org.apache.druid.java.util.common.Intervals;
@@ -510,7 +512,9 @@ public class MSQTaskQueryMaker implements QueryMaker
);
- final List<AggregateProjectionSpec> projectionSpecs =
getProjections(targetDataSource, plannerContext);
+ final ResolvedTable tableMetadata = getTableMetadata(targetDataSource,
plannerContext);
+ final BaseTableProjectionSpec baseTable = getBaseTable(tableMetadata);
+ final List<AggregateProjectionSpec> projectionSpecs =
getProjections(tableMetadata);
final DataSourceMSQDestination dataSourceDestination = new
DataSourceMSQDestination(
targetDataSource.getDestinationName(),
@@ -518,6 +522,7 @@ public class MSQTaskQueryMaker implements QueryMaker
segmentSortOrder,
replaceTimeChunks,
null,
+ baseTable,
projectionSpecs,
terminalStageSpecFactory.createTerminalStageSpec(
plannerContext
@@ -550,16 +555,19 @@ public class MSQTaskQueryMaker implements QueryMaker
return tuningConfig;
}
- private static List<AggregateProjectionSpec> getProjections(
- IngestDestination targetDataSource,
- PlannerContext plannerContext
- )
+ @Nullable
+ private static ResolvedTable getTableMetadata(IngestDestination
targetDataSource, PlannerContext plannerContext)
{
- final List<AggregateProjectionSpec> projectionSpecs;
final MetadataCatalog metadataCatalog =
plannerContext.getPlannerToolbox().catalogResolver().getMetadataCatalog();
- final ResolvedTable tableMetadata = metadataCatalog.resolveTable(
+ return metadataCatalog.resolveTable(
TableId.datasource(targetDataSource.getDestinationName())
);
+ }
+
+ @Nullable
+ private static List<AggregateProjectionSpec> getProjections(@Nullable
ResolvedTable tableMetadata)
+ {
+ final List<AggregateProjectionSpec> projectionSpecs;
if (tableMetadata != null) {
final List<DatasourceProjectionMetadata> projectionMetadata =
tableMetadata.decodeProperty(
DatasourceDefn.PROJECTIONS_KEYS_PROPERTY
@@ -576,4 +584,24 @@ public class MSQTaskQueryMaker implements QueryMaker
}
return projectionSpecs;
}
+
+ /**
+ * Derives the physical base table spec for the target datasource from the
catalog, if the table declares a
+ * {@link DatasourceDefn#BASE_TABLE_PROPERTY} layout; combines the layout
with the declared catalog columns via
+ * {@link DatasourceBaseTableMetadata#createSpec}.
+ */
+ @Nullable
+ private static BaseTableProjectionSpec getBaseTable(@Nullable ResolvedTable
tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ final DatasourceBaseTableMetadata baseTableMetadata =
tableMetadata.decodeProperty(
+ DatasourceDefn.BASE_TABLE_PROPERTY
+ );
+ if (baseTableMetadata == null) {
+ return null;
+ }
+ return baseTableMetadata.createSpec(tableMetadata.spec().columns());
+ }
}
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 5c83a850a3b..8e00d2780b9 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
@@ -26,6 +26,8 @@ import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.druid.catalog.MapMetadataCatalog;
+import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata;
+import org.apache.druid.catalog.model.Columns;
import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
import org.apache.druid.catalog.model.TableId;
import org.apache.druid.catalog.model.table.DatasourceDefn;
@@ -76,6 +78,7 @@ import org.apache.druid.sql.calcite.planner.CatalogResolver;
import org.apache.druid.sql.calcite.planner.ColumnMapping;
import org.apache.druid.sql.calcite.planner.ColumnMappings;
import org.apache.druid.sql.calcite.util.CalciteTests;
+import org.apache.druid.timeline.ClusterGroupTuples;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NumberedShardSpec;
import org.hamcrest.CoreMatchers;
@@ -232,6 +235,45 @@ public class MSQInsertTest extends MSQTestBase
)
.buildSpec()
);
+ metadataCatalog.addSpec(
+ TableId.datasource("fooClustered"),
+ TableBuilder.datasource("fooClustered", Granularities.DAY.toString())
+ // declared order is the physical segment order: the
clustering column leads
+ .column("channel", Columns.SQL_VARCHAR)
+ .timeColumn()
+ .column("page", Columns.SQL_VARCHAR)
+ .column("user", Columns.SQL_VARCHAR)
+ .column("added", Columns.SQL_BIGINT)
+ .column("deleted", Columns.SQL_BIGINT)
+ .column("delta", Columns.SQL_BIGINT)
+ .sealed(true)
+ .baseTable(
+ new
ClusteredValueGroupsBaseTableMetadata(ImmutableList.of("channel"), null)
+ )
+ .property(
+ DatasourceDefn.PROJECTIONS_KEYS_PROPERTY,
+ ImmutableList.of(
+ new DatasourceProjectionMetadata(
+
AggregateProjectionSpec.builder("channel_added_hourly")
+ .virtualColumns(
+
Granularities.toVirtualColumn(
+
Granularities.HOUR,
+
Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+ )
+ )
+ .groupingColumns(
+ new
LongDimensionSchema(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME),
+ new
StringDimensionSchema("channel")
+ )
+ .aggregators(
+ new
LongSumAggregatorFactory("sum_added", "added")
+ )
+ .build()
+ )
+ )
+ )
+ .buildSpec()
+ );
return new LiveCatalogResolver(metadataCatalog);
}
@@ -709,6 +751,127 @@ public class MSQInsertTest extends MSQTestBase
}
+ @MethodSource("data")
+ @ParameterizedTest(name = "{index}:with context {0}")
+ public void testInsertOnExternalDataSourceWithCatalogClusteredBaseTable(
+ String contextName,
+ Map<String, Object> context
+ ) throws IOException
+ {
+ final File toRead = getResourceAsTemporaryFile("/wikipedia-sampled.json");
+ final String toReadFileNameAsJson =
queryFramework().queryJsonMapper().writeValueAsString(toRead.getAbsolutePath());
+
+ // The physical spec uses the declared catalog columns verbatim: the
declared order is the segment order, with
+ // the clustering prefix leading and __time an explicit positional column.
+ RowSignature rowSignature = RowSignature.builder()
+ .add("channel", ColumnType.STRING)
+ .add("__time", ColumnType.LONG)
+ .add("page", ColumnType.STRING)
+ .add("user", ColumnType.STRING)
+ .add("added", ColumnType.LONG)
+ .add("deleted", ColumnType.LONG)
+ .add("delta", ColumnType.LONG)
+ .build();
+
+ // The catalog also declares a projection: aggregate projections are
segment-wide views that coexist with the
+ // per-group clustered layout.
+ AggregateProjectionMetadata expectedProjection = new
AggregateProjectionMetadata(
+ AggregateProjectionSchema.schemaBuilder("channel_added_hourly")
+
.timeColumnName(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME)
+ .virtualColumns(
+ Granularities.toVirtualColumn(
+ Granularities.HOUR,
+
Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+ )
+ )
+
.groupAndOrder(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME, "channel")
+ .aggregators(new
LongSumAggregatorFactory("sum_added", "added"))
+ .build(),
+ 16
+ );
+
+ testIngestQuery().setSql(" insert into fooClustered 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("fooClustered")
+ .setExpectedRowSignature(rowSignature)
+ .setQueryContext(context)
+ .setExpectedSegments(ImmutableSet.of(SegmentId.of(
+ "fooClustered",
+ Intervals.of("2016-06-27/P1D"),
+ "test",
+ 0
+ )))
+ .setExpectedProjections(List.of(expectedProjection))
+ .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: cluster groups in
clustering-value order, each group
+ // internally ordered by the derived column order
(channel, __time, page, ...).
+ .setExpectedResultRows(
+ ImmutableList.of(
+ new Object[]{"#ceb.wikipedia", 1466985660000L,
"Neqerssuaq", "Lsjbot", 4150L, 0L, 4150L},
+ new Object[]{"#de.wikipedia", 1466992920000L,
"Benutzer Diskussion:Squasher/Archiv/2016", "TaxonBot", 2560L, 0L, 2560L},
+ new Object[]{"#de.wikipedia", 1466992980000L,
"Benutzer Diskussion:HerrSonderbar", "GiftBot", 364L, 0L, 364L},
+ new Object[]{"#en.wikipedia", 1466985600000L,
"Bailando 2015", "181.230.118.178", 2L, 0L, 2L},
+ new Object[]{"#en.wikipedia", 1466985600000L,
"Richie Rich's Christmas Wish", "JasonAQuest", 0L, 2L, -2L},
+ new Object[]{"#en.wikipedia", 1466985660000L,
"Panama Canal", "Mariordo", 496L, 0L, 496L},
+ new Object[]{"#en.wikipedia", 1466992980000L,
"File:Paint.net 4.0.6 screenshot.png", "Calvin Hogg", 0L, 463L, -463L},
+ new Object[]{"#es.wikipedia", 1466985660000L,
"Sumo (banda)", "181.110.165.189", 0L, 173L, -173L},
+ new Object[]{"#es.wikipedia", 1466989320000L,
"Clasificación para la Eurocopa Sub-21 de 2017", "Guly600", 4L, 0L, 4L},
+ new Object[]{"#id.wikipedia", 1466989320000L,
"Ibnu Sina", "Ftihikam", 106L, 0L, 106L},
+ new Object[]{"#pl.wikipedia", 1466985600000L,
"Kategoria:Dyskusje nad usunięciem artykułu zakończone bez konsensusu − lipiec
2016", "Beau.bot", 270L, 0L, 270L},
+ new Object[]{"#pt.wikipedia", 1466992920000L,
"Dobromir Zhechev", "Ceresta", 1926L, 0L, 1926L},
+ new Object[]{"#ru.wikipedia", 1466985720000L,
"Википедия:Опросы/Унификация шаблонов «Не переведено»", "Wanderer777", 196L,
0L, 196L},
+ new Object[]{"#sh.wikipedia", 1466985660000L, "El
Terco, Bachíniva", "Kolega2357", 0L, 1L, -1L},
+ new Object[]{"#sh.wikipedia", 1466985720000L,
"Hermanos Díaz, Ascensión", "Kolega2357", 0L, 1L, -1L},
+ new Object[]{"#sh.wikipedia", 1466989320000L, "El
Sicomoro, Ascensión", "Kolega2357", 0L, 1L, -1L},
+ new Object[]{"#sh.wikipedia", 1466992920000L,
"Trinidad Jiménez G., Benemérito de las Américas", "Kolega2357", 0L, 1L, -1L},
+ new Object[]{"#sv.wikipedia", 1466985600000L,
"Salo Toraut", "Lsjbot", 31L, 0L, 31L},
+ new Object[]{"#zh.wikipedia", 1466989320000L,
"中共十八大以来的反腐败工作", "2001:DA8:207:E132:94DC:BA03:DFDF:8F9F", 18L, 0L, 18L},
+ new Object[]{"#zh.wikipedia", 1466992920000L,
"Wikipedia:頁面存廢討論/記錄/2016/06/27", "Tigerzeng", 1986L, 0L, 1986L}
+ )
+ )
+ .setExpectedCountersForStageWorkerChannel(
+ CounterSnapshotMatcher
+
.with().rows(20).bytes(toRead.length()).files(1).totalFiles(1),
+ 0, 0, "input0"
+ )
+
.setExpectedSegmentGenerationProgressCountersForStageWorker(
+ CounterSnapshotMatcher
+ .with().segmentRowsProcessed(20),
+ 1, 0
+ )
+ .verifyResults();
+ }
+
@MethodSource("data")
@ParameterizedTest(name = "{index}:with context {0}")
public void testInsertOnFoo1WithGroupByLimitWithoutClusterBy(String
contextName, Map<String, Object> context)
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
index 10d9b09d81a..fe3a7cd7f79 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
@@ -120,7 +120,7 @@ public class MSQCompactionRunnerTest
null
);
private static final LongDimensionSchema LONG_DIMENSION = new
LongDimensionSchema("long_dim");
- private static final NestedDataColumnSchema NESTED_DIMENSION = new
NestedDataColumnSchema("nested_dim", 5);
+ private static final NestedDataColumnSchema NESTED_DIMENSION = new
NestedDataColumnSchema("nested_dim",
NestedDataColumnSchema.DEFAULT_FORMAT_VERSION);
private static final AutoTypeColumnSchema AUTO_DIMENSION =
AutoTypeColumnSchema.of("auto_dim");
private static final List<DimensionSchema> DIMENSIONS = ImmutableList.of(
STRING_DIMENSION,
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
index 02cb12d7444..19ae4829f59 100644
--- a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
+++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
@@ -218,6 +218,7 @@ import org.apache.druid.storage.StorageConnectorModule;
import org.apache.druid.storage.StorageConnectorProvider;
import org.apache.druid.storage.local.LocalFileStorageConnector;
import org.apache.druid.test.utils.TestSegmentManager;
+import org.apache.druid.timeline.ClusterGroupTuples;
import org.apache.druid.timeline.CompactionState;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.PruneLoadSpec;
@@ -1289,6 +1290,8 @@ public class MSQTestBase extends BaseCalciteQueryTest
private List<AggregateProjectionMetadata> expectedProjections = null;
+ private ClusterGroupTuples expectedClusterGroups = null;
+
private IngestTester()
{
// nothing to do
@@ -1336,6 +1339,12 @@ public class MSQTestBase extends BaseCalciteQueryTest
return this;
}
+ public IngestTester setExpectedClusterGroups(ClusterGroupTuples
expectedClusterGroups)
+ {
+ this.expectedClusterGroups = expectedClusterGroups;
+ return this;
+ }
+
public void verifyResults()
{
Preconditions.checkArgument(
@@ -1441,16 +1450,24 @@ public class MSQTestBase extends BaseCalciteQueryTest
// assert query granularity
Assert.assertEquals(expectedQueryGranularity,
queryableIndex.getMetadata().getQueryGranularity());
- // assert aggregator factories
+ // assert aggregator factories; clustered base table segments have
no aggregator metadata (never rollup),
+ // so treat null as empty
Assert.assertArrayEquals(
expectedAggregatorFactories.toArray(new AggregatorFactory[0]),
- queryableIndex.getMetadata().getAggregators()
+ queryableIndex.getMetadata().getAggregators() == null
+ ? new AggregatorFactory[0]
+ : queryableIndex.getMetadata().getAggregators()
);
if (expectedProjections != null) {
Assert.assertEquals(expectedProjections,
queryableIndex.getMetadata().getProjections());
}
+ if (expectedClusterGroups != null) {
+ Assert.assertEquals(expectedClusterGroups,
dataSegment.getClusterGroups());
+
Assert.assertNotNull(queryableIndex.getMetadata().getClusteredBaseTable());
+ }
+
for (List<Object> row :
FrameTestUtil.readRowsFromCursorFactory(cursorFactory).toList()) {
// transforming rows for sketch assertions
List<Object> transformedRow = row.stream()
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java
index ef519108834..af88c704d1b 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java
@@ -48,7 +48,7 @@ import org.apache.druid.query.policy.PolicyEnforcer;
import org.apache.druid.query.rowsandcols.serde.WireTransferableContext;
import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.IndexMerger;
-import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.IndexMergerV10;
import org.apache.druid.segment.SegmentWrangler;
import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.incremental.NoopRowIngestionMeters;
@@ -316,11 +316,10 @@ public class MSQTestWorkerContext implements WorkerContext
@Override
public IndexMerger indexMerger()
{
- return new IndexMergerV9(
+ return new IndexMergerV10(
mapper,
indexIO(),
- OffHeapMemorySegmentWriteOutMediumFactory.instance(),
- true
+ OffHeapMemorySegmentWriteOutMediumFactory.instance()
);
}
diff --git
a/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java
b/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java
index b8179e463c1..216fcf5ef50 100644
---
a/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java
+++
b/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java
@@ -38,6 +38,7 @@ import org.apache.druid.segment.DimensionHandlerUtils;
import org.apache.druid.segment.IndexSpec;
import org.apache.druid.segment.NestedDataColumnSchema;
import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.ColumnTypeFactory;
import org.apache.druid.segment.column.TypeSignature;
import org.apache.druid.segment.column.ValueType;
import org.apache.druid.segment.incremental.IncrementalIndex;
@@ -90,7 +91,11 @@ public abstract class DimensionSchema
case DOUBLE:
return new DoubleDimensionSchema(name);
default:
- // the auto column indexer can handle any type
+ // Use the auto column with type coercion when dealing with arrays of
primitive types.
+ // Otherwise, allow type inference, which may yield a column of a
different type.
+ if (type.isPrimitiveArray()) {
+ return new AutoTypeColumnSchema(name,
ColumnTypeFactory.ofType(type), null);
+ }
return AutoTypeColumnSchema.of(name);
}
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/DefaultColumnFormatConfig.java
b/processing/src/main/java/org/apache/druid/segment/DefaultColumnFormatConfig.java
index 56118b02686..af26e802eb6 100644
---
a/processing/src/main/java/org/apache/druid/segment/DefaultColumnFormatConfig.java
+++
b/processing/src/main/java/org/apache/druid/segment/DefaultColumnFormatConfig.java
@@ -37,7 +37,7 @@ public class DefaultColumnFormatConfig
public static Integer validateNestedFormatVersion(@Nullable Integer
formatVersion)
{
if (formatVersion != null) {
- if (formatVersion != 5) {
+ if (formatVersion != NestedDataColumnSchema.DEFAULT_FORMAT_VERSION) {
LOG.warn("Unsupported nested column format version[%s], using default
version instead", formatVersion);
return null;
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/NestedDataColumnSchema.java
b/processing/src/main/java/org/apache/druid/segment/NestedDataColumnSchema.java
index 4d7a2790a80..ca6c372872e 100644
---
a/processing/src/main/java/org/apache/druid/segment/NestedDataColumnSchema.java
+++
b/processing/src/main/java/org/apache/druid/segment/NestedDataColumnSchema.java
@@ -36,6 +36,13 @@ import java.util.Objects;
*/
public class NestedDataColumnSchema extends DimensionSchema
{
+ /**
+ * The current (and only supported) nested column format version. This is
sort of a lie: the segment does not
+ * actually contain a v5 nested column, rather v0 of the 'nested common
format', but as far as this schema is
+ * concerned it is v5.
+ */
+ public static final int DEFAULT_FORMAT_VERSION = 5;
+
final int formatVersion;
@Nullable
final NestedCommonFormatColumnFormatSpec columnFormatSpec;
@@ -54,9 +61,7 @@ public class NestedDataColumnSchema extends DimensionSchema
} else if (defaultFormatConfig.getNestedColumnFormatVersion() != null) {
formatVersion = defaultFormatConfig.getNestedColumnFormatVersion();
} else {
- // this is sort of a lie... it's not really v5 in the segment, rather
its v0 of the 'nested common format'
- // but as far as this is concerned it is v5
- formatVersion = 5;
+ formatVersion = DEFAULT_FORMAT_VERSION;
}
DefaultColumnFormatConfig.validateNestedFormatVersion(formatVersion);
if (columnFormatSpec == null) {
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/NestedDataComplexTypeSerde.java
b/processing/src/main/java/org/apache/druid/segment/nested/NestedDataComplexTypeSerde.java
index a1496dd8dcc..6bd7e468834 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/NestedDataComplexTypeSerde.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/NestedDataComplexTypeSerde.java
@@ -159,7 +159,7 @@ public class NestedDataComplexTypeSerde extends
ComplexMetricSerde
@Override
public DimensionSchema getColumnSchema(String columnName)
{
- return new NestedDataColumnSchema(columnName, 5);
+ return new NestedDataColumnSchema(columnName,
NestedDataColumnSchema.DEFAULT_FORMAT_VERSION);
}
@Override
diff --git
a/processing/src/test/java/org/apache/druid/data/input/impl/DimensionSchemaTest.java
b/processing/src/test/java/org/apache/druid/data/input/impl/DimensionSchemaTest.java
index 4341cc7529d..1a601955b5d 100644
---
a/processing/src/test/java/org/apache/druid/data/input/impl/DimensionSchemaTest.java
+++
b/processing/src/test/java/org/apache/druid/data/input/impl/DimensionSchemaTest.java
@@ -21,6 +21,8 @@ package org.apache.druid.data.input.impl;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.exc.InvalidTypeIdException;
+import org.apache.druid.segment.AutoTypeColumnSchema;
+import org.apache.druid.segment.column.ColumnType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -69,4 +71,49 @@ public class DimensionSchemaTest
OBJECT_MAPPER.readValue(noType, DimensionSchema.class)
);
}
+
+ @Test
+ public void testGetDefaultSchemaForBuiltInType()
+ {
+ Assertions.assertEquals(
+ new StringDimensionSchema("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.STRING)
+ );
+ Assertions.assertEquals(
+ new LongDimensionSchema("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo", ColumnType.LONG)
+ );
+ Assertions.assertEquals(
+ new FloatDimensionSchema("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo", ColumnType.FLOAT)
+ );
+ Assertions.assertEquals(
+ new DoubleDimensionSchema("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.DOUBLE)
+ );
+ // Primitive arrays cast an auto column to the given type rather than
leaving the physical type to be inferred
+ // from the ingested values (an all-null batch has no values to infer the
array type from). The auto schema
+ // stores ARRAY<FLOAT> as ARRAY<DOUBLE>.
+ Assertions.assertEquals(
+ new AutoTypeColumnSchema("foo", ColumnType.STRING_ARRAY, null),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.STRING_ARRAY)
+ );
+ Assertions.assertEquals(
+ new AutoTypeColumnSchema("foo", ColumnType.LONG_ARRAY, null),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.LONG_ARRAY)
+ );
+ Assertions.assertEquals(
+ new AutoTypeColumnSchema("foo", ColumnType.DOUBLE_ARRAY, null),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.FLOAT_ARRAY)
+ );
+ // Complex types remain untyped auto columns.
+ Assertions.assertEquals(
+ AutoTypeColumnSchema.of("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.NESTED_DATA)
+ );
+ Assertions.assertEquals(
+ AutoTypeColumnSchema.of("foo"),
+ DimensionSchema.getDefaultSchemaForBuiltInType("foo",
ColumnType.ofComplex("hyperUnique"))
+ );
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/NestedDataColumnSchemaTest.java
b/processing/src/test/java/org/apache/druid/segment/NestedDataColumnSchemaTest.java
index 66fd98ce043..99071b686ca 100644
---
a/processing/src/test/java/org/apache/druid/segment/NestedDataColumnSchemaTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/NestedDataColumnSchemaTest.java
@@ -83,7 +83,7 @@ public class NestedDataColumnSchemaTest
{
final String there = "{\"type\":\"json\", \"name\":\"test\"}";
NestedDataColumnSchema andBack = MAPPER.readValue(there,
NestedDataColumnSchema.class);
- Assertions.assertEquals(new NestedDataColumnSchema("test", 5), andBack);
+ Assertions.assertEquals(new NestedDataColumnSchema("test",
NestedDataColumnSchema.DEFAULT_FORMAT_VERSION), andBack);
}
@Test
diff --git
a/server/src/main/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadata.java
b/server/src/main/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadata.java
new file mode 100644
index 00000000000..e84cc853c63
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadata.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.catalog.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import
org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.segment.NestedDataColumnSchema;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.utils.CollectionUtils;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Catalog layout metadata for {@link
ClusteredValueGroupsBaseTableProjectionSpec} base tables. Declares the
+ * {@link #clusteringColumns} (names of declared catalog columns that rows are
clustered by) and optional
+ * {@link #virtualColumns} that compute stored columns at ingest time;
everything else about the physical spec
+ * (column names, types, and order) is taken from the catalog column list by
{@link #createSpec(List)}. The declared
+ * column order is the physical segment order, so, mirroring the physical
spec, the clustering columns must be
+ * declared as the leading prefix of the column list.
+ */
+@JsonTypeName(ClusteredValueGroupsBaseTableMetadata.TYPE_NAME)
+public class ClusteredValueGroupsBaseTableMetadata implements
DatasourceBaseTableMetadata
+{
+ public static final String TYPE_NAME =
ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME;
+
+ private final List<String> clusteringColumns;
+ private final VirtualColumns virtualColumns;
+
+ @JsonCreator
+ public ClusteredValueGroupsBaseTableMetadata(
+ @JsonProperty("clusteringColumns") List<String> clusteringColumns,
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns
+ )
+ {
+ this.clusteringColumns = clusteringColumns == null ?
Collections.emptyList() : clusteringColumns;
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY :
virtualColumns;
+ }
+
+ @Override
+ @JsonProperty("type")
+ public String getType()
+ {
+ return TYPE_NAME;
+ }
+
+ @JsonProperty("clusteringColumns")
+ public List<String> getClusteringColumns()
+ {
+ return clusteringColumns;
+ }
+
+ @Override
+ @JsonProperty("virtualColumns")
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ public VirtualColumns getVirtualColumns()
+ {
+ return virtualColumns;
+ }
+
+ /**
+ * Creates the physical spec from the declared catalog columns, used
verbatim: the declared column order is the
+ * physical segment order, so the clustering columns must be declared as the
leading prefix of the column list (in
+ * {@link #clusteringColumns} order); anything else is a validation error.
Every clustering column must be a declared
+ * column, a clustering column computed by a virtual column at ingest time
is still a stored, queryable column, so it
+ * too must appear in the column list. All ordering and layout rules are
enforced by the spec itself.
+ */
+ @Override
+ public ClusteredValueGroupsBaseTableProjectionSpec
createSpec(List<ColumnSpec> columns)
+ {
+ if (CollectionUtils.isNullOrEmpty(columns)) {
+ throw InvalidInput.exception(
+ "Cannot define a [%s] base table without declared columns; the
catalog column list defines the table schema",
+ TYPE_NAME
+ );
+ }
+ final Set<String> declaredNames = new HashSet<>();
+ final List<DimensionSchema> specColumns = new ArrayList<>(columns.size());
+ for (ColumnSpec column : columns) {
+ declaredNames.add(column.name());
+ specColumns.add(toDimensionSchema(column));
+ }
+ for (String clusteringColumn : clusteringColumns) {
+ if (!declaredNames.contains(clusteringColumn)) {
+ throw InvalidInput.exception(
+ "clustering column [%s] is not a declared column; clustering
columns must be declared as the leading"
+ + " prefix of the table's column list, including columns computed
by a virtual column at ingest time"
+ + " (they are stored columns)",
+ clusteringColumn
+ );
+ }
+ }
+ return ClusteredValueGroupsBaseTableProjectionSpec.builder()
+
.virtualColumns(virtualColumns)
+ .columns(specColumns)
+
.clusteringColumns(clusteringColumns)
+ .build();
+ }
+
+ private static DimensionSchema toDimensionSchema(ColumnSpec column)
+ {
+ ColumnType druidType = Columns.druidType(column);
+ if (druidType == null) {
+ druidType = ColumnType.STRING;
+ }
+ if (druidType.isPrimitive() || druidType.isPrimitiveArray()) {
+ // The declared type is retained in the ingestion schema (primitive
arrays are cast, rather than left to an
+ // untyped auto column whose type is inferred from the ingested values;
note that the auto schema stores
+ // FLOAT ARRAY as DOUBLE ARRAY).
+ return DimensionSchema.getDefaultSchemaForBuiltInType(column.name(),
druidType);
+ }
+ if (ColumnType.NESTED_DATA.equals(druidType)) {
+ return new NestedDataColumnSchema(column.name(),
NestedDataColumnSchema.DEFAULT_FORMAT_VERSION);
+ }
+ // Other complex types cannot be ingested into a clustered base table:
there is no dimension handler for them,
+ // and clustered base tables have no aggregators to produce them.
+ throw InvalidInput.exception(
+ "column [%s] has unsupported type [%s] for a clustered base table;
supported types are primitive, primitive"
+ + " array, and COMPLEX<json> columns",
+ column.name(),
+ druidType
+ );
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ClusteredValueGroupsBaseTableMetadata that =
(ClusteredValueGroupsBaseTableMetadata) o;
+ return Objects.equals(clusteringColumns, that.clusteringColumns)
+ && Objects.equals(virtualColumns, that.virtualColumns);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(clusteringColumns, virtualColumns);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "ClusteredValueGroupsBaseTableMetadata{" +
+ "clusteringColumns=" + clusteringColumns +
+ ", virtualColumns=" + virtualColumns +
+ '}';
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/catalog/model/DatasourceBaseTableMetadata.java
b/server/src/main/java/org/apache/druid/catalog/model/DatasourceBaseTableMetadata.java
new file mode 100644
index 00000000000..b73cfd7f4b4
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/catalog/model/DatasourceBaseTableMetadata.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.catalog.model;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.impl.BaseTableProjectionSpec;
+import org.apache.druid.segment.VirtualColumns;
+
+import java.util.List;
+
+/**
+ * Catalog model for the physical layout of a datasource "base table", the
counterpart of
+ * {@link BaseTableProjectionSpec} for tables defined in the catalog. Unlike
the physical spec, this metadata does NOT
+ * declare the table schema: the catalog column list remains the single source
of truth for column names, types, and
+ * order. Implementations carry only the layout details the column list cannot
express (such as clustering columns and
+ * ingest-time virtual columns), and {@link #createSpec(List)} combines the
two into the physical spec used to
+ * generate segments.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include =
JsonTypeInfo.As.EXISTING_PROPERTY, property = "type")
+@JsonSubTypes(value = {
+ @JsonSubTypes.Type(
+ name = ClusteredValueGroupsBaseTableMetadata.TYPE_NAME,
+ value = ClusteredValueGroupsBaseTableMetadata.class
+ )
+})
+public interface DatasourceBaseTableMetadata
+{
+ /**
+ * The type discriminator, declared as a regular JSON property ({@link
JsonTypeInfo.As#EXISTING_PROPERTY}) rather
+ * than a synthetic one so that it is emitted even when an instance is
serialized from an untyped context;catalog
+ * property values live in a plain {@code Map<String, Object>}, where
Jackson serializes by runtime type and would
+ * otherwise drop a synthetic type id.
+ */
+ String getType();
+
+ /**
+ * Virtual columns computed at ingest time to materialize stored columns.
+ */
+ VirtualColumns getVirtualColumns();
+
+ /**
+ * Creates the physical {@link BaseTableProjectionSpec} by combining this
layout metadata with the declared catalog
+ * columns. Throws {@link org.apache.druid.error.DruidException} if the
layout is inconsistent with the declared
+ * columns, so this doubles as the cross-validation of the catalog property
against the column list.
+ */
+ BaseTableProjectionSpec createSpec(List<ColumnSpec> columns);
+}
diff --git
a/server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java
b/server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java
index 39027ac8f09..1ee50c75c6e 100644
---
a/server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java
+++
b/server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java
@@ -19,14 +19,18 @@
package org.apache.druid.catalog.model.facade;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableMap;
import org.apache.druid.catalog.model.CatalogUtils;
import org.apache.druid.catalog.model.ColumnSpec;
import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.DatasourceBaseTableMetadata;
import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.table.ClusterKeySpec;
import org.apache.druid.catalog.model.table.DatasourceDefn;
+import org.apache.druid.error.InvalidSqlInput;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.segment.column.ColumnType;
@@ -104,6 +108,12 @@ public class DatasourceFacade extends TableFacade
private final List<ColumnFacade> columns;
private final Map<String, ColumnFacade> columnIndex;
+ /**
+ * Memoized: facades live as long as their cached table entry and {@link
#insertColumn} consults this per column,
+ * while decoding the property does a Jackson conversion each time.
+ */
+ private final Supplier<DatasourceBaseTableMetadata> baseTableMetadata =
+ Suppliers.memoize(this::decodeBaseTableMetadata);
public DatasourceFacade(ResolvedTable resolved)
{
@@ -146,7 +156,7 @@ public class DatasourceFacade extends TableFacade
return jsonMapper().convertValue(value,
ClusterKeySpec.CLUSTER_KEY_LIST_TYPE_REF);
}
catch (Exception e) {
- LOG.error("Failed to convert a catalog %s property of value [%s]",
+ LOG.error("Failed to convert a catalog [%s] property of value [%s]",
DatasourceDefn.CLUSTER_KEYS_PROPERTY,
value
);
@@ -176,6 +186,45 @@ public class DatasourceFacade extends TableFacade
return columnIndex.get(name);
}
+ /**
+ * Resolves a column as the target of an INSERT/REPLACE write, enforcing the
catalog write rules: a column that is
+ * not declared is rejected when the table schema is {@link #isSealed()
sealed}, and a column computed by a base
+ * table layout virtual column at ingest time is always rejected, its stored
value is always the computed one, so
+ * the supplied values would be silently discarded. Returns the column
facade for a declared column, or null for an
+ * undeclared column of a non-sealed table (the caller adds the column based
on the query type).
+ *
+ * @param columnName name of the column the query writes to
+ * @param targetTableName display name of the target table, used only in
error messages. This is the table this
+ * facade describes, but supplied by the caller: a
{@link ResolvedTable} does not know its
+ * own name (identity lives on {@link
org.apache.druid.catalog.model.TableMetadata}), and
+ * SQL callers report the name as written in the
query rather than the catalog identifier.
+ */
+ @Nullable
+ public ColumnFacade insertColumn(String columnName, String targetTableName)
+ {
+ final ColumnFacade definedCol = column(columnName);
+ if (definedCol == null) {
+ if (isSealed()) {
+ throw InvalidSqlInput.exception(
+ "Column [%s] is not defined in the target table [%s] strict
schema",
+ columnName,
+ targetTableName
+ );
+ }
+ return null;
+ }
+ final DatasourceBaseTableMetadata baseTable = baseTableMetadata.get();
+ if (baseTable != null && baseTable.getVirtualColumns().exists(columnName))
{
+ throw InvalidSqlInput.exception(
+ "Column [%s] of target table [%s] is computed by a virtual column at
ingest time and cannot be written"
+ + " directly; supply the virtual column's input columns instead",
+ columnName,
+ targetTableName
+ );
+ }
+ return definedCol;
+ }
+
@Nullable
public List<DatasourceProjectionMetadata> projections()
{
@@ -188,11 +237,38 @@ public class DatasourceFacade extends TableFacade
}
catch (Exception e) {
LOG.error(
- "Failed to convert a catalog %s property of value [%s]",
+ "Failed to convert a catalog [%s] property of value [%s]",
DatasourceDefn.PROJECTIONS_KEYS_PROPERTY,
value
);
return null;
}
}
+
+ @Nullable
+ public DatasourceBaseTableMetadata baseTableMetadata()
+ {
+ return baseTableMetadata.get();
+ }
+
+ @Nullable
+ private DatasourceBaseTableMetadata decodeBaseTableMetadata()
+ {
+ Object value = property(DatasourceDefn.BASE_TABLE_PROPERTY);
+ if (value == null) {
+ return null;
+ }
+ try {
+ return jsonMapper().convertValue(value,
DatasourceDefn.BaseTableDefn.TYPE_REF);
+ }
+ catch (Exception e) {
+ LOG.error(
+ e,
+ "Failed to convert a catalog [%s] property of value [%s]",
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ value
+ );
+ return null;
+ }
+ }
}
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 723beaf2d89..ef3a41dba46 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
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.druid.catalog.model.ColumnSpec;
import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.DatasourceBaseTableMetadata;
import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
import org.apache.druid.catalog.model.ModelProperties;
import org.apache.druid.catalog.model.ModelProperties.GranularityPropertyDefn;
@@ -30,6 +31,7 @@ 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;
@@ -63,6 +65,14 @@ public class DatasourceDefn extends TableDefn
public static final String PROJECTIONS_KEYS_PROPERTY = "projections";
+ /**
+ * Physical layout of the base table. The layout combines with the declared
column list (which remains the source of
+ * truth for column names, types, and order) to derive the physical spec
used to generate segments;
+ *
+ * @see DatasourceBaseTableMetadata#createSpec
+ */
+ public static final String BASE_TABLE_PROPERTY = "baseTable";
+
/**
* The set of existing columns to "delete" (actually, just hide) from the
* SQL layer. Used to "remove" unwanted columns to avoid the need to rewrite
@@ -85,12 +95,37 @@ public class DatasourceDefn extends TableDefn
new ClusterKeysDefn(),
new HiddenColumnsDefn(),
new ModelProperties.BooleanPropertyDefn(SEALED_PROPERTY),
- new ProjectionsDefn()
+ new ProjectionsDefn(),
+ new BaseTableDefn()
),
null
);
}
+ @Override
+ 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());
+ }
+ }
+
@Override
protected void validateColumn(ColumnSpec spec)
{
@@ -182,4 +217,14 @@ public class DatasourceDefn extends TableDefn
super(PROJECTIONS_KEYS_PROPERTY, "DatasourceProjectionMetadata list",
TYPE_REF);
}
}
+
+ public static class BaseTableDefn extends
ModelProperties.TypeRefPropertyDefn<DatasourceBaseTableMetadata>
+ {
+ public static final TypeReference<DatasourceBaseTableMetadata> TYPE_REF =
new TypeReference<>() {};
+
+ public BaseTableDefn()
+ {
+ super(BASE_TABLE_PROPERTY, "DatasourceBaseTableMetadata", TYPE_REF);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/druid/catalog/model/table/TableBuilder.java
b/server/src/main/java/org/apache/druid/catalog/model/table/TableBuilder.java
index e2a9cbe6537..bb32ca68665 100644
---
a/server/src/main/java/org/apache/druid/catalog/model/table/TableBuilder.java
+++
b/server/src/main/java/org/apache/druid/catalog/model/table/TableBuilder.java
@@ -24,6 +24,7 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.druid.catalog.model.ColumnSpec;
import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.DatasourceBaseTableMetadata;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.TableDefn;
import org.apache.druid.catalog.model.TableId;
@@ -141,6 +142,11 @@ public class TableBuilder
return property(DatasourceDefn.CLUSTER_KEYS_PROPERTY,
Arrays.asList(clusterKeys));
}
+ public TableBuilder baseTable(DatasourceBaseTableMetadata baseTable)
+ {
+ return property(DatasourceDefn.BASE_TABLE_PROPERTY, baseTable);
+ }
+
public TableBuilder hiddenColumns(List<String> hiddenColumns)
{
return property(DatasourceDefn.HIDDEN_COLUMNS_PROPERTY, hiddenColumns);
diff --git
a/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java
b/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java
new file mode 100644
index 00000000000..2d99f03a8e2
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.catalog.model;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import
org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.DoubleDimensionSchema;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.segment.AutoTypeColumnSchema;
+import org.apache.druid.segment.NestedDataColumnSchema;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class ClusteredValueGroupsBaseTableMetadataTest
+{
+ private final ObjectMapper mapper = new
DefaultObjectMapper().setInjectableValues(
+ new InjectableValues.Std().addValue(ExprMacroTable.class,
ExprMacroTable.nil())
+ );
+
+ // Declared order is the physical segment order: clustering columns lead,
__time is an explicit positional column.
+ private static final List<ColumnSpec> COLUMNS = Arrays.asList(
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, Columns.SQL_TIMESTAMP, null),
+ new ColumnSpec("region", null, null),
+ new ColumnSpec("delta", Columns.SQL_BIGINT, null),
+ new ColumnSpec("value", Columns.SQL_DOUBLE, null)
+ );
+
+ @Test
+ public void testSerde() throws Exception
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant_lower"),
+ VirtualColumns.create(
+ new ExpressionVirtualColumn("tenant_lower", "lower(\"tenant\")",
ColumnType.STRING, ExprMacroTable.nil())
+ )
+ );
+ final String json = mapper.writeValueAsString(metadata);
+ final DatasourceBaseTableMetadata fromJson = mapper.readValue(json,
DatasourceBaseTableMetadata.class);
+ Assert.assertEquals(metadata, fromJson);
+ }
+
+ @Test
+ public void testSerdeNoVirtualColumns() throws Exception
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Arrays.asList("tenant", "region"),
+ null
+ );
+ final String json = mapper.writeValueAsString(metadata);
+ Assert.assertFalse(json.contains("virtualColumns"));
+ final DatasourceBaseTableMetadata fromJson = mapper.readValue(json,
DatasourceBaseTableMetadata.class);
+ Assert.assertEquals(metadata, fromJson);
+ }
+
+ @Test
+ public void testSerdeAsUntypedMapValue() throws Exception
+ {
+ // Catalog property values are serialized from a Map<String, Object>,
where Jackson serializes by runtime type;
+ // the type discriminator must survive that path (it is an
EXISTING_PROPERTY for this reason).
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ final String json =
mapper.writeValueAsString(Collections.singletonMap("baseTable", metadata));
+ Assert.assertTrue(json.contains("\"type\":\"clusteredValueGroups\""));
+ final Map<String, Object> untyped = mapper.readValue(json, new
TypeReference<>() {});
+ Assert.assertEquals(
+ metadata,
+ mapper.convertValue(untyped.get("baseTable"),
DatasourceBaseTableMetadata.class)
+ );
+ }
+
+ @Test
+ public void testCreateSpec()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ // The declared column order is the physical segment order, used verbatim;
types map through Columns.druidType
+ // with untyped -> STRING.
+ Assert.assertEquals(
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new
StringDimensionSchema("tenant"),
+ new
LongDimensionSchema(Columns.TIME_COLUMN),
+ new
StringDimensionSchema("region"),
+ new
LongDimensionSchema("delta"),
+ new
DoubleDimensionSchema("value")
+ )
+ .clusteringColumns("tenant")
+ .build(),
+ metadata.createSpec(COLUMNS)
+ );
+ }
+
+ @Test
+ public void testCreateSpecWithVirtualColumn()
+ {
+ // A clustering column computed at ingest time: the virtual column
materializes the stored, declared
+ // 'tenant_lower' column, reading the stored 'tenant' column (virtual
column inputs must themselves be stored
+ // columns or other virtual columns).
+ final VirtualColumns virtualColumns = VirtualColumns.create(
+ new ExpressionVirtualColumn("tenant_lower", "lower(\"tenant\")",
ColumnType.STRING, ExprMacroTable.nil())
+ );
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant_lower"),
+ virtualColumns
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tenant_lower", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec("region", Columns.SQL_VARCHAR, null)
+ );
+ Assert.assertEquals(
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+
.virtualColumns(virtualColumns)
+ .columns(
+ new
StringDimensionSchema("tenant_lower"),
+ new
LongDimensionSchema(Columns.TIME_COLUMN),
+ new
StringDimensionSchema("tenant"),
+ new
StringDimensionSchema("region")
+ )
+
.clusteringColumns("tenant_lower")
+ .build(),
+ metadata.createSpec(columns)
+ );
+ }
+
+ @Test
+ public void testCreateSpecMultipleClusteringColumns()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Arrays.asList("region", "tenant"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("region", Columns.SQL_VARCHAR, null),
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("delta", Columns.SQL_BIGINT, null)
+ );
+ Assert.assertEquals(
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new
StringDimensionSchema("region"),
+ new
StringDimensionSchema("tenant"),
+ new
LongDimensionSchema(Columns.TIME_COLUMN),
+ new
LongDimensionSchema("delta")
+ )
+
.clusteringColumns("region", "tenant")
+ .build(),
+ metadata.createSpec(columns)
+ );
+ }
+
+ @Test
+ public void testCreateSpecRetainsDeclaredArrayAndNestedTypes()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("tags", Columns.SQL_VARCHAR_ARRAY, null),
+ new ColumnSpec("vals", Columns.SQL_BIGINT_ARRAY, null),
+ new ColumnSpec("ratios", Columns.SQL_FLOAT_ARRAY, null),
+ new ColumnSpec("attrs", ColumnType.NESTED_DATA.asTypeString(), null)
+ );
+ // Declared types are retained in the ingestion schema rather than left to
inference: arrays cast an auto column
+ // to the declared type (an all-null batch has no values to infer from;
FLOAT ARRAY is stored as DOUBLE ARRAY by
+ // the auto schema), and COMPLEX<json> uses the dedicated nested column
schema.
+ Assert.assertEquals(
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new
StringDimensionSchema("tenant"),
+ new
LongDimensionSchema(Columns.TIME_COLUMN),
+ new
AutoTypeColumnSchema("tags", ColumnType.STRING_ARRAY, null),
+ new
AutoTypeColumnSchema("vals", ColumnType.LONG_ARRAY, null),
+ new
AutoTypeColumnSchema("ratios", ColumnType.DOUBLE_ARRAY, null),
+ new
NestedDataColumnSchema("attrs", NestedDataColumnSchema.DEFAULT_FORMAT_VERSION)
+ )
+ .clusteringColumns("tenant")
+ .build(),
+ metadata.createSpec(columns)
+ );
+ }
+
+ @Test
+ public void testCreateSpecUnsupportedComplexTypeFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("unique_things", "COMPLEX<hyperUnique>", null)
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(columns));
+ Assert.assertTrue(e.getMessage().contains("column [unique_things] has
unsupported type [COMPLEX<hyperUnique>]"));
+ }
+
+ @Test
+ public void testCreateSpecClusteringColumnsNotLeadingPrefixFails()
+ {
+ // 'region' is declared, but not as part of the leading prefix of the
column list; the declared order is the
+ // physical segment order, so this is an error rather than a silent
reorder.
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("region"),
+ null
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(COLUMNS));
+ Assert.assertTrue(e.getMessage().contains("clusteringColumns must be the
leading prefix of columns"));
+ }
+
+ @Test
+ public void testCreateSpecClusteringColumnsWrongOrderFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Arrays.asList("tenant", "region"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("region", Columns.SQL_VARCHAR, null),
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null)
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(columns));
+ Assert.assertTrue(e.getMessage().contains("clusteringColumns must be the
leading prefix of columns"));
+ }
+
+ @Test
+ public void testCreateSpecUndeclaredClusteringColumnFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("no_such_column"),
+ null
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(COLUMNS));
+ Assert.assertTrue(e.getMessage().contains("clustering column
[no_such_column] is not a declared column"));
+ }
+
+ @Test
+ public void testCreateSpecMissingTimeColumnFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec("region", Columns.SQL_VARCHAR, null)
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(columns));
+ Assert.assertTrue(e.getMessage().contains("must include [__time]"));
+ }
+
+ @Test
+ public void testCreateSpecDisallowedClusteringTypeFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tags"),
+ null
+ );
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tags", Columns.SQL_VARCHAR_ARRAY, null),
+ new ColumnSpec(Columns.TIME_COLUMN, Columns.SQL_TIMESTAMP, null)
+ );
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(columns));
+ Assert.assertTrue(e.getMessage().contains("unsupported type"));
+ }
+
+ @Test
+ public void testCreateSpecEmptyClusteringColumnsFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(null, null);
+ final DruidException e = Assert.assertThrows(DruidException.class, () ->
metadata.createSpec(COLUMNS));
+ Assert.assertTrue(e.getMessage().contains("clusteringColumns must be
non-empty"));
+ }
+
+ @Test
+ public void testCreateSpecNoDeclaredColumnsFails()
+ {
+ final DatasourceBaseTableMetadata metadata = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant"),
+ null
+ );
+ final DruidException e = Assert.assertThrows(
+ DruidException.class,
+ () -> metadata.createSpec(Collections.emptyList())
+ );
+ Assert.assertTrue(e.getMessage().contains("without declared columns"));
+ }
+
+ @Test
+ public void testEquals()
+ {
+ EqualsVerifier.forClass(ClusteredValueGroupsBaseTableMetadata.class)
+ .usingGetClass()
+ .verify();
+ }
+}
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 b93db97acc6..b4e06cb1a01 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
@@ -19,23 +19,30 @@
package org.apache.druid.catalog.model.table;
+import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import nl.jqno.equalsverifier.EqualsVerifier;
import org.apache.druid.catalog.CatalogTest;
+import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata;
import org.apache.druid.catalog.model.ColumnSpec;
import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.DatasourceBaseTableMetadata;
import org.apache.druid.catalog.model.ResolvedTable;
import org.apache.druid.catalog.model.TableDefn;
import org.apache.druid.catalog.model.TableDefnRegistry;
import org.apache.druid.catalog.model.TableSpec;
import org.apache.druid.catalog.model.facade.DatasourceFacade;
import org.apache.druid.catalog.model.facade.DatasourceFacade.ColumnFacade;
+import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.segment.VirtualColumns;
import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -139,6 +146,142 @@ public class DatasourceTableTest
}
}
+ @Test
+ public void testSpecWithBaseTableProp()
+ {
+ // The declared column order is the physical segment order: the clustering
columns must lead.
+ final List<ColumnSpec> columns = Arrays.asList(
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("region", Columns.SQL_VARCHAR, null)
+ );
+
+ {
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(Collections.singletonList("tenant"), null)
+ ),
+ columns
+ );
+ expectValidationSucceeds(spec);
+ }
+
+ {
+ // A base table layout requires 'sealed': the declared columns define
the physical segment schema, so
+ // undeclared columns cannot be ingested.
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(Collections.singletonList("tenant"), null)
+ ),
+ columns
+ );
+ ResolvedTable table = registry.resolve(spec);
+ DruidException e = assertThrows(DruidException.class, table::validate);
+ assertTrue(e.getMessage().contains("must also set [sealed] to true"));
+ }
+
+ {
+ // Validation is cross-checked against the declared columns: an
undeclared clustering column fails at
+ // catalog write time.
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(Collections.singletonList("no_such_column"),
null)
+ ),
+ columns
+ );
+ ResolvedTable table = registry.resolve(spec);
+ assertThrows(DruidException.class, table::validate);
+ }
+
+ {
+ // The clustering columns must be declared as the leading prefix of the
column list; declared-but-not-first
+ // is an error, not a silent reorder.
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(Collections.singletonList("region"), null)
+ ),
+ columns
+ );
+ ResolvedTable table = registry.resolve(spec);
+ assertThrows(DruidException.class, table::validate);
+ }
+
+ {
+ // __time must be declared in the column list.
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new
ClusteredValueGroupsBaseTableMetadata(Collections.singletonList("tenant"), null)
+ ),
+ Collections.singletonList(new ColumnSpec("tenant",
Columns.SQL_VARCHAR, null))
+ );
+ ResolvedTable table = registry.resolve(spec);
+ assertThrows(DruidException.class, table::validate);
+ }
+ }
+
+ @Test
+ public void testBaseTableFacade()
+ {
+ final ObjectMapper mapperWithInjectables = new
DefaultObjectMapper().setInjectableValues(
+ new InjectableValues.Std().addValue(ExprMacroTable.class,
ExprMacroTable.nil())
+ );
+ final TableDefnRegistry registryWithInjectables = new
TableDefnRegistry(mapperWithInjectables);
+ final DatasourceBaseTableMetadata baseTable = new
ClusteredValueGroupsBaseTableMetadata(
+ Collections.singletonList("tenant_lower"),
+ VirtualColumns.create(
+ new ExpressionVirtualColumn("tenant_lower", "lower(\"tenant\")",
ColumnType.STRING, ExprMacroTable.nil())
+ )
+ );
+ TableSpec spec = new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY, baseTable
+ ),
+ Arrays.asList(
+ new ColumnSpec("tenant_lower", Columns.SQL_VARCHAR, null),
+ new ColumnSpec(Columns.TIME_COLUMN, null, null),
+ new ColumnSpec("tenant", Columns.SQL_VARCHAR, null)
+ )
+ );
+ ResolvedTable table = registryWithInjectables.resolve(spec);
+ table.validate();
+ DatasourceFacade facade = new DatasourceFacade(table);
+ assertEquals(baseTable, facade.baseTableMetadata());
+
+ // insertColumn enforces the catalog write rules: declared columns
resolve, undeclared columns are rejected by
+ // the sealed schema, and columns computed by a virtual column at ingest
time cannot be written directly.
+ assertSame(facade.column("tenant"), facade.insertColumn("tenant",
"druid.tbl"));
+ DruidException sealedError =
+ assertThrows(DruidException.class, () ->
facade.insertColumn("no_such_column", "druid.tbl"));
+ assertTrue(sealedError.getMessage().contains("not defined in the target
table [druid.tbl] strict schema"));
+ DruidException computedError =
+ assertThrows(DruidException.class, () ->
facade.insertColumn("tenant_lower", "druid.tbl"));
+ assertTrue(computedError.getMessage().contains("computed by a virtual
column at ingest time"));
+
+ // An undeclared column of a non-sealed table resolves to null: the caller
adds it from the query type.
+ DatasourceFacade unsealed = new
DatasourceFacade(registryWithInjectables.resolve(new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(),
+ Collections.singletonList(new ColumnSpec(Columns.TIME_COLUMN, null,
null))
+ )));
+ assertNull(unsealed.insertColumn("anything", "druid.tbl"));
+ }
+
@Test
public void testAllProperties()
{
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
index 66d66174cc5..aa0781b6ba8 100644
---
a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
@@ -534,21 +534,12 @@ public class DruidSqlValidator extends
BaseDruidSqlValidator
return sourceType;
}
- // disable sealed mode validation if catalog validation is disabled.
- final boolean isStrict = tableMetadata.isSealed();
final List<Map.Entry<String, RelDataType>> fields = new ArrayList<>();
for (RelDataTypeField sourceField : sourceFields) {
final String colName = sourceField.getName();
- final DatasourceFacade.ColumnFacade definedCol =
tableMetadata.column(colName);
+ final DatasourceFacade.ColumnFacade definedCol =
+ tableMetadata.insertColumn(colName,
insert.getTargetTable().toString());
if (definedCol == null) {
- if (isStrict) {
- throw InvalidSqlInput.exception(
- "Column [%s] is not defined in the target table [%s] strict
schema",
- colName,
- insert.getTargetTable()
- );
- }
-
// Table is not strict: add a new column based on the SELECT column.
fields.add(Pair.of(colName, sourceField.getType()));
continue;
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
index d89d618bda1..fcd498450c3 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
@@ -19,9 +19,11 @@
package org.apache.druid.sql.calcite;
+import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata;
import org.apache.druid.catalog.model.ColumnSpec;
import org.apache.druid.catalog.model.Columns;
import org.apache.druid.catalog.model.ResolvedTable;
@@ -36,6 +38,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.query.OrderBy;
import org.apache.druid.query.QueryContexts;
import org.apache.druid.query.TableDataSource;
@@ -43,8 +46,10 @@ import
org.apache.druid.query.aggregation.cardinality.CardinalityAggregatorFacto
import
org.apache.druid.query.aggregation.hyperloglog.HyperUniquesAggregatorFactory;
import org.apache.druid.query.dimension.DefaultDimensionSpec;
import org.apache.druid.query.groupby.GroupByQuery;
+import org.apache.druid.segment.VirtualColumns;
import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
import
org.apache.druid.sql.calcite.CalciteCatalogIngestionDmlTest.CatalogIngestionDmlComponentSupplier;
import org.apache.druid.sql.calcite.external.ExternalDataSource;
import org.apache.druid.sql.calcite.external.Externals;
@@ -92,7 +97,9 @@ public abstract class CalciteCatalogIngestionDmlTest extends
CalciteIngestionDml
super(tempFolderProducer);
}
- private static final ObjectMapper MAPPER = new DefaultObjectMapper();
+ private static final ObjectMapper MAPPER = new
DefaultObjectMapper().setInjectableValues(
+ new InjectableValues.Std().addValue(ExprMacroTable.class,
ExprMacroTable.nil())
+ );
public static ImmutableMap<String, DatasourceTable> RESOLVED_TABLES =
ImmutableMap.of(
"hourDs", new DatasourceTable(
RowSignature.builder().addTimeColumn().build(),
@@ -352,6 +359,73 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
.build()),
false
)
+ ),
+ "tableWithBaseTable", new DatasourceTable(
+ RowSignature.builder()
+ .add("tenant_lower", ColumnType.STRING)
+ .addTimeColumn()
+ .add("tenant", ColumnType.STRING)
+ .add("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build(),
+ new DatasourceTable.PhysicalDatasourceMetadata(
+ new TableDataSource("tableWithBaseTable"),
+ RowSignature.builder()
+ .add("tenant_lower", ColumnType.STRING)
+ .addTimeColumn()
+ .add("tenant", ColumnType.STRING)
+ .add("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build(),
+ false,
+ false
+ ),
+ new DatasourceTable.EffectiveMetadata(
+ new DatasourceFacade(new ResolvedTable(
+ new TableDefn(
+ "tableWithBaseTable",
+ DatasourceDefn.TABLE_TYPE,
+ null,
+ null
+ ),
+ new TableSpec(
+ DatasourceDefn.TABLE_TYPE,
+ ImmutableMap.of(
+ DatasourceDefn.SEALED_PROPERTY, true,
+ DatasourceDefn.BASE_TABLE_PROPERTY,
+ new ClusteredValueGroupsBaseTableMetadata(
+ ImmutableList.of("tenant_lower"),
+ VirtualColumns.create(
+ new ExpressionVirtualColumn(
+ "tenant_lower",
+ "lower(\"tenant\")",
+ ColumnType.STRING,
+ ExprMacroTable.nil()
+ )
+ )
+ )
+ ),
+ // declared order is the physical segment order: the
clustering column leads. 'tenant' is
+ // declared because virtual column inputs must
themselves be stored columns.
+ ImmutableList.of(
+ new ColumnSpec("tenant_lower", Columns.STRING,
null),
+ new ColumnSpec("__time", Columns.TIME_COLUMN,
null),
+ new ColumnSpec("tenant", Columns.STRING, null),
+ new ColumnSpec("dim1", Columns.STRING, null),
+ new ColumnSpec("cnt", Columns.LONG, null)
+ )
+ ),
+ MAPPER
+ )),
+
DatasourceTable.EffectiveMetadata.toEffectiveColumns(RowSignature.builder()
+ .add("tenant_lower", ColumnType.STRING)
+ .addTimeColumn()
+ .add("tenant", ColumnType.STRING)
+ .add("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build()),
+ false
+ )
)
);
@@ -996,6 +1070,117 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
.verify();
}
+ /**
+ * Inserting declared columns into a table with a base table layout
succeeds. The computed clustering column
+ * 'tenant_lower' is not (and cannot be) supplied by the query: the virtual
column populates it at ingest time from
+ * the supplied 'tenant' column.
+ */
+ @Test
+ public void testInsertIntoBaseTableCatalogTable()
+ {
+ ExternalDataSource externalDataSource = new ExternalDataSource(
+ new InlineInputSource("2022-12-26T12:34:56,extra,10,\"20\",foo\n"),
+ new CsvInputFormat(ImmutableList.of("a", "b", "c", "d", "e"), null,
false, false, 0, null),
+ RowSignature.builder()
+ .add("a", ColumnType.STRING)
+ .add("b", ColumnType.STRING)
+ .add("c", ColumnType.LONG)
+ .add("d", ColumnType.STRING)
+ .add("e", ColumnType.STRING)
+ .build()
+ );
+ final RowSignature signature = RowSignature.builder()
+ .add("__time", ColumnType.LONG)
+ .add("tenant", ColumnType.STRING)
+ .add("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithBaseTable") + "\n"
+
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS tenant,\n" +
+ " d AS dim1,\n" +
+ " c AS cnt\n" +
+ "FROM TABLE(inline(\n" +
+ " data => ARRAY['2022-12-26T12:34:56,extra,10,\"20\",foo'],\n" +
+ " format => 'csv'))\n" +
+ " (a VARCHAR, b VARCHAR, c BIGINT, d VARCHAR, e VARCHAR)\n" +
+ "PARTITIONED BY ALL TIME")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("tableWithBaseTable", signature)
+ .expectResources(dataSourceWrite("tableWithBaseTable"),
Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0",
"timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG)
+ )
+ .columns("v0", "b", "d", "c")
+ .columnTypes(ColumnType.LONG, ColumnType.STRING,
ColumnType.STRING, ColumnType.LONG)
+
.context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * A base table layout derives the physical segment schema from the declared
columns, so a column that is not
+ * declared cannot be stored: base table layouts require the sealed
property, which rejects undeclared columns.
+ */
+ @Test
+ public void testInsertAddNonDefinedColumnIntoBaseTableCatalogTable()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithBaseTable") + "\n"
+
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS tenant,\n" +
+ " d AS dim1,\n" +
+ " c AS cnt,\n" +
+ " e AS extra\n" +
+ "FROM TABLE(inline(\n" +
+ " data => ARRAY['2022-12-26T12:34:56,extra,10,\"20\",foo'],\n" +
+ " format => 'csv'))\n" +
+ " (a VARCHAR, b VARCHAR, c BIGINT, d VARCHAR, e VARCHAR)\n" +
+ "PARTITIONED BY ALL TIME")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectValidationError(
+ DruidException.class,
+ "Column [extra] is not defined in the target table
[druid.tableWithBaseTable] strict schema"
+ )
+ .verify();
+ }
+
+ /**
+ * Columns computed by the base table layout's ingest-time virtual columns
always store the computed value, so a
+ * query output with that name would never be read; writing to them directly
is rejected rather than silently
+ * discarding the supplied values.
+ */
+ @Test
+ public void testInsertComputedColumnIntoBaseTableCatalogTable()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithBaseTable") + "\n"
+
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " LOWER(b) AS tenant_lower,\n" +
+ " d AS dim1,\n" +
+ " c AS cnt\n" +
+ "FROM TABLE(inline(\n" +
+ " data => ARRAY['2022-12-26T12:34:56,extra,10,\"20\",foo'],\n" +
+ " format => 'csv'))\n" +
+ " (a VARCHAR, b VARCHAR, c BIGINT, d VARCHAR, e VARCHAR)\n" +
+ "PARTITIONED BY ALL TIME")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectValidationError(
+ DruidException.class,
+ "Column [tenant_lower] of target table [druid.tableWithBaseTable]
is computed by a virtual column at"
+ + " ingest time and cannot be written directly; supply the virtual
column's input columns instead"
+ )
+ .verify();
+ }
/**
* Inserting into a catalog table with a WITH source succeeds
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
