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/01/17 21:47:25 UTC
[PR] Integrate catalog schema validation into planner -WIP (druid)
zachjsh opened a new pull request, #15711:
URL: https://github.com/apache/druid/pull/15711
### 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 the datasource table schemas defined in the catalog to be validated against when ingesting data into the underlying datasource, during SQL based ingestion. This allows for SQL based ingestion into a table with a declared schema to produce segments that conform to that schema. If partitioning and clustering is not defined at ingestion time, defaults for these parameters, as defined in the catalog for the table, if found, are used.
TODO: add more tests.
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15711:
URL: https://github.com/apache/druid/pull/15711#issuecomment-2087756735
This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1456553673
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java:
##########
@@ -133,6 +141,41 @@
this.context = config.getContext();
this.connectionConfig = connConfig(context);
this.typeSystem = config.getTypeSystem();
+
+ // With the catalog, schemas can provide functions.
+ // Add the necessary indirection. The type factory used here
+ // is the Druid one, since the per-query one is not yet available
+ // here. Nor are built-in function associated with per-query types.
+ this.operatorTable = new ChainedSqlOperatorTable(
Review Comment:
## Deprecated method or constructor invocation
Invoking [ChainedSqlOperatorTable.ChainedSqlOperatorTable](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/6469)
##########
sql/src/main/java/org/apache/druid/sql/calcite/run/QueryMaker.java:
##########
@@ -28,6 +29,17 @@
*/
public interface QueryMaker
{
+ /**
+ * Do everything that would be done to run a query, don't actually run.
+ * Instead return what would have been sent to the execution engine.
+ * The result is a Jackson-serializable query plan.
+ */
+ default Object explain(DruidQuery druidQuery)
Review Comment:
## Useless parameter
The parameter 'druidQuery' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/6471)
##########
sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerCaptureHook.java:
##########
@@ -33,11 +35,14 @@
private RelRoot relRoot;
private SqlInsert insertNode;
private SqlNode sqlNode;
+ private DruidRel<?> druidRel;
+ private Object execPlan;
@Override
public void captureSql(String sql)
{
- // Not used at present. Add a field to capture this if you need it.
+ this.druidRel = druidRel;
Review Comment:
## Self assignment
This expression assigns druidRel to itself.
[Show more details](https://github.com/apache/druid/security/code-scanning/6470)
##########
sql/src/main/java/org/apache/druid/sql/calcite/run/SqlEngine.java:
##########
@@ -94,6 +94,7 @@
QueryMaker buildQueryMakerForInsert(
String targetDataSource,
RelRoot relRoot,
- PlannerContext plannerContext
+ PlannerContext plannerContext,
+ RelDataType targetType
Review Comment:
## Useless parameter
The parameter 'targetType' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/6472)
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1473616138
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -19,38 +19,762 @@
package org.apache.druid.sql.calcite.planner;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.prepare.BaseDruidSqlValidator;
import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelRecordType;
import org.apache.calcite.runtime.CalciteContextException;
import org.apache.calcite.runtime.CalciteException;
+import org.apache.calcite.sql.SqlBasicCall;
import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlInsert;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlNumericLiteral;
import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.BasicSqlType;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.validate.IdentifierNamespace;
+import org.apache.calcite.sql.validate.OrderByScope;
+import org.apache.calcite.sql.validate.SqlValidatorException;
+import org.apache.calcite.sql.validate.SqlValidatorImpl;
+import org.apache.calcite.sql.validate.SqlValidatorNamespace;
import org.apache.calcite.sql.validate.SqlValidatorScope;
+import org.apache.calcite.sql.validate.SqlValidatorTable;
+import org.apache.calcite.sql.validate.ValidatorShim;
+import org.apache.calcite.util.Pair;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.error.InvalidSqlInput;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.PeriodGranularity;
+import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
+import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
+import org.apache.druid.sql.calcite.parser.DruidSqlParserUtils;
import org.apache.druid.sql.calcite.run.EngineFeature;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
/**
* Druid extended SQL validator. (At present, it doesn't actually
* have any extensions yet, but it will soon.)
*/
class DruidSqlValidator extends BaseDruidSqlValidator
{
+ private static final Pattern UNNAMED_COLUMN_PATTERN = Pattern.compile("^EXPR\\$\\d+$", Pattern.CASE_INSENSITIVE);
+
+ // Copied here from MSQE since that extension is not visible here.
+ public static final String CTX_ROWS_PER_SEGMENT = "msqRowsPerSegment";
+
+ public interface ValidatorContext
+ {
+ Map<String, Object> queryContextMap();
+ CatalogResolver catalog();
+ String druidSchemaName();
+ ObjectMapper jsonMapper();
+ }
+
+ private final ValidatorContext validatorContext;
private final PlannerContext plannerContext;
protected DruidSqlValidator(
- SqlOperatorTable opTab,
- CalciteCatalogReader catalogReader,
- JavaTypeFactory typeFactory,
- Config validatorConfig,
- PlannerContext plannerContext
+ final SqlOperatorTable opTab,
+ final CalciteCatalogReader catalogReader,
+ final JavaTypeFactory typeFactory,
+ final Config validatorConfig,
+ final PlannerContext plannerContext,
+ final ValidatorContext validatorContext
)
{
super(opTab, catalogReader, typeFactory, validatorConfig);
this.plannerContext = plannerContext;
+ this.validatorContext = validatorContext;
+ }
+
+ /**
+ * Druid-specific validation for an INSERT statement. In Druid, the columns are
+ * matched by name. A datasource, by default, allows the insertion of arbitrary columns,
+ * but the catalog may enforce a strict schema (all columns must exist). Destination
+ * types are set by the catalog, where available, else by the query.
+ * <p>
+ * The Druid {@code INSERT} statement is non-standard in a variety of ways:
+ * <ul>
+ * <li>Allows the target table to not yet exist. Instead, {@code INSERT}
+ * creates it.</li>
+ * <li>Does not allow specifying the list of columns:
+ * {@code INSERT INTO dst (a, b, c) ...}</li>
+ * <li>When given without target columns (the only form allowed), columns are
+ * not matched by schema position as in standard SQL, but rather by name.</li>
+ * <li>There is no requirement that the target columns already exist. In fact,
+ * even if the target column exists, any existing type is ignored if not specified
+ * in the catalog.</li>
+ * <li>The source can only be a {@code SELECT} statement, not {@code VALUES}.</li>
+ * <li>Types thus propagate upwards from the {@code SELECT} to the the target
+ * table. Standard SQL says that types propagate down from the target table to the
+ * source.</li>
+ * <li>The __time column is special in multiple ways.</li>
+ * <li>Includes the {@code CLUSTERED BY} and {@code PARTITIONED BY} clauses.</li>
+ * </ul>
+ * The result is that the validation for the Druid {@code INSERT} is wildly customized
+ * relative to standard SQL.
+ */
+ // TODO: Ensure the source and target are not the same
+ @Override
+ public void validateInsert(final SqlInsert insert)
+ {
+ final DruidSqlIngest ingestNode = (DruidSqlIngest) insert;
+ if (insert.isUpsert()) {
+ throw new IAE("UPSERT is not supported.");
+ }
+
+ // SQL-style INSERT INTO dst (a, b, c) is not (yet) supported.
+ final String operationName = insert.getOperator().getName();
+ if (insert.getTargetColumnList() != null) {
+ throw new IAE("%s with a target column list is not supported.", operationName);
+ }
+
+ // The target namespace is both the target table ID and the row type for that table.
+ final SqlValidatorNamespace targetNamespace = getNamespace(insert);
+ final IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+
+ // The target is a new or existing datasource.
+ final DatasourceTable table = validateInsertTarget(targetNamespace, insertNs, operationName);
+ final SqlValidatorTable target = insertNs.resolve().getTable();
+
+ // An existing datasource may have metadata.
+ final DatasourceFacade tableMetadata = table == null ? null : table.effectiveMetadata().catalogMetadata();
+
+ // Validate segment granularity, which depends on nothing else.
+ validateSegmentGranularity(operationName, ingestNode, tableMetadata);
+
+ // The source must be a SELECT
+ final SqlNode source = insert.getSource();
+ ensureNoOrderBy(source, operationName);
+
+ // Convert CLUSTERED BY, or the catalog equivalent, to an ORDER BY clause
+ final SqlNodeList catalogClustering = convertCatalogClustering(tableMetadata);
+ rewriteClusteringToOrderBy(source, ingestNode, catalogClustering);
+
+ // Validate the source statement. Validates the ORDER BY pushed down in the above step.
+ // Because of the non-standard Druid semantics, we can't define the target type: we don't know
+ // the target columns yet, and we can't infer types when they must come from the SELECT.
+ // Normally, the target type is known, and is pushed into the SELECT. In Druid, the SELECT
+ // usually defines the target types, unless the catalog says otherwise. Since catalog entries
+ // are optional, we don't know the target type until we validate the SELECT. (Also, we won't
+ // know names and we match by name.) Thus, we'd have to validate (to know names and types)
+ // to get the target types, but we need the target types to validate. Catch-22. So, we punt.
+ final SqlValidatorScope scope;
+ if (source instanceof SqlSelect) {
+ final SqlSelect sqlSelect = (SqlSelect) source;
+ validateSelect(sqlSelect, unknownType);
+ scope = null;
+ } else {
+ scope = scopes.get(source);
+ validateQuery(source, scope, unknownType);
+ }
+
+ final SqlValidatorNamespace sourceNamespace = namespaces.get(source);
+ final RelRecordType sourceType = (RelRecordType) sourceNamespace.getRowType();
+
+ // Validate the __time column
+ int timeColumnIndex = sourceType.getFieldNames().indexOf(Columns.TIME_COLUMN);
+ if (timeColumnIndex != -1) {
+ validateTimeColumn(sourceType, timeColumnIndex);
+ }
+
+ // Validate clustering against the SELECT row type. Clustering has additional
+ // constraints beyond what was validated for the pushed-down ORDER BY.
+ // Though we pushed down clustering above, only now can we validate it after
+ // we've determined the SELECT row type.
+ validateClustering(sourceType, ingestNode, catalogClustering);
+
+ // Determine the output (target) schema.
+ final RelDataType targetType = validateTargetType(scope, target, insert, sourceType, tableMetadata);
+
+ // Set the type for the INSERT/REPLACE node
+ setValidatedNodeType(insert, targetType);
+
+ // Segment size
+ if (tableMetadata != null && !validatorContext.queryContextMap().containsKey(CTX_ROWS_PER_SEGMENT)) {
+ final Integer targetSegmentRows = tableMetadata.targetSegmentRows();
+ if (targetSegmentRows != null) {
+ validatorContext.queryContextMap().put(CTX_ROWS_PER_SEGMENT, targetSegmentRows);
+ }
+ }
+ }
+
+ /**
+ * Validate the target table. Druid {@code INSERT/REPLACE} can create a new datasource,
+ * or insert into an existing one. If the target exists, it must be a datasource. If it
+ * does not exist, the target must be in the datasource schema, normally "druid".
+ */
+ private DatasourceTable validateInsertTarget(
+ final SqlValidatorNamespace targetNamespace,
+ final IdentifierNamespace insertNs,
+ final String operationName
+ )
+ {
+ // Get the target table ID
+ final SqlIdentifier destId = insertNs.getId();
+ if (destId.names.isEmpty()) {
+ // I don't think this can happen, but include a branch for it just in case.
+ throw new IAE("%s requires a target table.", operationName);
+ }
+
+ // Druid does not support 3+ part names.
+ final int n = destId.names.size();
+ if (n > 2) {
+ throw new IAE("Druid does not support 3+ part names: [%s]", destId);
+ }
+ String tableName = destId.names.get(n - 1);
+
+ // If this is a 2-part name, the first part must be the datasource schema.
+ if (n == 2 && !validatorContext.druidSchemaName().equals(destId.names.get(0))) {
+ throw new IAE("Cannot %s into [%s] because the table is not in the 'druid' schema",
+ operationName,
+ destId
+ );
+ }
+ try {
+ // Try to resolve the table. Will fail if this is an INSERT into a new table.
Review Comment:
This causes MSQ ingestion to fail on non-catalog tables.
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1475419052
##########
sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlUnparseTest.java:
##########
@@ -58,7 +58,7 @@ public void testUnparseReplaceAll() throws ParseException
+ "OVERWRITE ALL\n"
+ "SELECT *\n"
+ " FROM \"foo\"\n"
- + "PARTITIONED BY ALL TIME "
+ + "PARTITIONED BY 'ALL TIME' "
Review Comment:
Changed back
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15711:
URL: https://github.com/apache/druid/pull/15711#issuecomment-2033314331
This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1475419510
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java:
##########
@@ -133,6 +141,41 @@
this.context = config.getContext();
this.connectionConfig = connConfig(context);
this.typeSystem = config.getTypeSystem();
+
+ // With the catalog, schemas can provide functions.
+ // Add the necessary indirection. The type factory used here
+ // is the Druid one, since the per-query one is not yet available
+ // here. Nor are built-in function associated with per-query types.
+ this.operatorTable = new ChainedSqlOperatorTable(
Review Comment:
removed
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1475419200
##########
sql/src/main/codegen/includes/common.ftl:
##########
@@ -18,58 +18,51 @@
*/
// Using fully qualified name for Pair class, since Calcite also has a same class name being used in the Parser.jj
-org.apache.druid.java.util.common.Pair<Granularity, String> PartitionGranularity() :
+SqlNode PartitionGranularity() :
{
SqlNode e;
- Granularity granularity;
- String unparseString;
+ SqlNode result;
}
{
(
<HOUR>
{
- granularity = Granularities.HOUR;
- unparseString = "HOUR";
+ result = SqlLiteral.createCharString(DruidSqlParserUtils.HOUR_GRAIN, getPos());
Review Comment:
using `SqlLiteral.createSymbol` as you suggested
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "kgyrtkirk (via GitHub)" <gi...@apache.org>.
kgyrtkirk commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1474088487
##########
sql/src/main/codegen/includes/common.ftl:
##########
@@ -18,58 +18,51 @@
*/
// Using fully qualified name for Pair class, since Calcite also has a same class name being used in the Parser.jj
-org.apache.druid.java.util.common.Pair<Granularity, String> PartitionGranularity() :
+SqlNode PartitionGranularity() :
{
SqlNode e;
- Granularity granularity;
- String unparseString;
+ SqlNode result;
}
{
(
<HOUR>
{
- granularity = Granularities.HOUR;
- unparseString = "HOUR";
+ result = SqlLiteral.createCharString(DruidSqlParserUtils.HOUR_GRAIN, getPos());
Review Comment:
I wonder if it would be possible to use `SqlLiteral.createSymbol` here instead; that could remove the need for the string based matching as well...
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java:
##########
@@ -133,6 +141,41 @@ public CalcitePlanner(FrameworkConfig config)
this.context = config.getContext();
this.connectionConfig = connConfig(context);
this.typeSystem = config.getTypeSystem();
+
+ // With the catalog, schemas can provide functions.
+ // Add the necessary indirection. The type factory used here
+ // is the Druid one, since the per-query one is not yet available
+ // here. Nor are built-in function associated with per-query types.
+ this.operatorTable = new ChainedSqlOperatorTable(
Review Comment:
I wonder if this is new functionality - could it be in a separate PR?
##########
sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlUnparseTest.java:
##########
@@ -58,7 +58,7 @@ public void testUnparseReplaceAll() throws ParseException
+ "OVERWRITE ALL\n"
+ "SELECT *\n"
+ " FROM \"foo\"\n"
- + "PARTITIONED BY ALL TIME "
+ + "PARTITIONED BY 'ALL TIME' "
Review Comment:
is the result of the unparse valid ?
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1475419327
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -19,38 +19,762 @@
package org.apache.druid.sql.calcite.planner;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.prepare.BaseDruidSqlValidator;
import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelRecordType;
import org.apache.calcite.runtime.CalciteContextException;
import org.apache.calcite.runtime.CalciteException;
+import org.apache.calcite.sql.SqlBasicCall;
import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlInsert;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlNumericLiteral;
import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.BasicSqlType;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.validate.IdentifierNamespace;
+import org.apache.calcite.sql.validate.OrderByScope;
+import org.apache.calcite.sql.validate.SqlValidatorException;
+import org.apache.calcite.sql.validate.SqlValidatorImpl;
+import org.apache.calcite.sql.validate.SqlValidatorNamespace;
import org.apache.calcite.sql.validate.SqlValidatorScope;
+import org.apache.calcite.sql.validate.SqlValidatorTable;
+import org.apache.calcite.sql.validate.ValidatorShim;
+import org.apache.calcite.util.Pair;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.error.InvalidSqlInput;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.PeriodGranularity;
+import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
+import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
+import org.apache.druid.sql.calcite.parser.DruidSqlParserUtils;
import org.apache.druid.sql.calcite.run.EngineFeature;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
/**
* Druid extended SQL validator. (At present, it doesn't actually
* have any extensions yet, but it will soon.)
*/
class DruidSqlValidator extends BaseDruidSqlValidator
{
+ private static final Pattern UNNAMED_COLUMN_PATTERN = Pattern.compile("^EXPR\\$\\d+$", Pattern.CASE_INSENSITIVE);
+
+ // Copied here from MSQE since that extension is not visible here.
+ public static final String CTX_ROWS_PER_SEGMENT = "msqRowsPerSegment";
+
+ public interface ValidatorContext
+ {
+ Map<String, Object> queryContextMap();
+ CatalogResolver catalog();
+ String druidSchemaName();
+ ObjectMapper jsonMapper();
+ }
+
+ private final ValidatorContext validatorContext;
private final PlannerContext plannerContext;
protected DruidSqlValidator(
- SqlOperatorTable opTab,
- CalciteCatalogReader catalogReader,
- JavaTypeFactory typeFactory,
- Config validatorConfig,
- PlannerContext plannerContext
+ final SqlOperatorTable opTab,
+ final CalciteCatalogReader catalogReader,
+ final JavaTypeFactory typeFactory,
+ final Config validatorConfig,
+ final PlannerContext plannerContext,
+ final ValidatorContext validatorContext
)
{
super(opTab, catalogReader, typeFactory, validatorConfig);
this.plannerContext = plannerContext;
+ this.validatorContext = validatorContext;
+ }
+
+ /**
+ * Druid-specific validation for an INSERT statement. In Druid, the columns are
+ * matched by name. A datasource, by default, allows the insertion of arbitrary columns,
+ * but the catalog may enforce a strict schema (all columns must exist). Destination
+ * types are set by the catalog, where available, else by the query.
+ * <p>
+ * The Druid {@code INSERT} statement is non-standard in a variety of ways:
+ * <ul>
+ * <li>Allows the target table to not yet exist. Instead, {@code INSERT}
+ * creates it.</li>
+ * <li>Does not allow specifying the list of columns:
+ * {@code INSERT INTO dst (a, b, c) ...}</li>
+ * <li>When given without target columns (the only form allowed), columns are
+ * not matched by schema position as in standard SQL, but rather by name.</li>
+ * <li>There is no requirement that the target columns already exist. In fact,
+ * even if the target column exists, any existing type is ignored if not specified
+ * in the catalog.</li>
+ * <li>The source can only be a {@code SELECT} statement, not {@code VALUES}.</li>
+ * <li>Types thus propagate upwards from the {@code SELECT} to the the target
+ * table. Standard SQL says that types propagate down from the target table to the
+ * source.</li>
+ * <li>The __time column is special in multiple ways.</li>
+ * <li>Includes the {@code CLUSTERED BY} and {@code PARTITIONED BY} clauses.</li>
+ * </ul>
+ * The result is that the validation for the Druid {@code INSERT} is wildly customized
+ * relative to standard SQL.
+ */
+ // TODO: Ensure the source and target are not the same
+ @Override
+ public void validateInsert(final SqlInsert insert)
+ {
+ final DruidSqlIngest ingestNode = (DruidSqlIngest) insert;
+ if (insert.isUpsert()) {
+ throw new IAE("UPSERT is not supported.");
+ }
+
+ // SQL-style INSERT INTO dst (a, b, c) is not (yet) supported.
+ final String operationName = insert.getOperator().getName();
+ if (insert.getTargetColumnList() != null) {
+ throw new IAE("%s with a target column list is not supported.", operationName);
+ }
+
+ // The target namespace is both the target table ID and the row type for that table.
+ final SqlValidatorNamespace targetNamespace = getNamespace(insert);
+ final IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+
+ // The target is a new or existing datasource.
+ final DatasourceTable table = validateInsertTarget(targetNamespace, insertNs, operationName);
+ final SqlValidatorTable target = insertNs.resolve().getTable();
+
+ // An existing datasource may have metadata.
+ final DatasourceFacade tableMetadata = table == null ? null : table.effectiveMetadata().catalogMetadata();
+
+ // Validate segment granularity, which depends on nothing else.
+ validateSegmentGranularity(operationName, ingestNode, tableMetadata);
+
+ // The source must be a SELECT
+ final SqlNode source = insert.getSource();
+ ensureNoOrderBy(source, operationName);
+
+ // Convert CLUSTERED BY, or the catalog equivalent, to an ORDER BY clause
+ final SqlNodeList catalogClustering = convertCatalogClustering(tableMetadata);
+ rewriteClusteringToOrderBy(source, ingestNode, catalogClustering);
+
+ // Validate the source statement. Validates the ORDER BY pushed down in the above step.
+ // Because of the non-standard Druid semantics, we can't define the target type: we don't know
+ // the target columns yet, and we can't infer types when they must come from the SELECT.
+ // Normally, the target type is known, and is pushed into the SELECT. In Druid, the SELECT
+ // usually defines the target types, unless the catalog says otherwise. Since catalog entries
+ // are optional, we don't know the target type until we validate the SELECT. (Also, we won't
+ // know names and we match by name.) Thus, we'd have to validate (to know names and types)
+ // to get the target types, but we need the target types to validate. Catch-22. So, we punt.
+ final SqlValidatorScope scope;
+ if (source instanceof SqlSelect) {
+ final SqlSelect sqlSelect = (SqlSelect) source;
+ validateSelect(sqlSelect, unknownType);
+ scope = null;
+ } else {
+ scope = scopes.get(source);
+ validateQuery(source, scope, unknownType);
+ }
+
+ final SqlValidatorNamespace sourceNamespace = namespaces.get(source);
+ final RelRecordType sourceType = (RelRecordType) sourceNamespace.getRowType();
+
+ // Validate the __time column
+ int timeColumnIndex = sourceType.getFieldNames().indexOf(Columns.TIME_COLUMN);
+ if (timeColumnIndex != -1) {
+ validateTimeColumn(sourceType, timeColumnIndex);
+ }
+
+ // Validate clustering against the SELECT row type. Clustering has additional
+ // constraints beyond what was validated for the pushed-down ORDER BY.
+ // Though we pushed down clustering above, only now can we validate it after
+ // we've determined the SELECT row type.
+ validateClustering(sourceType, ingestNode, catalogClustering);
+
+ // Determine the output (target) schema.
+ final RelDataType targetType = validateTargetType(scope, target, insert, sourceType, tableMetadata);
+
+ // Set the type for the INSERT/REPLACE node
+ setValidatedNodeType(insert, targetType);
+
+ // Segment size
+ if (tableMetadata != null && !validatorContext.queryContextMap().containsKey(CTX_ROWS_PER_SEGMENT)) {
+ final Integer targetSegmentRows = tableMetadata.targetSegmentRows();
+ if (targetSegmentRows != null) {
+ validatorContext.queryContextMap().put(CTX_ROWS_PER_SEGMENT, targetSegmentRows);
+ }
+ }
+ }
+
+ /**
+ * Validate the target table. Druid {@code INSERT/REPLACE} can create a new datasource,
+ * or insert into an existing one. If the target exists, it must be a datasource. If it
+ * does not exist, the target must be in the datasource schema, normally "druid".
+ */
+ private DatasourceTable validateInsertTarget(
+ final SqlValidatorNamespace targetNamespace,
+ final IdentifierNamespace insertNs,
+ final String operationName
+ )
+ {
+ // Get the target table ID
+ final SqlIdentifier destId = insertNs.getId();
+ if (destId.names.isEmpty()) {
+ // I don't think this can happen, but include a branch for it just in case.
+ throw new IAE("%s requires a target table.", operationName);
+ }
+
+ // Druid does not support 3+ part names.
+ final int n = destId.names.size();
+ if (n > 2) {
+ throw new IAE("Druid does not support 3+ part names: [%s]", destId);
+ }
+ String tableName = destId.names.get(n - 1);
+
+ // If this is a 2-part name, the first part must be the datasource schema.
+ if (n == 2 && !validatorContext.druidSchemaName().equals(destId.names.get(0))) {
+ throw new IAE("Cannot %s into [%s] because the table is not in the 'druid' schema",
+ operationName,
+ destId
+ );
+ }
+ try {
+ // Try to resolve the table. Will fail if this is an INSERT into a new table.
Review Comment:
fixed
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1473616512
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java:
##########
@@ -133,6 +141,41 @@
this.context = config.getContext();
this.connectionConfig = connConfig(context);
this.typeSystem = config.getTypeSystem();
+
+ // With the catalog, schemas can provide functions.
+ // Add the necessary indirection. The type factory used here
+ // is the Druid one, since the per-query one is not yet available
+ // here. Nor are built-in function associated with per-query types.
+ this.operatorTable = new ChainedSqlOperatorTable(
Review Comment:
Need to check on this.
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15711: Integrate catalog schema validation into planner -WIP
URL: https://github.com/apache/druid/pull/15711
--
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] Integrate catalog schema validation into planner -WIP (druid)
Posted by "zachjsh (via GitHub)" <gi...@apache.org>.
zachjsh commented on code in PR #15711:
URL: https://github.com/apache/druid/pull/15711#discussion_r1475417927
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java:
##########
@@ -133,6 +141,41 @@ public CalcitePlanner(FrameworkConfig config)
this.context = config.getContext();
this.connectionConfig = connConfig(context);
this.typeSystem = config.getTypeSystem();
+
+ // With the catalog, schemas can provide functions.
+ // Add the necessary indirection. The type factory used here
+ // is the Druid one, since the per-query one is not yet available
+ // here. Nor are built-in function associated with per-query types.
+ this.operatorTable = new ChainedSqlOperatorTable(
Review Comment:
Thanks! removed this
--
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