From dd5dc500cebb2d038e4c5ef23adb1e1acae4c377 Mon Sep 17 00:00:00 2001 From: zachjsh Date: Fri, 17 May 2024 11:49:09 -0400 Subject: [PATCH] Catalog integration tests (#16424) * * add new catalog IT with failure to ensure that it is run in CI * * actually add failing test referred to and fix checkstyle * * add some tests * * fix checkstyle * * add test descriptions * * add more tests --- .../catalog/ITCatalogIngestAndQueryTest.java | 448 ++++++++++++++++++ .../catalog/ITCatalogIngestErrorTest.java | 251 ++++++++++ .../catalog/ITCatalogInsertAndQueryTest.java | 42 ++ .../catalog/ITCatalogReplaceAndQueryTest.java | 42 ++ .../testsEx/catalog/ITCatalogRestTest.java | 13 + .../resources/catalog/clustering_select.sql | 23 + .../resources/catalog/implicitCast_select.sql | 23 + .../catalog/multiClustering_select.sql | 35 ++ .../testing/utils/MsqTestQueryHelper.java | 19 +- 9 files changed, 894 insertions(+), 2 deletions(-) create mode 100644 integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestAndQueryTest.java create mode 100644 integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestErrorTest.java create mode 100644 integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogInsertAndQueryTest.java create mode 100644 integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogReplaceAndQueryTest.java create mode 100644 integration-tests-ex/cases/src/test/resources/catalog/clustering_select.sql create mode 100644 integration-tests-ex/cases/src/test/resources/catalog/implicitCast_select.sql create mode 100644 integration-tests-ex/cases/src/test/resources/catalog/multiClustering_select.sql diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestAndQueryTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestAndQueryTest.java new file mode 100644 index 00000000000..eff9e2e197b --- /dev/null +++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestAndQueryTest.java @@ -0,0 +1,448 @@ +/* + * 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.testsEx.catalog; + +import com.google.common.collect.ImmutableList; +import com.google.inject.Inject; +import org.apache.druid.catalog.model.Columns; +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.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.msq.sql.SqlTaskStatus; +import org.apache.druid.testing.utils.DataLoaderHelper; +import org.apache.druid.testing.utils.MsqTestQueryHelper; +import org.apache.druid.testsEx.cluster.CatalogClient; +import org.apache.druid.testsEx.cluster.DruidClusterClient; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Tests that expect succesfully ingestng data into catalog defined tables and querying the data + * gives expected results. + */ +public abstract class ITCatalogIngestAndQueryTest +{ + public static final Logger LOG = new Logger(ITCatalogIngestAndQueryTest.class); + + @Inject + private MsqTestQueryHelper msqHelper; + @Inject + private DataLoaderHelper dataLoaderHelper; + @Inject + private DruidClusterClient clusterClient; + private CatalogClient client; + + private final String operationName; + private final String dmlPrefixPattern; + + public ITCatalogIngestAndQueryTest() + { + this.operationName = getOperationName(); + this.dmlPrefixPattern = getDmlPrefixPattern(); + } + + public abstract String getOperationName(); + public abstract String getDmlPrefixPattern(); + + @Before + public void initializeClient() + { + client = new CatalogClient(clusterClient); + } + + /** + * Create table with columns: + *

+ * __time LONG + * double_col1 DOUBLE + *

+ * And insert the following data: + *

+ * __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 + *

+ * When querying the table with query: 'SELECT * from ##tableName', 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. + *

+ * __time, bigint_col1 + * 2022-12-26T12:34:56,8.0 + * 2022-12-26T12:34:56,8.0 + * 2022-12-26T12:34:56,9.0 + * 2022-12-26T12:34:56,10.0 + * + */ + @Test + public void testInsertImplicitCast() throws Exception + { + String queryFile = "/catalog/implicitCast_select.sql"; + String tableName = "testImplicitCast" + operationName; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("double_col1", "DOUBLE") + .build(); + + client.createTable(table, true); + LOG.info("table created:\n%s", client.readTable(table.id())); + String queryInline = + StringUtils.format(dmlPrefixPattern, tableName) + "\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " c AS double_col1\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 + LOG.info("Running query:\n%s", queryInline); + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskSuccesfully(queryInline); + + if (sqlTaskStatus.getState().isFailure()) { + Assert.fail(StringUtils.format( + "Unable to start the task successfully.\nPossible exception: %s", + sqlTaskStatus.getError() + )); + } + + msqHelper.pollTaskIdForSuccess(sqlTaskStatus.getTaskId()); + dataLoaderHelper.waitUntilDatasourceIsReady(tableName); + + msqHelper.testQueriesFromFile(queryFile, tableName); + } + + /** + * Create table with columns: + *

+ * __time LONG + * double_col1 DOUBLE + *

+ * and clustering columns defined in catalog as + *

+ * bigInt_col1 + *

+ * And insert the following data: + *

+ * __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 + *

+ * When querying the table with query: 'SELECT * from ##tableName', because of the clustering + * defined on the table, the data should be reordered to: + *

+ * __time, bigint_col1 + * 2022-12-26T12:34:56,8 + * 2022-12-26T12:34:56,8 + * 2022-12-26T12:34:56,9 + * 2022-12-26T12:34:56,10 + * + */ + @Test + public void testInsertWithClusteringFromCatalog() throws Exception + { + String queryFile = "/catalog/clustering_select.sql"; + String tableName = "testWithClusteringFromCatalog" + operationName; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("bigint_col1", "BIGINT") + .property( + DatasourceDefn.CLUSTER_KEYS_PROPERTY, + ImmutableList.of(new ClusterKeySpec("bigint_col1", false)) + ) + .build(); + + client.createTable(table, true); + LOG.info("table created:\n%s", client.readTable(table.id())); + String queryInline = + StringUtils.format(dmlPrefixPattern, tableName) + "\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " c AS bigint_col1\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 + LOG.info("Running query:\n%s", queryInline); + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskSuccesfully(queryInline); + + if (sqlTaskStatus.getState().isFailure()) { + Assert.fail(StringUtils.format( + "Unable to start the task successfully.\nPossible exception: %s", + sqlTaskStatus.getError() + )); + } + + msqHelper.pollTaskIdForSuccess(sqlTaskStatus.getTaskId()); + dataLoaderHelper.waitUntilDatasourceIsReady(tableName); + + msqHelper.testQueriesFromFile(queryFile, tableName); + } + + /** + * Create table with columns: + *

+ * __time LONG + * double_col1 DOUBLE + *

+ * and clustering columns defined in query as + *

+ * bigInt_col1 + *

+ * And insert the following data: + *

+ * __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 + *

+ * When querying the table with query: 'SELECT * from ##tableName', because of the clustering + * defined on the table, the data should be reordered to: + *

+ * __time, bigint_col1 + * 2022-12-26T12:34:56,8 + * 2022-12-26T12:34:56,8 + * 2022-12-26T12:34:56,9 + * 2022-12-26T12:34:56,10 + * + */ + @Test + public void testInsertWithClusteringFromQuery() throws Exception + { + String queryFile = "/catalog/clustering_select.sql"; + String tableName = "testWithClusteringFromQuery" + operationName; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("bigint_col1", "BIGINT") + .build(); + + client.createTable(table, true); + LOG.info("table created:\n%s", client.readTable(table.id())); + String queryInline = + StringUtils.format(dmlPrefixPattern, tableName) + "\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " c AS bigint_col1\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" + + "CLUSTERED BY \"bigint_col1\"\n"; + + // Submit the task and wait for the datasource to get loaded + LOG.info("Running query:\n%s", queryInline); + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskSuccesfully(queryInline); + + if (sqlTaskStatus.getState().isFailure()) { + Assert.fail(StringUtils.format( + "Unable to start the task successfully.\nPossible exception: %s", + sqlTaskStatus.getError() + )); + } + + msqHelper.pollTaskIdForSuccess(sqlTaskStatus.getTaskId()); + dataLoaderHelper.waitUntilDatasourceIsReady(tableName); + + msqHelper.testQueriesFromFile(queryFile, tableName); + } + + /** + * Create table with columns: + *

+ * __time LONG + * varchar_col1 VARCHAR + * bigint_col1 BIGINT + * float_col1 FLOAT + * varchar_col2 VARCHAR + *

+ * and multiple clustering columns defined in catalog as + *

+ * bigInt_col1, varchar_col2 + *

+ * And insert the following data: + *

+ * __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 + *

+ * When querying the table with query: 'SELECT * from ##tableName', because of the clustering + * defined on the table, the data should be reordered to: + *

+ * __time, varchar_col1, bigint_col1, float_col1, varchar_col2 + * 2022-12-26T12:34:56,extra,8,"50",2.0,fop + * 2022-12-26T12:34:56,extra,8,"40",2.0,foq + * 2022-12-26T12:34:56,extra,9,"30",2.0,foo + * 2022-12-26T12:34:56,extra,10,"20",2.0,foo + * + */ + @Test + public void testInsertWithMultiClusteringFromCatalog() throws Exception + { + String queryFile = "/catalog/multiClustering_select.sql"; + String tableName = "testWithMultiClusteringFromCatalog" + operationName; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col1", "VARCHAR") + .column("bigint_col1", "BIGINT") + .column("float_col1", "FLOAT") + .column("varchar_col2", "VARCHAR") + .property( + DatasourceDefn.CLUSTER_KEYS_PROPERTY, + ImmutableList.of(new ClusterKeySpec("bigint_col1", false), new ClusterKeySpec("varchar_col2", false)) + ) + .build(); + + client.createTable(table, true); + LOG.info("table created:\n%s", client.readTable(table.id())); + 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" + + " e AS float_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 + LOG.info("Running query:\n%s", queryInline); + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskSuccesfully(queryInline); + + if (sqlTaskStatus.getState().isFailure()) { + Assert.fail(StringUtils.format( + "Unable to start the task successfully.\nPossible exception: %s", + sqlTaskStatus.getError() + )); + } + + msqHelper.pollTaskIdForSuccess(sqlTaskStatus.getTaskId()); + dataLoaderHelper.waitUntilDatasourceIsReady(tableName); + + msqHelper.testQueriesFromFile(queryFile, tableName); + } + + /** + * Create table with columns: + *

+ * __time LONG + * varchar_col1 VARCHAR + * bigint_col1 BIGINT + * float_col1 FLOAT + * varchar_col2 VARCHAR + *

+ * and multiple clustering columns defined in query as + *

+ * bigInt_col1, varchar_col2 + *

+ * And insert the following data: + *

+ * __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 + *

+ * When querying the table with query: 'SELECT * from ##tableName', because of the clustering + * defined on the query, the data should be reordered to: + *

+ * __time, varchar_col1, bigint_col1, float_col1, varchar_col2 + * 2022-12-26T12:34:56,extra,8,"50",2.0,fop + * 2022-12-26T12:34:56,extra,8,"40",2.0,foq + * 2022-12-26T12:34:56,extra,9,"30",2.0,foo + * 2022-12-26T12:34:56,extra,10,"20",2.0,foo + * + */ + @Test + public void testInsertWithMultiClusteringFromQuery() throws Exception + { + String queryFile = "/catalog/multiClustering_select.sql"; + String tableName = "testWithMultiClusteringFromQuery" + operationName; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col1", "VARCHAR") + .column("bigint_col1", "BIGINT") + .column("float_col1", "FLOAT") + .column("varchar_col2", "VARCHAR") + .build(); + + client.createTable(table, true); + LOG.info("table created:\n%s", client.readTable(table.id())); + 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" + + " e AS float_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" + + "CLUSTERED BY \"bigint_col1\", \"varchar_col2\"\n"; + + // Submit the task and wait for the datasource to get loaded + LOG.info("Running query:\n%s", queryInline); + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskSuccesfully(queryInline); + + if (sqlTaskStatus.getState().isFailure()) { + Assert.fail(StringUtils.format( + "Unable to start the task successfully.\nPossible exception: %s", + sqlTaskStatus.getError() + )); + } + + msqHelper.pollTaskIdForSuccess(sqlTaskStatus.getTaskId()); + dataLoaderHelper.waitUntilDatasourceIsReady(tableName); + + msqHelper.testQueriesFromFile(queryFile, tableName); + } +} diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestErrorTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestErrorTest.java new file mode 100644 index 00000000000..87e35e2b646 --- /dev/null +++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogIngestErrorTest.java @@ -0,0 +1,251 @@ +/* + * 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.testsEx.catalog; + +import com.google.common.collect.ImmutableMap; +import com.google.inject.Inject; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.TableMetadata; +import org.apache.druid.catalog.model.table.DatasourceDefn; +import org.apache.druid.catalog.model.table.TableBuilder; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.msq.sql.SqlTaskStatus; +import org.apache.druid.sql.http.SqlQuery; +import org.apache.druid.testing.utils.MsqTestQueryHelper; +import org.apache.druid.testsEx.categories.Catalog; +import org.apache.druid.testsEx.cluster.CatalogClient; +import org.apache.druid.testsEx.cluster.DruidClusterClient; +import org.apache.druid.testsEx.config.DruidTestRunner; +import org.jboss.netty.handler.codec.http.HttpResponseStatus; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.concurrent.ExecutionException; + +import static org.junit.Assert.assertTrue; + +/** + * Tests that expect failures when ingestng data into catalog defined tables. + */ +@RunWith(DruidTestRunner.class) +@Category(Catalog.class) +public class ITCatalogIngestErrorTest +{ + @Inject + private MsqTestQueryHelper msqHelper; + @Inject + private DruidClusterClient clusterClient; + private CatalogClient client; + + @Before + public void initializeClient() + { + client = new CatalogClient(clusterClient); + } + + /** + * If the segment grain is absent in the catalog and absent in the PARTITIONED BY clause in the query, then + * validation error. + */ + @Test + public void testInsertNoPartitonedByFromCatalogOrQuery() throws ExecutionException, InterruptedException + { + String tableName = "testInsertNoPartitonedByFromCatalogOrQuery"; + TableMetadata table = new TableBuilder(TableId.datasource(tableName), DatasourceDefn.TABLE_TYPE) + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col", "VARCHAR") + .column("bigint_col", "BIGINT") + .column("float_col", "FLOAT") + .build(); + + client.createTable(table, true); + String queryInline = + StringUtils.format( + "INSERT INTO %s\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " b AS varchar_col,\n" + + " c AS bigint_col,\n" + + " e AS float_col\n" + + "FROM TABLE(\n" + + " EXTERN(\n" + + " '{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,extra,10,\\\"20\\\",2.0,foo\"}',\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", + tableName + ); + + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskWithExpectedStatusCode(sqlQueryFromString(queryInline), null, null, HttpResponseStatus.BAD_REQUEST); + assertTrue(sqlTaskStatus.getError() != null && sqlTaskStatus.getError() + .getUnderlyingException() + .getMessage() + .equals( + "Operation [INSERT] requires a PARTITIONED BY to be explicitly defined, but none was found.") + ); + } + + /** + * Adding a new column during ingestion that is not defined in a sealed table should fail with + * proper validation error. + */ + @Test + public void testInsertNonDefinedColumnIntoSealedCatalogTable() throws ExecutionException, InterruptedException + { + String tableName = "testInsertNonDefinedColumnIntoSealedCatalogTable"; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col", "VARCHAR") + .column("bigint_col", "BIGINT") + .column("float_col", "FLOAT") + .property(DatasourceDefn.SEALED_PROPERTY, true) + .build(); + + client.createTable(table, true); + String queryInline = + StringUtils.format( + "INSERT INTO %s\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " b AS varchar_col,\n" + + " c AS bigint_col,\n" + + " e AS float_col,\n" + + " c AS extra\n" + + "FROM TABLE(\n" + + " EXTERN(\n" + + " '{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,extra,10,\\\"20\\\",2.0,foo\"}',\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" + + "PARTITIONED BY DAY\n", + tableName + ); + + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskWithExpectedStatusCode(sqlQueryFromString(queryInline), null, null, HttpResponseStatus.BAD_REQUEST); + assertTrue(sqlTaskStatus.getError() != null && sqlTaskStatus.getError() + .getUnderlyingException() + .getMessage() + .equals( + "Column [extra] is not defined in the target table [druid.testInsertNonDefinedColumnIntoSealedCatalogTable] strict schema") + ); + } + + /** + * Assigning a column during ingestion, to an input type that is not compatible with the defined type of the + * column, should result in a proper validation error. + */ + @Test + public void testInsertWithIncompatibleTypeAssignment() throws ExecutionException, InterruptedException + { + String tableName = "testInsertWithIncompatibleTypeAssignment"; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col", "VARCHAR") + .column("bigint_col", "BIGINT") + .column("float_col", "FLOAT") + .property(DatasourceDefn.SEALED_PROPERTY, true) + .build(); + + client.createTable(table, true); + String queryInline = + StringUtils.format( + "INSERT INTO %s\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " ARRAY[b] AS varchar_col,\n" + + " c AS bigint_col,\n" + + " e AS float_col\n" + + "FROM TABLE(\n" + + " EXTERN(\n" + + " '{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,extra,10,\\\"20\\\",2.0,foo\"}',\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" + + "PARTITIONED BY DAY\n", + tableName + ); + + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskWithExpectedStatusCode(sqlQueryFromString(queryInline), null, null, HttpResponseStatus.BAD_REQUEST); + assertTrue(sqlTaskStatus.getError() != null && sqlTaskStatus.getError() + .getUnderlyingException() + .getMessage() + .equals( + "Cannot assign to target field 'varchar_col' of type VARCHAR from source field 'varchar_col' of type VARCHAR ARRAY (line [4], column [3])") + ); + } + + /** + * Assigning a complex type column during ingestion, to an input type that is not compatible with the defined type of + * the column, should result in a proper validation error. + */ + @Test + public void testInsertGroupByWithIncompatibleTypeAssignment() throws ExecutionException, InterruptedException + { + String tableName = "testInsertGroupByWithIncompatibleTypeAssignment"; + TableMetadata table = TableBuilder.datasource(tableName, "P1D") + .column(Columns.TIME_COLUMN, Columns.LONG) + .column("varchar_col", "VARCHAR") + .column("bigint_col", "BIGINT") + .column("float_col", "FLOAT") + .column("hll_col", "COMPLEX") + .property(DatasourceDefn.SEALED_PROPERTY, true) + .build(); + + client.createTable(table, true); + String queryInline = + StringUtils.format( + "INSERT INTO %s\n" + + "SELECT\n" + + " TIME_PARSE(a) AS __time,\n" + + " b AS varchar_col,\n" + + " c AS bigint_col,\n" + + " e AS float_col,\n" + + " ARRAY[b] AS hll_col\n" + + "FROM TABLE(\n" + + " EXTERN(\n" + + " '{\"type\":\"inline\",\"data\":\"2022-12-26T12:34:56,extra,10,\\\"20\\\",2.0,foo\"}',\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" + + "PARTITIONED BY DAY\n", + tableName + ); + + SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTaskWithExpectedStatusCode(sqlQueryFromString(queryInline), null, null, HttpResponseStatus.BAD_REQUEST); + assertTrue(sqlTaskStatus.getError() != null && sqlTaskStatus.getError() + .getUnderlyingException() + .getMessage() + .equals( + "Cannot assign to target field 'hll_col' of type COMPLEX from source field 'hll_col' of type VARCHAR ARRAY (line [7], column [3])") + ); + } + + private static SqlQuery sqlQueryFromString(String queryString) + { + return new SqlQuery(queryString, null, false, false, false, ImmutableMap.of(), null); + } +} diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogInsertAndQueryTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogInsertAndQueryTest.java new file mode 100644 index 00000000000..26c26a5a718 --- /dev/null +++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogInsertAndQueryTest.java @@ -0,0 +1,42 @@ +/* + * 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.testsEx.catalog; + +import org.apache.druid.testsEx.categories.Catalog; +import org.apache.druid.testsEx.config.DruidTestRunner; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +@RunWith(DruidTestRunner.class) +@Category(Catalog.class) +public class ITCatalogInsertAndQueryTest extends ITCatalogIngestAndQueryTest +{ + @Override + public String getOperationName() + { + return "INSERT"; + } + + @Override + public String getDmlPrefixPattern() + { + return "INSERT INTO \"%s\""; + } +} diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogReplaceAndQueryTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogReplaceAndQueryTest.java new file mode 100644 index 00000000000..d6f1898a5f5 --- /dev/null +++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogReplaceAndQueryTest.java @@ -0,0 +1,42 @@ +/* + * 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.testsEx.catalog; + +import org.apache.druid.testsEx.categories.Catalog; +import org.apache.druid.testsEx.config.DruidTestRunner; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +@RunWith(DruidTestRunner.class) +@Category(Catalog.class) +public class ITCatalogReplaceAndQueryTest extends ITCatalogIngestAndQueryTest +{ + @Override + public String getOperationName() + { + return "REPLACE"; + } + + @Override + public String getDmlPrefixPattern() + { + return "REPLACE INTO \"%s\" OVERWRITE ALL"; + } +} diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogRestTest.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogRestTest.java index e6b9153a232..8082fc0aa7c 100644 --- a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogRestTest.java +++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/catalog/ITCatalogRestTest.java @@ -19,6 +19,7 @@ package org.apache.druid.testsEx.catalog; +import com.google.common.collect.ImmutableList; import com.google.inject.Inject; import org.apache.druid.catalog.http.TableEditRequest.DropColumns; import org.apache.druid.catalog.http.TableEditRequest.HideColumns; @@ -28,6 +29,7 @@ import org.apache.druid.catalog.model.CatalogUtils; 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.catalog.model.table.TableBuilder; import org.apache.druid.java.util.common.ISE; @@ -106,6 +108,17 @@ public class ITCatalogRestTest () -> client.createTable(table, false) ); } + + // DESC cluster keys not supported + { + final TableMetadata table = TableBuilder.datasource("foo", "P1D") + .property(DatasourceDefn.CLUSTER_KEYS_PROPERTY, ImmutableList.of(new ClusterKeySpec("clusterKeyA", true))) + .build(); + assertThrows( + Exception.class, + () -> client.createTable(table, false) + ); + } } /** diff --git a/integration-tests-ex/cases/src/test/resources/catalog/clustering_select.sql b/integration-tests-ex/cases/src/test/resources/catalog/clustering_select.sql new file mode 100644 index 00000000000..43b47180f42 --- /dev/null +++ b/integration-tests-ex/cases/src/test/resources/catalog/clustering_select.sql @@ -0,0 +1,23 @@ +[ + { + "query": "SELECT * FROM %%DATASOURCE%%", + "expectedResults": [ + { + "__time": 1672058096000, + "bigint_col1": 8 + }, + { + "__time": 1672058096000, + "bigint_col1": 8 + }, + { + "__time": 1672058096000, + "bigint_col1": 9 + }, + { + "__time": 1672058096000, + "bigint_col1": 10 + } + ] + } +] \ No newline at end of file diff --git a/integration-tests-ex/cases/src/test/resources/catalog/implicitCast_select.sql b/integration-tests-ex/cases/src/test/resources/catalog/implicitCast_select.sql new file mode 100644 index 00000000000..290575d5af9 --- /dev/null +++ b/integration-tests-ex/cases/src/test/resources/catalog/implicitCast_select.sql @@ -0,0 +1,23 @@ +[ + { + "query": "SELECT * FROM %%DATASOURCE%%", + "expectedResults": [ + { + "__time": 1672058096000, + "double_col1": 8.0 + }, + { + "__time": 1672058096000, + "double_col1": 8.0 + }, + { + "__time": 1672058096000, + "double_col1": 9.0 + }, + { + "__time": 1672058096000, + "double_col1": 10.0 + } + ] + } +] \ No newline at end of file diff --git a/integration-tests-ex/cases/src/test/resources/catalog/multiClustering_select.sql b/integration-tests-ex/cases/src/test/resources/catalog/multiClustering_select.sql new file mode 100644 index 00000000000..b83302e0d0d --- /dev/null +++ b/integration-tests-ex/cases/src/test/resources/catalog/multiClustering_select.sql @@ -0,0 +1,35 @@ +[ + { + "query": "SELECT * FROM %%DATASOURCE%%", + "expectedResults": [ + { + "__time": 1672058096000, + "bigint_col1": 8, + "varchar_col2": "fop", + "varchar_col1": "extra", + "float_col1": 2.0 + }, + { + "__time": 1672058096000, + "bigint_col1": 8, + "varchar_col2": "foq", + "varchar_col1": "extra", + "float_col1": 2.0 + }, + { + "__time": 1672058096000, + "bigint_col1": 9, + "varchar_col2": "foo", + "varchar_col1": "extra", + "float_col1": 2.0 + }, + { + "__time": 1672058096000, + "bigint_col1": 10, + "varchar_col2": "foo", + "varchar_col1": "extra", + "float_col1": 2.0 + } + ] + } +] \ No newline at end of file diff --git a/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java b/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java index 2a2386869e4..4efd5c7d42d 100644 --- a/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java +++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java @@ -121,14 +121,29 @@ public class MsqTestQueryHelper extends AbstractTestQueryHelper