You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "zachjsh (via GitHub)" <gi...@apache.org> on 2024/04/10 20:12:38 UTC
[PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
zachjsh opened a new pull request, #16260:
URL: https://github.com/apache/druid/pull/16260
### Description
This PR contains a portion of the changes from the inactive draft PR for integrating the catalog with the Calcite planner https://github.com/apache/druid/pull/13686 from @paul-rogers, allowing for tables that are defined in the catalog to have any defined clustering columns used in DML INSERT/REPLACE operations without needing to be specified at query time. If the user specified a clustering columns at query time, these columns are preferred to the catalog defined clustering columns. This PR is built on top of https://github.com/apache/druid/pull/16223, so that change should be merged first
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1562501886
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -88,6 +101,39 @@ public class DruidSqlValidator extends BaseDruidSqlValidator
private final PlannerContext plannerContext;
+ // This part is a bit sad. By the time we get here, the validator will have created
+ // the ORDER BY namespace if we had a real ORDER BY. We have to "catch up" and do the
+ // work that registerQuery() should have done. That's kind of OK. But, the orderScopes
+ // variable is private, so we have to play dirty tricks to get at it.
+ //
+ // Warning: this may no longer work if Java forbids access to private fields in a
+ // future release.
+ private static final Field CLAUSE_SCOPES_FIELD;
+ private static final Object ORDER_CLAUSE;
+
+ static {
+ try {
+ CLAUSE_SCOPES_FIELD = FieldUtils.getDeclaredField(
+ SqlValidatorImpl.class,
+ "clauseScopes",
+ true
+ );
+ }
+ catch (RuntimeException e) {
+ throw new ISE(e, "SqlValidatorImpl.clauseScopes is not accessible");
+ }
+
+ try {
+ Class<?> innerClass = Class.forName("org.apache.calcite.sql.validate.SqlValidatorImpl$Clause");
+ Method method = innerClass.getMethod("valueOf", Class.class, String.class);
+ method.setAccessible(true);
+ ORDER_CLAUSE = method.invoke(null, innerClass, "ORDER");
Review Comment:
this means we are not anymore use the api - instead open up a field via reflection; there must be a better way.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1580023194
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -397,6 +459,34 @@ private Granularity getEffectiveGranularity(
return effectiveGranularity;
}
+ @Nullable
+ private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ final List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+ if (CollectionUtils.isNullOrEmpty(keyCols)) {
+ return null;
+ }
+ final SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+ for (ClusterKeySpec keyCol : keyCols) {
+ final SqlIdentifier colIdent = new SqlIdentifier(
+ Collections.singletonList(keyCol.expr()),
+ null, SqlParserPos.ZERO,
+ Collections.singletonList(SqlParserPos.ZERO)
+ );
Review Comment:
About whether there is a check that al keyCols are present in the selected column list, see the following tests:
`testInsertTableWithClusteringWithClusteringOnNewColumnFromQuery`
`testInsertTableWithClusteringWithClusteringOnBadColumn`
Do these cover the cases you are talking about?
About the join issue, do you have a concrete query in example, just to clarify?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh merged PR #16260:
URL: https://github.com/apache/druid/pull/16260
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1580019321
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -281,6 +286,37 @@ public void validateInsert(final SqlInsert insert)
}
}
+ @Override
+ protected SelectNamespace createSelectNamespace(
+ SqlSelect select,
+ SqlNode enclosingNode)
+ {
+ if (enclosingNode instanceof DruidSqlIngest) {
+ // The target is a new or existing datasource.
+ // The target namespace is both the target table ID and the row type for that table.
+ final SqlValidatorNamespace targetNamespace = Objects.requireNonNull(
+ getNamespace(enclosingNode),
+ () -> "namespace for " + enclosingNode
+ );
+ final IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+ SqlIdentifier identifier = insertNs.getId();
+ SqlValidatorTable catalogTable = getCatalogReader().getTable(identifier.names);
+ if (catalogTable != null) {
Review Comment:
if the ingestNode already has the clustering columns, they will be used. There are existing tests which test that the clustering columns are used in the plan returned from dml query, when clustering is defined at query time, and the table is / it not in catalog. Let me know if this covers the issue that think could occur.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1569365887
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -348,6 +398,42 @@ private DatasourceTable validateInsertTarget(
}
}
+ /**
+ * Clustering is a kind of ordering. We push the CLUSTERED BY clause into the source query as
+ * an ORDER BY clause. In an ideal world, clustering would be outside of SELECT: it is an operation
+ * applied after the data is selected. For now, we push it into the SELECT, then MSQ pulls it back
+ * out. This is unfortunate as errors will be attributed to ORDER BY, which the user did not
+ * actually specify (it is an error to do so.) However, with the current hybrid structure, it is
+ * not possible to add the ORDER by later: doing so requires access to the order by namespace
+ * which is not visible to subclasses.
+ */
+ private void rewriteClusteringToOrderBy(SqlNode source, DruidSqlIngest ingestNode, @Nullable SqlNodeList catalogClustering)
Review Comment:
Thank you for the suggestion, this worked great! Let me know if good now.
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -88,6 +101,39 @@ public class DruidSqlValidator extends BaseDruidSqlValidator
private final PlannerContext plannerContext;
+ // This part is a bit sad. By the time we get here, the validator will have created
+ // the ORDER BY namespace if we had a real ORDER BY. We have to "catch up" and do the
+ // work that registerQuery() should have done. That's kind of OK. But, the orderScopes
+ // variable is private, so we have to play dirty tricks to get at it.
+ //
+ // Warning: this may no longer work if Java forbids access to private fields in a
+ // future release.
+ private static final Field CLAUSE_SCOPES_FIELD;
+ private static final Object ORDER_CLAUSE;
+
+ static {
+ try {
+ CLAUSE_SCOPES_FIELD = FieldUtils.getDeclaredField(
+ SqlValidatorImpl.class,
+ "clauseScopes",
+ true
+ );
+ }
+ catch (RuntimeException e) {
+ throw new ISE(e, "SqlValidatorImpl.clauseScopes is not accessible");
+ }
+
+ try {
+ Class<?> innerClass = Class.forName("org.apache.calcite.sql.validate.SqlValidatorImpl$Clause");
+ Method method = innerClass.getMethod("valueOf", Class.class, String.class);
+ method.setAccessible(true);
+ ORDER_CLAUSE = method.invoke(null, innerClass, "ORDER");
Review Comment:
Took your suggestion here, great tip!
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1560032789
##########
sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java:
##########
@@ -210,11 +287,727 @@
final DatasourceTable.PhysicalDatasourceMetadata dsMetadata
)
{
- if (resolvedTables.get(tableName) != null) {
- return resolvedTables.get(tableName);
+ if (RESOLVED_TABLES.get(tableName) != null) {
+ return RESOLVED_TABLES.get(tableName);
}
return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
}
};
}
+
+ /**
+ * If the segment grain is given in the catalog and absent in the PARTITIONED BY clause in the query, then use the
+ * value from the catalog.
+ */
+ @Test
+ public void testInsertHourGrainPartitonedByFromCatalog()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "hourDs") + "\n" +
+ "SELECT * FROM foo")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("hourDs", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.HOUR))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * If the segment grain is given in the catalog, and also by PARTITIONED BY, then
+ * the query value is used.
+ */
+ @Test
+ public void testInsertHourGrainWithDayPartitonedByFromQuery()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "hourDs") + "\n" +
+ "SELECT * FROM foo\n" +
+ "PARTITIONED BY day")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("hourDs", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.DAY))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * 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 testInsertNoPartitonedByFromCatalog()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "noPartitonedBy") + "\n" +
+ "SELECT * FROM foo")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectValidationError(
+ DruidException.class,
+ StringUtils.format("Operation [%s] requires a PARTITIONED BY to be explicitly defined, but none was found.", operationName)
+ )
+ .verify();
+ }
+
+ /**
+ * If the segment grain is absent in the catalog, but given by PARTITIONED BY, then
+ * the query value is used.
+ */
+ @Test
+ public void testInsertNoPartitonedByWithDayPartitonedByFromQuery()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "noPartitonedBy") + "\n" +
+ "SELECT * FROM foo\n" +
+ "PARTITIONED BY day")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("noPartitonedBy", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("noPartitonedBy"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.DAY))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Adding a new column during ingestion that is not defined in a non-sealed table should succeed.
+ */
+ @Test
+ public void testInsertAddNonDefinedColumnIntoNonSealedCatalogTable()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .add("m2", ColumnType.DOUBLE)
+ .add("extra2", ColumnType.LONG)
+ .add("extra3", ColumnType.STRING)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "foo") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " 1 AS cnt,\n" +
+ " c AS m2,\n" +
+ " CAST(d AS BIGINT) AS extra2,\n" +
+ " e AS extra3\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("foo", signature)
+ .expectResources(dataSourceWrite("foo"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG),
+ expressionVirtualColumn("v2", "CAST(\"c\", 'DOUBLE')", ColumnType.DOUBLE),
+ expressionVirtualColumn("v3", "CAST(\"d\", 'LONG')", ColumnType.LONG)
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "e", "v0", "v1", "v2", "v3")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition. Should use
+ * the catalog defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringFromCatalog()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " 1 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("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("b", ScanQuery.Order.ASCENDING),
+ new ScanQuery.OrderBy("d", ScanQuery.Order.DESCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query. Should use the query defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringFromQuery()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " 1 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\n" +
+ "CLUSTERED BY dim1")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("b", ScanQuery.Order.ASCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query on column that has not been defined in the table catalog definition.
+ * Should use the query defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringOnNewColumnFromQuery()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("dim3", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " e AS dim3,\n" +
+ " 1 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\n" +
+ "CLUSTERED BY dim3")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("e", ScanQuery.Order.ASCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "e", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query on column that has not been defined in the table catalog definition.
+ * or in the select clause. Should fail with validation error.
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringOnBadColumn()
+ {
+ 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),
+ RowSignature.builder()
+ .add("a", ColumnType.STRING)
+ .add("b", ColumnType.STRING)
+ .add("c", ColumnType.LONG)
+ .add("d", ColumnType.STRING)
+ .add("e", ColumnType.STRING)
+ .build()
+ );
Review Comment:
## Unread local variable
Variable 'ExternalDataSource externalDataSource' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/7257)
##########
sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java:
##########
@@ -210,11 +287,727 @@
final DatasourceTable.PhysicalDatasourceMetadata dsMetadata
)
{
- if (resolvedTables.get(tableName) != null) {
- return resolvedTables.get(tableName);
+ if (RESOLVED_TABLES.get(tableName) != null) {
+ return RESOLVED_TABLES.get(tableName);
}
return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
}
};
}
+
+ /**
+ * If the segment grain is given in the catalog and absent in the PARTITIONED BY clause in the query, then use the
+ * value from the catalog.
+ */
+ @Test
+ public void testInsertHourGrainPartitonedByFromCatalog()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "hourDs") + "\n" +
+ "SELECT * FROM foo")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("hourDs", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.HOUR))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * If the segment grain is given in the catalog, and also by PARTITIONED BY, then
+ * the query value is used.
+ */
+ @Test
+ public void testInsertHourGrainWithDayPartitonedByFromQuery()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "hourDs") + "\n" +
+ "SELECT * FROM foo\n" +
+ "PARTITIONED BY day")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("hourDs", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("hourDs"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.DAY))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * 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 testInsertNoPartitonedByFromCatalog()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "noPartitonedBy") + "\n" +
+ "SELECT * FROM foo")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectValidationError(
+ DruidException.class,
+ StringUtils.format("Operation [%s] requires a PARTITIONED BY to be explicitly defined, but none was found.", operationName)
+ )
+ .verify();
+ }
+
+ /**
+ * If the segment grain is absent in the catalog, but given by PARTITIONED BY, then
+ * the query value is used.
+ */
+ @Test
+ public void testInsertNoPartitonedByWithDayPartitonedByFromQuery()
+ {
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "noPartitonedBy") + "\n" +
+ "SELECT * FROM foo\n" +
+ "PARTITIONED BY day")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("noPartitonedBy", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceWrite("noPartitonedBy"), dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("__time", "cnt", "dim1", "dim2", "dim3", "m1", "m2", "unique_dim1")
+ .context(queryContextWithGranularity(Granularities.DAY))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Adding a new column during ingestion that is not defined in a non-sealed table should succeed.
+ */
+ @Test
+ public void testInsertAddNonDefinedColumnIntoNonSealedCatalogTable()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .add("m2", ColumnType.DOUBLE)
+ .add("extra2", ColumnType.LONG)
+ .add("extra3", ColumnType.STRING)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "foo") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " 1 AS cnt,\n" +
+ " c AS m2,\n" +
+ " CAST(d AS BIGINT) AS extra2,\n" +
+ " e AS extra3\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("foo", signature)
+ .expectResources(dataSourceWrite("foo"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG),
+ expressionVirtualColumn("v2", "CAST(\"c\", 'DOUBLE')", ColumnType.DOUBLE),
+ expressionVirtualColumn("v3", "CAST(\"d\", 'LONG')", ColumnType.LONG)
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "e", "v0", "v1", "v2", "v3")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition. Should use
+ * the catalog defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringFromCatalog()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " 1 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("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("b", ScanQuery.Order.ASCENDING),
+ new ScanQuery.OrderBy("d", ScanQuery.Order.DESCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query. Should use the query defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringFromQuery()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " 1 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\n" +
+ "CLUSTERED BY dim1")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("b", ScanQuery.Order.ASCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query on column that has not been defined in the table catalog definition.
+ * Should use the query defined clustering
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringOnNewColumnFromQuery()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("dim3", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
+ testIngestionQuery()
+ .sql(StringUtils.format(dmlPrefixPattern, "tableWithClustering") + "\n" +
+ "SELECT\n" +
+ " TIME_PARSE(a) AS __time,\n" +
+ " b AS dim1,\n" +
+ " d AS dim2,\n" +
+ " e AS dim3,\n" +
+ " 1 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\n" +
+ "CLUSTERED BY dim3")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("tableWithClustering", signature)
+ .expectResources(dataSourceWrite("tableWithClustering"), Externals.externalRead("EXTERNAL"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource(externalDataSource)
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .virtualColumns(
+ expressionVirtualColumn("v0", "timestamp_parse(\"a\",null,'UTC')", ColumnType.LONG),
+ expressionVirtualColumn("v1", "1", ColumnType.LONG)
+ )
+ .orderBy(
+ ImmutableList.of(
+ new ScanQuery.OrderBy("e", ScanQuery.Order.ASCENDING)
+ )
+ )
+ // Scan query lists columns in alphabetical order independent of the
+ // SQL project list or the defined schema.
+ .columns("b", "d", "e", "v0", "v1")
+ .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * Insert into a catalog table that has clustering defined on the table definition, but user specifies
+ * clustering on the ingest query on column that has not been defined in the table catalog definition.
+ * or in the select clause. Should fail with validation error.
+ */
+ @Test
+ public void testInsertTableWithClusteringWithClusteringOnBadColumn()
+ {
+ 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),
+ 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("dim1", ColumnType.STRING)
+ .add("dim2", ColumnType.STRING)
+ .add("dim3", ColumnType.STRING)
+ .add("cnt", ColumnType.LONG)
+ .build();
Review Comment:
## Unread local variable
Variable 'RowSignature signature' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/7258)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1580023194
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -397,6 +459,34 @@ private Granularity getEffectiveGranularity(
return effectiveGranularity;
}
+ @Nullable
+ private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ final List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+ if (CollectionUtils.isNullOrEmpty(keyCols)) {
+ return null;
+ }
+ final SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+ for (ClusterKeySpec keyCol : keyCols) {
+ final SqlIdentifier colIdent = new SqlIdentifier(
+ Collections.singletonList(keyCol.expr()),
+ null, SqlParserPos.ZERO,
+ Collections.singletonList(SqlParserPos.ZERO)
+ );
Review Comment:
About whether there is a check that al keyCols are present in the selected column list, see the following tests:
`testInsertTableWithClusteringWithClusteringOnNewColumnFromQuery`
`testInsertTableWithClusteringWithClusteringOnBadColumn`
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1562521486
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -348,6 +398,42 @@ private DatasourceTable validateInsertTarget(
}
}
+ /**
+ * Clustering is a kind of ordering. We push the CLUSTERED BY clause into the source query as
+ * an ORDER BY clause. In an ideal world, clustering would be outside of SELECT: it is an operation
+ * applied after the data is selected. For now, we push it into the SELECT, then MSQ pulls it back
+ * out. This is unfortunate as errors will be attributed to ORDER BY, which the user did not
+ * actually specify (it is an error to do so.) However, with the current hybrid structure, it is
+ * not possible to add the ORDER by later: doing so requires access to the order by namespace
+ * which is not visible to subclasses.
+ */
+ private void rewriteClusteringToOrderBy(SqlNode source, DruidSqlIngest ingestNode, @Nullable SqlNodeList catalogClustering)
Review Comment:
instead of this creative reflectioned approach ; how about overriding the `createSelectNamespace`;
in that method you will see:
* the current `SqlSelect` node
* the encosingNode - which will be `SqlInsert` or something...so can give access the catalog orderby somehow
* full control on hiw the `SelectNamespace` is being created - it might be even possible to supply an alternate implementation
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] INSERT/REPLACE can omit clustering when catalog has default (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #16260:
URL: https://github.com/apache/druid/pull/16260#discussion_r1577475455
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -348,6 +398,42 @@ private DatasourceTable validateInsertTarget(
}
}
+ /**
+ * Clustering is a kind of ordering. We push the CLUSTERED BY clause into the source query as
+ * an ORDER BY clause. In an ideal world, clustering would be outside of SELECT: it is an operation
+ * applied after the data is selected. For now, we push it into the SELECT, then MSQ pulls it back
+ * out. This is unfortunate as errors will be attributed to ORDER BY, which the user did not
+ * actually specify (it is an error to do so.) However, with the current hybrid structure, it is
+ * not possible to add the ORDER by later: doing so requires access to the order by namespace
+ * which is not visible to subclasses.
+ */
+ private void rewriteClusteringToOrderBy(SqlNode source, DruidSqlIngest ingestNode, @Nullable SqlNodeList catalogClustering)
Review Comment:
much better - thank you for the updates!
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -281,6 +286,37 @@ public void validateInsert(final SqlInsert insert)
}
}
+ @Override
+ protected SelectNamespace createSelectNamespace(
+ SqlSelect select,
+ SqlNode enclosingNode)
+ {
+ if (enclosingNode instanceof DruidSqlIngest) {
+ // The target is a new or existing datasource.
+ // The target namespace is both the target table ID and the row type for that table.
+ final SqlValidatorNamespace targetNamespace = Objects.requireNonNull(
+ getNamespace(enclosingNode),
+ () -> "namespace for " + enclosingNode
+ );
+ final IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+ SqlIdentifier identifier = insertNs.getId();
+ SqlValidatorTable catalogTable = getCatalogReader().getTable(identifier.names);
+ if (catalogTable != null) {
Review Comment:
wouldn't the fall-thru from this condtional will cause that the `CLUSTER BY` on the `ingestNode` will not be applied (line399 right now); even if its there - is that okay?
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -397,6 +459,34 @@ private Granularity getEffectiveGranularity(
return effectiveGranularity;
}
+ @Nullable
+ private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ final List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+ if (CollectionUtils.isNullOrEmpty(keyCols)) {
+ return null;
+ }
+ final SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+ for (ClusterKeySpec keyCol : keyCols) {
+ final SqlIdentifier colIdent = new SqlIdentifier(
+ Collections.singletonList(keyCol.expr()),
+ null, SqlParserPos.ZERO,
+ Collections.singletonList(SqlParserPos.ZERO)
+ );
Review Comment:
I was wondering what will happen in the following case:
* say colunmn `c` is a `clusterKey`
* we are selecting from a join which has column `c` on both sides
but it seems like the column in the select list will take precedence.
one more thing I was wondering about: do we have a check that all `keyCols` are present in the selected column list?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org