Catalog integration tests

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<hyperUnique>")
+ .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<hyperUnique> 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);
+ }
+}

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\"";
+ }
+}

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";
+ }
+}

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.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 void testErrors()
  () -> 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)
+ );
+ }
  }
 
  /**

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
+ }
+ ]
+ }
+]