capistrant commented on code in PR #19830:
URL: https://github.com/apache/druid/pull/19830#discussion_r3844829117
##########
server/src/main/java/org/apache/druid/catalog/model/Columns.java:
##########
@@ -76,12 +77,25 @@ private Columns()
{
}
+ /**
+ * The Druid type of a column, which for {@code __time} is always {@link
ColumnType#LONG} whatever was declared.
Review Comment:
The end reads confusing to me. are you meaning "regardless of what was
declared" or something in that direction
##########
docs/development/extensions-core/catalog.md:
##########
@@ -43,6 +43,177 @@ allowing queries to be more concise, and simpler to write.
This also allows the
written into a defined column of the table is consistent with that columns
definition, minimizing errors where unexpected
data is written into a particular column of the table.
+### SQL DDL
Review Comment:
nice doc, ty for adding it up front. greatly helped set stage for review
##########
extensions-core/druid-catalog/src/test/java/org/apache/druid/server/http/catalog/EditorTest.java:
##########
@@ -486,4 +496,295 @@ public void testUpdateColumns() throws CatalogException
CatalogUtils.columnNames(revised.spec().columns())
);
}
+
+ @Test
+ public void testAddAndDropProjection() throws CatalogException
+ {
+ final String tableName = "projections";
+ final TableMetadata table = TableBuilder.datasource(tableName, "P1D")
+ .timeColumn()
+ .column("dim", "VARCHAR")
+ .column("met", "BIGINT")
+ .build();
+ catalog.tables().create(table);
+
+ final DatasourceProjectionMetadata daily = new
DatasourceProjectionMetadata(
+ AggregateProjectionSpec.builder("daily")
+ .groupingColumns(new
StringDimensionSchema("dim"))
+ .aggregators(new
LongSumAggregatorFactory("sum_met", "met"))
+ .build()
+ );
+
+ assertTrue(new TableEditor(catalog, table.id(), new AddProjection(daily,
false)).go() > 0);
+ assertEquals(List.of(daily), projectionsOf(tableName));
+
+ // Adding the same name again is an error, unless the caller said to leave
it alone.
+ assertThrows(
+ CatalogException.class,
+ () -> new TableEditor(catalog, table.id(), new AddProjection(daily,
false)).go()
+ );
+ assertEquals(0, new TableEditor(catalog, table.id(), new
AddProjection(daily, true)).go());
+ assertEquals(List.of(daily), projectionsOf(tableName));
+
+ // Dropping a projection that is not there is likewise an error unless
tolerated.
+ assertThrows(
+ CatalogException.class,
+ () -> new TableEditor(catalog, table.id(), new DropProjection("nope",
false)).go()
+ );
+ assertEquals(0, new TableEditor(catalog, table.id(), new
DropProjection("nope", true)).go());
+
+ assertTrue(new TableEditor(catalog, table.id(), new
DropProjection("daily", false)).go() > 0);
+ assertNull(
+ catalog.tables().read(TableId.datasource(tableName))
+
.spec().properties().get(DatasourceDefn.PROJECTIONS_KEYS_PROPERTY)
+ );
+ }
+
+ /**
+ * {@code AddColumns} means add: a column that already exists is an error
rather than a silent in-place update, which
+ * is what plain {@code UpdateColumns} would do.
+ */
+ @Test
+ public void testAddColumns() throws CatalogException
+ {
+ final String tableName = "addCols";
+ final TableMetadata table = TableBuilder.datasource(tableName, "P1D")
+ .timeColumn()
+ .column("dim", "VARCHAR")
+ .build();
+ catalog.tables().create(table);
+
+ final TableMetadata revised = doEdit(
+ tableName,
+ new TableEditRequest.AddColumns(Collections.singletonList(new
ColumnSpec("met", "BIGINT", null)))
+ );
+ assertEquals(
+ Arrays.asList(Columns.TIME_COLUMN, "dim", "met"),
+ CatalogUtils.columnNames(revised.spec().columns())
+ );
+
+ final CatalogException e = assertThrows(
+ CatalogException.class,
+ () -> doEdit(
+ tableName,
+ new TableEditRequest.AddColumns(Collections.singletonList(new
ColumnSpec("dim", "BIGINT", null)))
+ )
+ );
+ assertTrue(e.getMessage().contains("Column [dim] already exists"),
e.getMessage());
+ // The rejected add did not change the existing column's type.
+ assertEquals("VARCHAR", columnType(tableName, "dim"));
+ }
+
+ /**
+ * {@code AlterColumns} means alter: a column that does not exist is an
error rather than being appended, so a
+ * misspelled target cannot quietly create a column.
+ */
+ @Test
+ public void testAlterColumns() throws CatalogException
+ {
+ final String tableName = "alterCols";
+ final TableMetadata table = TableBuilder.datasource(tableName, "P1D")
+ .timeColumn()
+ .column("dim", "VARCHAR")
+ .build();
+ catalog.tables().create(table);
+
+ doEdit(
+ tableName,
+ new TableEditRequest.AlterColumns(Collections.singletonList(new
ColumnSpec("dim", "BIGINT", null)))
+ );
+ assertEquals("BIGINT", columnType(tableName, "dim"));
+
+ final CatalogException e = assertThrows(
+ CatalogException.class,
+ () -> doEdit(
+ tableName,
+ new TableEditRequest.AlterColumns(Collections.singletonList(new
ColumnSpec("typo", "BIGINT", null)))
+ )
+ );
+ assertTrue(e.getMessage().contains("Column [typo] does not exist"),
e.getMessage());
+ // The misspelled target was not created.
+ assertEquals(
+ Arrays.asList(Columns.TIME_COLUMN, "dim"),
+
CatalogUtils.columnNames(catalog.tables().read(table.id()).spec().columns())
+ );
+ }
+
+ private String columnType(String tableName, String columnName) throws
CatalogException
+ {
+ return
catalog.tables().read(TableId.datasource(tableName)).spec().columns().stream()
+ .filter(c -> columnName.equals(c.name()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No column [" +
columnName + "]"))
+ .dataType();
+ }
Review Comment:
nit: don't interleave with tests here. sorry to be repeating so many of
these 😅
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/catalog/CatalogDdlAndIngestTest.java:
##########
@@ -0,0 +1,465 @@
+/*
+ * 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.testing.embedded.catalog;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.catalog.model.ColumnSpec;
+import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
+import org.apache.druid.catalog.model.TableId;
+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.java.util.common.StringUtils;
+import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.server.metrics.LatchableEmitter;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.msq.EmbeddedMSQApis;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * End-to-end coverage of catalog DDL: define a table with SQL, ingest into
it, and query it. This is what proves the
+ * Broker-to-Coordinator write path and the cache refresh actually work
against a running cluster, rather than against
+ * a recording stub.
+ */
+public class CatalogDdlAndIngestTest extends CatalogTestBase
+{
+ private TestCatalogClient client;
+ private EmbeddedMSQApis msqApis;
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ // Catalog DDL is opt-in.
+ broker.addProperty("druid.sql.planner.enableCatalogDdl", "true");
+ return super.createCluster();
+ }
+
+ @BeforeAll
+ public void initializeClient()
+ {
+ client = new TestCatalogClient(cluster);
+ msqApis = new EmbeddedMSQApis(cluster, overlord);
+ }
+
+ @Test
+ public void testCreateTableThenIngestAndQuery()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql(
+ "CREATE TABLE \"%s\" (\n"
+ + " __time TIMESTAMP,\n"
+ + " varchar_col1 VARCHAR,\n"
+ + " double_col1 DOUBLE\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY varchar_col1",
+ tableName
+ );
+
+ // The spec reached the Coordinator, in the declared column order.
+ final TableMetadata table =
client.readTable(TableId.datasource(tableName));
+ assertEquals(
+ List.of("__time", "varchar_col1", "double_col1"),
+ columnNames(table)
+ );
+ assertEquals(
+ List.of("TIMESTAMP", "VARCHAR", "DOUBLE"),
+
table.spec().columns().stream().map(ColumnSpec::dataType).collect(Collectors.toList())
+ );
+ assertEquals("P1D",
table.spec().properties().get(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY));
+
+ // An INSERT that omits PARTITIONED BY picks it up from the catalog entry
the DDL just wrote, which only works
+ // if this Broker's catalog cache saw the write.
+ ingest(
+ "INSERT INTO \"%s\"\n"
+ + "SELECT\n"
+ + " TIME_PARSE(a) AS __time,\n"
+ + " b AS varchar_col1,\n"
+ + " c AS double_col1\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + "
'{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,foo,10\\n2022-12-26T12:34:56,bar,20\"}',\n"
+ + "
'{\"type\":\"csv\",\"findColumnsFromHeader\":false,\"columns\":[\"a\",\"b\",\"c\"]}'\n"
+ + " )\n"
+ + ") EXTEND (a VARCHAR, b VARCHAR, c BIGINT)\n",
+ tableName
+ );
+
+ // The BIGINT source column is coerced to the DOUBLE declared by the DDL,
and rows come back in clustered order.
+ cluster.callApi().verifySqlQuery(
+ "SELECT * FROM %s",
+ tableName,
+ "2022-12-26T12:34:56.000Z,bar,20.0\n"
+ + "2022-12-26T12:34:56.000Z,foo,10.0"
+ );
+ }
+
+ @Test
+ public void testAlterTableAddColumnThenIngest()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR) PARTITIONED BY DAY", tableName);
+ cluster.callApi().runSql("ALTER TABLE \"%s\" ADD COLUMN b BIGINT",
tableName);
+
+ final TableMetadata table =
client.readTable(TableId.datasource(tableName));
+ assertEquals(List.of("__time", "a", "b"), columnNames(table));
+
+ ingest(
+ "INSERT INTO \"%s\"\n"
+ + "SELECT TIME_PARSE(x) AS __time, y AS a, z AS b\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + "
'{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,hello,7\"}',\n"
+ + "
'{\"type\":\"csv\",\"findColumnsFromHeader\":false,\"columns\":[\"x\",\"y\",\"z\"]}'\n"
+ + " )\n"
+ + ") EXTEND (x VARCHAR, y VARCHAR, z BIGINT)\n",
+ tableName
+ );
+
+ cluster.callApi().verifySqlQuery("SELECT * FROM %s", tableName,
"2022-12-26T12:34:56.000Z,hello,7");
+ }
+
+ @Test
+ public void testAlterTableColumnAndProperties()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR, b VARCHAR)", tableName);
+
+ cluster.callApi().runSql("ALTER TABLE \"%s\" ALTER COLUMN b SET DATA TYPE
BIGINT", tableName);
+ cluster.callApi().runSql("ALTER TABLE \"%s\" DROP COLUMN a", tableName);
+ cluster.callApi().runSql(
+ "ALTER TABLE \"%s\" SET PROPERTIES (segmentGranularity = 'P1D', sealed
= TRUE)",
+ tableName
+ );
+
+ final TableMetadata table =
client.readTable(TableId.datasource(tableName));
+ assertEquals(List.of("__time", "b"), columnNames(table));
+ assertEquals("BIGINT", table.spec().columns().get(1).dataType());
+ assertEquals("P1D",
table.spec().properties().get(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY));
+ assertEquals(true,
table.spec().properties().get(DatasourceDefn.SEALED_PROPERTY));
+
+ // A null value removes a property.
+ cluster.callApi().runSql("ALTER TABLE \"%s\" SET PROPERTIES (sealed =
NULL)", tableName);
+ assertNull(
+
client.readTable(TableId.datasource(tableName)).spec().properties().get(DatasourceDefn.SEALED_PROPERTY)
+ );
+ }
+
+ @Test
+ public void testCreateOrReplaceAndIfNotExists()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR) CLUSTERED BY a", tableName);
+
+ // IF NOT EXISTS leaves the original definition alone.
+ cluster.callApi().runSql("CREATE TABLE IF NOT EXISTS \"%s\" (__time
TIMESTAMP, zzz VARCHAR)", tableName);
+ assertEquals(List.of("__time", "a"),
columnNames(client.readTable(TableId.datasource(tableName))));
+
+ // OR REPLACE swaps the whole spec, including dropping the clustering the
first statement set.
+ cluster.callApi().runSql("CREATE OR REPLACE TABLE \"%s\" (__time
TIMESTAMP, b BIGINT)", tableName);
+ final TableMetadata replaced =
client.readTable(TableId.datasource(tableName));
+ assertEquals(List.of("__time", "b"), columnNames(replaced));
+
assertNull(replaced.spec().properties().get(DatasourceDefn.CLUSTER_KEYS_PROPERTY));
+ }
+
+ @Test
+ public void testCreateTableClusterKeys()
+ {
+ final String tableName = dataSource;
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR, b BIGINT) CLUSTERED BY a, b", tableName);
+
+ // Catalog properties round trip as untyped JSON, so decode before
asserting.
+ final TableMetadata table =
client.readTable(TableId.datasource(tableName));
+ final List<ClusterKeySpec> keys = TestHelper.JSON_MAPPER.convertValue(
+ table.spec().properties().get(DatasourceDefn.CLUSTER_KEYS_PROPERTY),
+ ClusterKeySpec.CLUSTER_KEY_LIST_TYPE_REF
+ );
+ assertEquals(List.of("a", "b"),
keys.stream().map(ClusterKeySpec::expr).collect(Collectors.toList()));
+ assertTrue(keys.stream().noneMatch(ClusterKeySpec::desc));
+ }
+
+ /**
+ * A rejection from the Coordinator's own validation must reach the SQL user
as the Coordinator worded it, not as
+ * a generic remote-call failure. Segment granularity is validated only on
the Coordinator, so it exercises the
+ * whole round trip.
+ */
+ @Test
+ public void testCoordinatorValidationErrorSurfacesToSqlUser()
+ {
+ final String tableName = dataSource;
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR)", tableName);
+
+ final Exception e = assertThrows(
+ Exception.class,
+ () -> cluster.callApi().runSql(
+ "ALTER TABLE \"%s\" SET PROPERTIES (segmentGranularity =
'not_a_granularity')",
+ tableName
+ )
+ );
+ assertTrue(e.getMessage().contains("granularity"), e.getMessage());
+ }
+
+ @Test
+ public void testCreateTableAlreadyExistsFails()
+ {
+ final String tableName = dataSource;
+ cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP, a
VARCHAR)", tableName);
+
+ final Exception e = assertThrows(
+ Exception.class,
+ () -> cluster.callApi().runSql("CREATE TABLE \"%s\" (__time TIMESTAMP,
a VARCHAR)", tableName)
+ );
+ assertTrue(e.getMessage().contains("duplicate table"), e.getMessage());
+ }
+
+ /**
+ * A projection defined in SQL must actually be used at query time, which is
the whole point of storing one: the
+ * specification the translator produces has to match what the planner
generates for the equivalent query.
+ */
+ @Test
+ public void testCreateTableWithProjectionThenQuery()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql(
+ "CREATE TABLE \"%s\" (\n"
+ + " __time TIMESTAMP,\n"
+ + " varchar_col1 VARCHAR,\n"
+ + " bigint_col1 BIGINT,\n"
+ + " PROJECTION by_varchar AS (\n"
+ + " SELECT varchar_col1, SUM(bigint_col1) AS sum_bigint_col1\n"
+ + " GROUP BY varchar_col1\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY",
+ tableName
+ );
+
+ ingest(
+ "INSERT INTO \"%s\"\n"
+ + "SELECT TIME_PARSE(a) AS __time, b AS varchar_col1, c AS
bigint_col1\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + "
'{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,foo,10\\n2022-12-26T12:34:56,foo,9"
+ + "\\n2022-12-26T12:34:56,bar,8\"}',\n"
+ + "
'{\"type\":\"csv\",\"findColumnsFromHeader\":false,\"columns\":[\"a\",\"b\",\"c\"]}'\n"
+ + " )\n"
+ + ") EXTEND (a VARCHAR, b VARCHAR, c BIGINT)\n",
+ tableName
+ );
+
+ final LatchableEmitter emitter = historical.latchableEmitter();
+ emitter.flush();
+
+ cluster.callApi().verifySqlQuery(
+ "SELECT varchar_col1, SUM(bigint_col1) FROM %s GROUP BY 1 ORDER BY 1",
+ tableName,
+ "bar,8\nfoo,19"
+ );
+
+ // The segment-scan metrics name the projection that served the query.
+ emitter.waitForEvent(
+ event ->
event.hasMetricName("query/segment/time").hasDimension("projection",
"by_varchar")
+ );
+ }
+
+ @Test
+ public void testAlterTableAddAndDropProjection()
+ {
+ final String tableName = dataSource;
+
+ cluster.callApi().runSql(
+ "CREATE TABLE \"%s\" (__time TIMESTAMP, a VARCHAR, b BIGINT)
PARTITIONED BY DAY",
+ tableName
+ );
+ cluster.callApi().runSql(
+ "ALTER TABLE \"%s\" ADD PROJECTION p AS (SELECT a, SUM(b) AS sum_b
GROUP BY a)",
+ tableName
+ );
+ assertEquals(1, projectionsOf(tableName).size());
+ assertEquals("p", projectionsOf(tableName).get(0).getSpec().getName());
+
+ // Adding it again is an error, but IF NOT EXISTS tolerates it.
+ assertThrows(
+ Exception.class,
+ () -> cluster.callApi().runSql(
+ "ALTER TABLE \"%s\" ADD PROJECTION p AS (SELECT a, SUM(b) AS sum_b
GROUP BY a)",
+ tableName
+ )
+ );
+ cluster.callApi().runSql(
+ "ALTER TABLE \"%s\" ADD IF NOT EXISTS PROJECTION p AS (SELECT a,
SUM(b) AS sum_b GROUP BY a)",
+ tableName
+ );
+ assertEquals(1, projectionsOf(tableName).size());
+
+ cluster.callApi().runSql("ALTER TABLE \"%s\" DROP PROJECTION p",
tableName);
+ assertNull(
+ client.readTable(TableId.datasource(tableName))
+
.spec().properties().get(DatasourceDefn.PROJECTIONS_KEYS_PROPERTY)
+ );
+ cluster.callApi().runSql("ALTER TABLE \"%s\" DROP PROJECTION IF EXISTS p",
tableName);
+ }
+
+ private List<DatasourceProjectionMetadata> projectionsOf(String tableName)
+ {
+ return TestHelper.JSON_MAPPER.convertValue(
+ client.readTable(TableId.datasource(tableName))
+
.spec().properties().get(DatasourceDefn.PROJECTIONS_KEYS_PROPERTY),
+ new TypeReference<List<DatasourceProjectionMetadata>>() {}
+ );
+ }
Review Comment:
nit: another interleaved private that could be moved out of test area
##########
sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogDdlTest.java:
##########
@@ -0,0 +1,954 @@
+/*
+ * 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.sql.calcite;
+
+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.DatasourceBaseTableMetadata;
+import org.apache.druid.catalog.model.DatasourceProjectionMetadata;
+import org.apache.druid.catalog.model.TableId;
+import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.TableSpec;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.catalog.model.table.DatasourceDefn;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthConfig;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+import org.apache.druid.server.security.ResourceType;
+import org.apache.druid.sql.DirectStatement;
+import org.apache.druid.sql.SqlQueryPlus;
+import
org.apache.druid.sql.calcite.CalciteCatalogDdlTest.CatalogDdlComponentSupplier;
+import org.apache.druid.sql.calcite.planner.CatalogTableWriter;
+import org.apache.druid.sql.calcite.planner.PlannerConfig;
+import org.apache.druid.sql.calcite.util.CalciteTests;
+import
org.apache.druid.sql.calcite.util.SqlTestFramework.StandardComponentSupplier;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests that catalog DDL statements plan into the catalog operations they
claim to, using a writer that records
+ * calls instead of contacting a Coordinator.
+ */
[email protected](CatalogDdlComponentSupplier.class)
+public class CalciteCatalogDdlTest extends BaseCalciteQueryTest
+{
+ private static final RecordingCatalogTableWriter WRITER = new
RecordingCatalogTableWriter();
+
+ public static class CatalogDdlComponentSupplier extends
StandardComponentSupplier
+ {
+ public CatalogDdlComponentSupplier(TempDirProducer tempFolderProducer)
+ {
+ super(tempFolderProducer);
+ }
+
+ @Override
+ public CatalogTableWriter createCatalogTableWriter()
+ {
+ return WRITER;
+ }
+ }
+
+ @BeforeEach
+ public void resetWriter()
+ {
+ WRITER.reset();
+ }
+
+ @Test
+ public void testCreateTable()
+ {
+ execute("CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT)");
+
+ assertEquals(1, WRITER.calls.size());
+ final RecordingCatalogTableWriter.Call call = WRITER.calls.get(0);
+ assertEquals("createTable", call.operation);
+ assertEquals(TableId.datasource("tbl"), call.tableId);
+ assertEquals(DatasourceDefn.TABLE_TYPE, call.spec.type());
+ assertEquals(ImmutableMap.of(), call.spec.properties());
+ assertEquals(
+ ImmutableList.of(
+ new ColumnSpec("__time", "TIMESTAMP", null),
+ new ColumnSpec("page", "VARCHAR", null),
+ new ColumnSpec("cnt", "BIGINT", null)
+ ),
+ call.spec.columns()
+ );
+ assertFalse(call.ifNotExists);
+ assertFalse(call.replace);
+ }
+
+ @Test
+ public void testCreateTableWithPartitioningAndClustering()
+ {
+ execute("CREATE TABLE tbl (page VARCHAR, cnt BIGINT) PARTITIONED BY DAY
CLUSTERED BY page, cnt");
+
+ final TableSpec spec = WRITER.calls.get(0).spec;
+ assertEquals("P1D",
spec.properties().get(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY));
+ assertEquals(
+ ImmutableList.of(new ClusterKeySpec("page", false), new
ClusterKeySpec("cnt", false)),
+ spec.properties().get(DatasourceDefn.CLUSTER_KEYS_PROPERTY)
+ );
+ }
+
+ @Test
+ public void testCreateTablePartitionedByAll()
+ {
+ execute("CREATE TABLE tbl (page VARCHAR) PARTITIONED BY ALL TIME");
+ assertEquals("ALL",
WRITER.calls.get(0).spec.properties().get(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY));
+ }
+
+ @Test
+ public void testCreateTableTypeCanonicalization()
+ {
+ execute(
+ "CREATE TABLE tbl (a CHAR, b INTEGER, c REAL, d DOUBLE, e VARCHAR
ARRAY, f TYPE('complex<json>'))"
+ );
+ assertEquals(
+ ImmutableList.of(
+ new ColumnSpec("a", "VARCHAR", null),
+ new ColumnSpec("b", "BIGINT", null),
+ new ColumnSpec("c", "FLOAT", null),
+ new ColumnSpec("d", "DOUBLE", null),
+ new ColumnSpec("e", "VARCHAR ARRAY", null),
+ new ColumnSpec("f", "COMPLEX<json>", null)
+ ),
+ WRITER.calls.get(0).spec.columns()
+ );
+ }
+
+ @Test
+ public void testCreateTableFlags()
+ {
+ execute("CREATE OR REPLACE TABLE tbl (a VARCHAR)");
+ assertTrue(WRITER.calls.get(0).replace);
+
+ WRITER.reset();
+ execute("CREATE TABLE IF NOT EXISTS tbl (a VARCHAR)");
+ assertTrue(WRITER.calls.get(0).ifNotExists);
+ }
+
+ @Test
+ public void testCreateTableInDruidSchema()
+ {
+ execute("CREATE TABLE druid.tbl (a VARCHAR)");
+ assertEquals(TableId.datasource("tbl"), WRITER.calls.get(0).tableId);
+ }
+
+ @Test
+ public void testResourceActionIsDatasourceWrite()
+ {
+ final DirectStatement stmt = statement("CREATE TABLE tbl (a VARCHAR)");
+ stmt.execute();
+ assertEquals(
+ Collections.singleton(new ResourceAction(new Resource("tbl",
ResourceType.DATASOURCE), Action.WRITE)),
+ stmt.resources()
+ );
+ }
+
+ @Test
+ public void testDdlReturnsNoRows()
+ {
+ final DirectStatement stmt = statement("CREATE TABLE tbl (a VARCHAR)");
+ final List<Object[]> results = stmt.execute().getResults().toList();
+ assertEquals(ImmutableList.of(), results);
+ }
+
+ @Test
+ public void testAlterTableAddColumn()
+ {
+ // ADD and ALTER differ only in which existence outcome is an error, and
that rule is enforced inside the
+ // Coordinator's update transaction, so all the statement does is pick the
verb. EditorTest covers the rules.
+ execute("ALTER TABLE tbl ADD COLUMN b BIGINT");
+
+ final RecordingCatalogTableWriter.Call call =
WRITER.lastCall("addColumns");
+ assertEquals(ImmutableList.of(new ColumnSpec("b", "BIGINT", null)),
call.columns);
+ }
+
+ @Test
+ public void testAlterTableDropColumn()
+ {
+ execute("ALTER TABLE tbl DROP COLUMN gone");
+ assertEquals(ImmutableList.of("gone"),
WRITER.lastCall("dropColumns").droppedColumns);
+ }
+
+ @Test
+ public void testAlterTableAlterColumn()
+ {
+ execute("ALTER TABLE tbl ALTER COLUMN cnt SET DATA TYPE DOUBLE");
+ assertEquals(
+ ImmutableList.of(new ColumnSpec("cnt", "DOUBLE", null)),
+ WRITER.lastCall("alterColumns").columns
+ );
+ }
+
+ @Test
+ public void testAlterTableSetProperties()
+ {
+ execute("ALTER TABLE tbl SET PROPERTIES (targetSegmentRows = 3000000,
sealed = TRUE, description = 'hi')");
+
+ final Map<String, Object> properties =
WRITER.lastCall("updateProperties").properties;
+ assertEquals(3000000L, properties.get("targetSegmentRows"));
+ assertEquals(true, properties.get("sealed"));
+ assertEquals("hi", properties.get("description"));
+ }
+
+ @Test
+ public void testAlterTableSetPropertyToNullRemovesIt()
+ {
+ execute("ALTER TABLE tbl SET PROPERTIES (description = NULL)");
+ final Map<String, Object> properties =
WRITER.lastCall("updateProperties").properties;
+ assertTrue(properties.containsKey("description"));
+ assertNull(properties.get("description"));
+ }
+
+ @Test
+ public void testCreateTableRejectsDuplicateColumn()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, a BIGINT)")
+ );
+ assertTrue(e.getMessage().contains("Column [a] is declared more than
once"));
+ }
+
+ @Test
+ public void testCreateTableRejectsUnsupportedType()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a TYPE('NOT_A_TYPE'))")
+ );
+ assertTrue(e.getMessage().contains("unsupported type"));
+ }
+
+ /**
+ * Any spelling that resolves to a LONG is accepted for the time column, and
is stored as written.
+ */
+ @Test
+ public void testCreateTableTimeColumnSpellings()
+ {
+ execute("CREATE TABLE tbl (__time BIGINT)");
+ assertEquals(ImmutableList.of(new ColumnSpec("__time", "BIGINT", null)),
WRITER.calls.get(0).spec.columns());
+
+ WRITER.reset();
+ execute("CREATE TABLE tbl (__time TYPE('LONG'))");
+ assertEquals(ImmutableList.of(new ColumnSpec("__time", "LONG", null)),
WRITER.calls.get(0).spec.columns());
+ }
+
+ @Test
+ public void testCreateTableRejectsNonLongTimeColumn()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (__time VARCHAR)")
+ );
+ assertTrue(e.getMessage().contains("Column [__time] must have type"));
+ }
+
+ @Test
+ public void testCreateTableRejectsNonDruidSchema()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE lookup.tbl (a VARCHAR)")
+ );
+ assertTrue(e.getMessage().contains("is not a Druid datasource"));
+ }
+
+ @Test
+ public void testCreateTableRejectsBothReplaceAndIfNotExists()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE OR REPLACE TABLE IF NOT EXISTS tbl (a VARCHAR)")
+ );
+ assertTrue(e.getMessage().contains("Cannot specify both OR REPLACE and IF
NOT EXISTS"));
+ }
+
+ @Test
+ public void testCreateTableRejectsClusteringExpression()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR) CLUSTERED BY a DESC")
+ );
+ assertTrue(e.getMessage().contains("must be a column name"));
+ }
+
+ /**
+ * The feature is off unless an operator turns it on, so that upgrading a
cluster does not silently widen what a
+ * datasource WRITE permission allows.
+ */
+ @Test
+ public void testDdlIsDisabledByDefault()
+ {
+ final DirectStatement stmt =
getSqlStatementFactory(PlannerConfig.builder().build(), new AuthConfig())
+ .directStatement(
+ SqlQueryPlus.builder("CREATE TABLE tbl (a VARCHAR)")
+ .auth(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .build()
+ );
+ final DruidException e = assertThrows(DruidException.class, stmt::execute);
+ assertTrue(e.getMessage().contains("druid.sql.planner.enableCatalogDdl"),
e.getMessage());
+ assertEquals(ImmutableList.of(), WRITER.calls);
+ }
+
+ /**
+ * The stored specification must be the one the planner would produce for
the equivalent query, since that is what
+ * makes a projection match at query time. Pinned as JSON so a change in
planner output is visible here.
+ */
+ @Test
+ public void testCreateTableWithProjection() throws Exception
+ {
+ execute(
+ "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION daily AS (SELECT TIME_FLOOR(__time, 'P1D'), page,
SUM(cnt) AS total GROUP BY 1, 2))"
+ );
+
+ assertEquals(
+ "[{\"spec\":{\"type\":\"aggregate\",\"name\":\"daily\","
+ + "\"virtualColumns\":[{\"type\":\"expression\",\"name\":\"v0\","
+ +
"\"expression\":\"timestamp_floor(\\\"__time\\\",'P1D',null,'UTC')\",\"outputType\":\"LONG\"}],"
+ +
"\"groupingColumns\":[{\"type\":\"long\",\"name\":\"v0\",\"multiValueHandling\":\"SORTED_ARRAY\","
+ +
"\"createBitmapIndex\":false},{\"type\":\"string\",\"name\":\"page\","
+ +
"\"multiValueHandling\":\"SORTED_ARRAY\",\"createBitmapIndex\":true}],"
+ +
"\"aggregators\":[{\"type\":\"longSum\",\"name\":\"total\",\"fieldName\":\"cnt\"}],"
+ + "\"ordering\":[{\"columnName\":\"v0\",\"order\":\"ascending\"},"
+ + "{\"columnName\":\"page\",\"order\":\"ascending\"}]}}]",
+ projectionsJson()
+ );
+ }
+
+ /**
+ * A projection body is planned under the statement's own context, so a SET
clause that changes how the equivalent
+ * query would plan changes the stored definition the same way. Here the
session time zone reaches the TIME_FLOOR.
+ */
+ @Test
+ public void testProjectionBodyHonorsStatementContext() throws Exception
+ {
+ execute(
+ "SET sqlTimeZone = 'America/Los_Angeles';\n"
+ + "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION daily AS (SELECT TIME_FLOOR(__time, 'P1D'), page,
SUM(cnt) AS total GROUP BY 1, 2))"
+ );
+
+ assertTrue(
+
projectionsJson().contains("timestamp_floor(\\\"__time\\\",'P1D',null,'America/Los_Angeles')"),
+ projectionsJson()
+ );
+ }
+
+ /**
+ * The overrides the lift depends on are applied on top of the statement's
context, so a SET clause cannot put the
+ * planner into a shape the lift does not understand.
+ */
+ @Test
+ public void testProjectionBodyContextCannotOverrideDeterministicOverrides()
throws Exception
+ {
+ execute(
+ "SET sqlUseGranularity = TRUE;\n"
+ + "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION daily AS (SELECT TIME_FLOOR(__time, 'P1D'), page,
SUM(cnt) AS total GROUP BY 1, 2))"
+ );
+
+ // Still lifted as an ordinary grouping column rather than a query
granularity, exactly as without the SET.
+ assertTrue(
+
projectionsJson().contains("timestamp_floor(\\\"__time\\\",'P1D',null,'UTC')"),
+ projectionsJson()
+ );
+ }
+
+ /**
+ * A projection defined with TIME_FLOOR must carry a granularity the segment
layer can recover, which is how the
+ * projection gets matched to time-grouped queries.
+ */
+ @Test
+ public void testProjectionGranularityIsRecoverable()
+ {
+ execute(
+ "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION hourly AS (SELECT TIME_FLOOR(__time, 'PT1H'), page,
SUM(cnt) AS total GROUP BY 1, 2))"
+ );
+
+ final AggregateProjectionSpec spec = projection(0).getSpec();
+ final String timeColumn = spec.toMetadataSchema().getTimeColumnName();
+ assertEquals("v0", timeColumn);
+ assertEquals(
+ Granularities.HOUR,
+
Granularities.fromVirtualColumn(spec.getVirtualColumns().getVirtualColumn(timeColumn))
+ );
+ }
+
+ @Test
+ public void testProjectionWithFilter()
+ {
+ execute(
+ "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION filtered AS (SELECT page, SUM(cnt) AS total WHERE page
<> 'skip' GROUP BY page))"
+ );
+ assertEquals("!page = skip",
projection(0).getSpec().getFilter().toString());
+ }
+
+ /**
+ * A time bound written in the body is moved into the query's intervals
during planning, and has to be put back:
+ * a projection stores a filter, not an interval.
+ */
+ @Test
+ public void testProjectionWithTimeFilter()
+ {
+ execute(
+ "CREATE TABLE tbl (__time TIMESTAMP, page VARCHAR, cnt BIGINT,"
+ + " PROJECTION recent AS (SELECT page, SUM(cnt) AS total"
+ + " WHERE __time >= TIMESTAMP '2020-01-01 00:00:00' GROUP BY page))"
+ );
+ assertNotNull(projection(0).getSpec().getFilter(), "time filter must
survive as a filter");
+
assertTrue(projection(0).getSpec().getFilter().getRequiredColumns().contains("__time"));
+ }
+
+ @Test
+ public void testProjectionSelectDistinct()
+ {
+ execute("CREATE TABLE tbl (a VARCHAR, PROJECTION d AS (SELECT DISTINCT
a))");
+ final AggregateProjectionSpec spec = projection(0).getSpec();
+ assertEquals(1, spec.getGroupingColumns().size());
+ assertEquals("a", spec.getGroupingColumns().get(0).getName());
+ assertEquals(0, spec.getAggregators().length);
+ }
+
+ @Test
+ public void testMultipleProjections()
+ {
+ execute(
+ "CREATE TABLE tbl (a VARCHAR, b BIGINT,"
+ + " PROJECTION p1 AS (SELECT a, SUM(b) AS s GROUP BY a),"
+ + " PROJECTION p2 AS (SELECT b, COUNT(*) AS c GROUP BY b))"
+ );
+ assertEquals(List.of("p1", "p2"),
List.of(projection(0).getSpec().getName(), projection(1).getSpec().getName()));
+ }
+
+ @Test
+ public void testAlterTableAddProjection()
+ {
+ WRITER.existing.put(TableId.datasource("tbl"), tableWithColumns("a"));
+ execute("ALTER TABLE tbl ADD PROJECTION p AS (SELECT a, COUNT(*) AS c
GROUP BY a)");
+
+ final RecordingCatalogTableWriter.Call call =
WRITER.lastCall("addProjection");
+ assertEquals("p", call.projection.getSpec().getName());
+ assertFalse(call.ifNotExists);
+ }
+
+ @Test
+ public void testAlterTableAddProjectionIfNotExists()
+ {
+ WRITER.existing.put(TableId.datasource("tbl"), tableWithColumns("a"));
+ execute("ALTER TABLE tbl ADD IF NOT EXISTS PROJECTION p AS (SELECT a GROUP
BY a)");
+ assertTrue(WRITER.lastCall("addProjection").ifNotExists);
+ }
+
+ @Test
+ public void testAlterTableDropProjection()
+ {
+ execute("ALTER TABLE tbl DROP PROJECTION p");
+ final RecordingCatalogTableWriter.Call call =
WRITER.lastCall("dropProjection");
+ assertEquals("p", call.projectionName);
+ assertFalse(call.ifExists);
+
+ WRITER.reset();
+ execute("ALTER TABLE tbl DROP PROJECTION IF EXISTS p");
+ assertTrue(WRITER.lastCall("dropProjection").ifExists);
+ }
+
+ @Test
+ public void testProjectionRejectsUnaliasedAggregate()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, b BIGINT, PROJECTION p AS
(SELECT a, SUM(b) GROUP BY a))")
+ );
+ assertTrue(e.getMessage().contains("no name"), e.getMessage());
+ }
+
+ @Test
+ public void testProjectionRejectsPostAggregation()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, b BIGINT, PROJECTION p AS
(SELECT a, AVG(b) AS m GROUP BY a))")
+ );
+ assertTrue(e.getMessage().contains("expression over aggregates"),
e.getMessage());
+ }
+
+ @Test
+ public void testProjectionRejectsUnknownColumn()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, PROJECTION p AS (SELECT
nope GROUP BY nope))")
+ );
+ assertTrue(e.getMessage().contains("nope"), e.getMessage());
+ }
+
+ @Test
+ public void testProjectionRejectsNonAggregatingBody()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, PROJECTION p AS (SELECT
a))")
+ );
+ assertTrue(e.getMessage().contains("does not aggregate"), e.getMessage());
+ }
+
+ /**
+ * {@code __base} names the table's own layout and is handled separately;
every other name beginning with the
+ * reserved prefix stays unavailable.
+ */
+ @Test
+ public void testProjectionRejectsOtherReservedNames()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute("CREATE TABLE tbl (a VARCHAR, PROJECTION __other AS
(SELECT a GROUP BY a))")
+ );
+ assertTrue(e.getMessage().contains("reserved name"), e.getMessage());
+ }
+
+ @Test
+ public void testProjectionRejectsDuplicateName()
+ {
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> execute(
+ "CREATE TABLE tbl (a VARCHAR, PROJECTION p AS (SELECT a GROUP BY
a),"
+ + " PROJECTION p AS (SELECT a GROUP BY a))"
+ )
+ );
+ assertTrue(e.getMessage().contains("declared more than once"),
e.getMessage());
+ }
+
+ @SuppressWarnings("unchecked")
+ private DatasourceProjectionMetadata projection(int index)
+ {
+ return ((List<DatasourceProjectionMetadata>)
WRITER.calls.get(0).spec.properties().get("projections")).get(index);
+ }
+
+ private String projectionsJson() throws Exception
+ {
+ return queryFramework().queryJsonMapper()
+
.writeValueAsString(WRITER.calls.get(0).spec.properties().get("projections"));
+ }
Review Comment:
nit: don't interleave these in the tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
