You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2023/01/18 01:34:54 UTC
[GitHub] [druid] github-code-scanning[bot] commented on a diff in pull request #13686: Integrate the catalog with the Calcite planner
github-code-scanning[bot] commented on code in PR #13686:
URL: https://github.com/apache/druid/pull/13686#discussion_r1073002223
##########
sql/src/main/java/org/apache/druid/sql/calcite/table/DatasourceTable.java:
##########
@@ -181,9 +300,10 @@
public String toString()
{
// Don't include the row signature: it is the same as in
- // physicalMetadata.
- return "DruidTable{" +
- physicalMetadata +
+ // effectiveMetadata.
+ return "DruidTable{physicalMetadata=" +
+ physicalMetadata == null ? "null" : physicalMetadata.toString() +
Review Comment:
## Useless null check
This check is useless, since [... + ...](1) always is non-null.
[Show more details](https://github.com/apache/druid/security/code-scanning/3693)
##########
server/src/main/java/org/apache/druid/catalog/model/TableDefn.java:
##########
@@ -85,10 +85,10 @@
public void validate(ResolvedTable table)
{
validate(table.properties(), table.jsonMapper());
- validateColumns(table.spec().columns());
+ validateColumns(table.spec().columns(), table.jsonMapper());
}
- public void validateColumns(List<ColumnSpec> columns)
+ public void validateColumns(List<ColumnSpec> columns, ObjectMapper jsonMapper)
Review Comment:
## Useless parameter
The parameter 'jsonMapper' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/3590)
##########
server/src/main/java/org/apache/druid/catalog/model/TypeParser.java:
##########
@@ -0,0 +1,281 @@
+/*
+ * 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.catalog.model;
+
+import org.apache.druid.catalog.model.MeasureTypes.MeasureType;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Parser for the SQL type format used in the catalog.
+ * <ul>
+ * <li>{@code null} - A null type means "use whatever Druid wants."</li>
+ * <li>Scalar - {@code VARCHAR}, {@code BIGINT}, etc.</li>
+ * <li>{@code TIMESTAMP('<grain'>} - truncated timestamp</li>
+ * <li>{@code fn(<type>)} - aggregate measure such as @{code COUNT},
+ * {@code COUNT()} or {@code SUM(BIGINT)}.</li>
+ * </ul>
+ */
+public class TypeParser
+{
+ public static final ParsedType ANY_TYPE = new ParsedType(null, ParsedType.Kind.ANY, null, null);
+ public static final ParsedType TIME_TYPE = new ParsedType(null, ParsedType.Kind.TIME, null, null);
+
+ public static class ParsedType
+ {
+ public enum Kind
+ {
+ ANY,
+ TIME,
+ DIMENSION,
+ MEASURE
+ }
+
+ private final String type;
+ private final Kind kind;
+ private final String timeGrain;
+ private final MeasureType measureType;
+
+ public ParsedType(String type, Kind kind, String timeGrain, MeasureType measureType)
+ {
+ this.type = type;
+ this.kind = kind;
+ this.timeGrain = timeGrain;
+ this.measureType = measureType;
+ }
+
+ public String type()
+ {
+ return type;
+ }
+
+ public Kind kind()
+ {
+ return kind;
+ }
+
+ public String timeGrain()
+ {
+ return timeGrain;
+ }
+
+ public MeasureType measure()
+ {
+ return measureType;
+ }
+ }
+
+ private static class Token
+ {
+ private enum Kind
+ {
+ SYMBOL,
+ OPEN,
+ STRING,
+ COMMA,
+ CLOSE
+ }
+
+ private final Kind kind;
+ private final String value;
+
+ public Token(Kind kind, String value)
+ {
+ this.kind = kind;
+ this.value = value;
+ }
+ }
+
+ private String input;
+ private int posn;
+
+ private TypeParser(String type)
+ {
+ this.input = type;
+ }
+
+ public static ParsedType parse(String type)
+ {
+ if (type == null) {
+ return ANY_TYPE;
+ }
+ return new TypeParser(type).parse();
+ }
+
+ private ParsedType parse()
+ {
+ Token token = parseToken();
+ if (token == null) {
+ return ANY_TYPE;
+ }
+ if (token.kind != Token.Kind.SYMBOL) {
+ throw new IAE("Invalid type name");
+ }
+ final String baseName = StringUtils.toUpperCase(token.value);
+ boolean isTime = Columns.isTimestamp(baseName);
+ token = parseToken();
+ if (token == null) {
+ if (isTime) {
+ return new ParsedType(baseName, ParsedType.Kind.TIME, null, null);
+ } else if (Columns.isScalar(baseName)) {
+ return new ParsedType(baseName, ParsedType.Kind.DIMENSION, null, null);
+ }
+ return analyzeAggregate(baseName, Collections.emptyList());
+ }
+ if (token.kind != Token.Kind.OPEN) {
+ throw new IAE("Invalid type name");
+ }
+ if (!isTime && Columns.isScalar(baseName)) {
+ throw new IAE("Invalid type name");
+ }
+ List<Token> args = new ArrayList<>();
+ token = parseToken();
+ if (token == null) {
+ throw new IAE("Invalid type name");
+ }
+ if (token.kind != Token.Kind.CLOSE) {
+ if (token.kind != Token.Kind.SYMBOL && token.kind != Token.Kind.STRING) {
+ throw new IAE("Invalid type name");
+ }
+ args.add(token);
+ while (true) {
+ token = parseToken();
+ if (token == null) {
+ throw new IAE("Invalid type name");
+ }
+ if (token.kind == Token.Kind.CLOSE) {
+ break;
+ }
+ if (token.kind != Token.Kind.COMMA) {
+ throw new IAE("Invalid type name");
+ }
+ token = parseToken();
+ if (token.kind != Token.Kind.SYMBOL && token.kind != Token.Kind.STRING) {
+ throw new IAE("Invalid type name");
+ }
+ args.add(token);
+ }
+ }
+ token = parseToken();
+ if (token != null) {
+ throw new IAE("Invalid type name");
+ }
+ if (isTime) {
+ return analyzeTimestamp(baseName, args);
+ } else {
+ return analyzeAggregate(baseName, args);
+ }
+ }
+
+ private ParsedType analyzeTimestamp(String baseName, List<Token> args)
Review Comment:
## Useless parameter
The parameter 'baseName' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/3591)
##########
server/src/main/java/org/apache/druid/catalog/model/facade/DatasourceFacade.java:
##########
@@ -38,10 +46,111 @@
{
private static final Logger LOG = new Logger(DatasourceFacade.class);
+ public static class ColumnFacade
+ {
+ public enum Kind
+ {
+ ANY,
+ TIME,
+ DIMENSION,
+ MEASURE
+ }
+
+ private final ColumnSpec spec;
+ private final ParsedType type;
+
+ public ColumnFacade(ColumnSpec spec)
+ {
+ this.spec = spec;
+ if (Columns.isTimeColumn(spec.name()) && spec.sqlType() == null) {
+ // For __time only, force a type if type is null.
+ this.type = TypeParser.TIME_TYPE;
+ } else {
+ this.type = TypeParser.parse(spec.sqlType());
+ }
+ }
+
+ public ColumnSpec spec()
+ {
+ return spec;
+ }
+
+ public ParsedType type()
+ {
+ return type;
+ }
+
+ public boolean hasType()
+ {
+ return type.kind() != ParsedType.Kind.ANY;
+ }
+
+ public boolean isTime()
+ {
+ return type.kind() == ParsedType.Kind.TIME;
+ }
+
+ public boolean isMeasure()
+ {
+ return type.kind() == ParsedType.Kind.MEASURE;
+ }
+
+ public ColumnType druidType()
+ {
+ switch (type.kind()) {
+ case DIMENSION:
+ return Columns.druidType(spec.sqlType());
+ case TIME:
+ return ColumnType.LONG;
+ case MEASURE:
+ return type.measure().storageType;
+ default:
+ return null;
+ }
+ }
+
+ public String sqlStorageType()
+ {
+ if (isTime()) {
+ // Time is special: its storage type is BIGINT, but SQL requires the
+ // type of TIMESTAMP to allow insertion validation.
+ return Columns.TIMESTAMP;
+ } else {
+ return Columns.sqlType(druidType());
+ }
+ }
+
+ @Override
+ public String toString()
+ {
+ return "{spec=" + spec + ", type=" + type + "}";
Review Comment:
## Use of default toString()
Default toString(): ParsedType inherits toString() from Object, and so is not suitable for printing.
[Show more details](https://github.com/apache/druid/security/code-scanning/3597)
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -19,24 +19,702 @@
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.sql.SqlBasicCall;
+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.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.SqlConformance;
+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.curator.shaded.com.google.common.base.Objects;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.facade.DatasourceFacade.ColumnFacade;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.common.utils.IdUtils;
+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.table.DatasourceTable;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
/**
- * Druid extended SQL validator. (At present, it doesn't actually
- * have any extensions yet, but it will soon.)
+ * Druid extended SQL validator.
*/
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;
+
protected DruidSqlValidator(
- SqlOperatorTable opTab,
- CalciteCatalogReader catalogReader,
- JavaTypeFactory typeFactory,
- SqlConformance conformance)
+ final SqlOperatorTable opTab,
+ final CalciteCatalogReader catalogReader,
+ final JavaTypeFactory typeFactory,
+ final SqlConformance conformance,
+ final ValidatorContext validatorContext
+ )
{
super(opTab, catalogReader, typeFactory, conformance);
+ 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(SqlInsert insert)
+ {
+ 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.
+ 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);
+ IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+
+ // The target is a new or existing datasource.
+ DatasourceTable table = validateInsertTarget(targetNamespace, insertNs, operationName);
+
+ // An existing datasource may have metadata.
+ 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
+ SqlSelect select = validateInsertSelect(insert, operationName);
+
+ // Convert CLUSTERED BY, or the catalog equivalent, to an ORDER BY clause
+ SqlNodeList catalogClustering = convertCatalogClustering(tableMetadata);
+ rewriteClusteringToOrderBy(select, ingestNode, catalogClustering);
+
+ // Validate the source statement. Validates the ORDER BY pushed down in the above step.
+ validateSelect(select, unknownType);
+ SqlValidatorScope selectScope = getSelectScope(select);
+
+ SqlValidatorNamespace sourceNamespace = namespaces.get(select);
+ 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(select, selectScope, sourceType, timeColumnIndex, ingestNode, catalogClustering);
+
+ // Determine the output (target) schema.
+ RelDataType targetType = validateTargetType(insert, sourceType, tableMetadata);
+
+ // Set the type for the INSERT/REPLACE node
+ setValidatedNodeType(insert, targetType);
+
+ // Segment size
+ if (tableMetadata != null) {
+ 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
+ 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.
+ int n = destId.names.size();
+ if (n > 2) {
+ throw new IAE("Table name is undefined: %s", destId.toString());
+ }
+ 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 it is not a Druid datasource.",
+ operationName,
+ destId
+ );
+ }
+ try {
+ // Try to resolve the table. Will fail if this is an INSERT into a new table.
+ validateNamespace(targetNamespace, unknownType);
+ SqlValidatorTable target = insertNs.resolve().getTable();
+ try {
+ return target.unwrap(DatasourceTable.class);
+ }
+ catch (Exception e) {
+ throw new IAE("Cannot %s into %s: it is not a datasource", operationName, destId);
+ }
+ }
+ catch (CalciteContextException e) {
+ // Something failed. Let's make sure it was the table lookup.
+ // The check is kind of a hack, but its the best we can do given that Calcite
+ // didn't expect this non-SQL use case.
+ if (e.getCause() instanceof SqlValidatorException && e.getMessage().contains(StringUtils.format("Object '%s' not found", tableName))) {
+ // The catalog implementation may be "strict": and require that the target
+ // table already exists, rather than the default "lenient" mode that can
+ // create a new table.
+ if (validatorContext.catalog().ingestRequiresExistingTable()) {
+ throw new IAE("Cannot %s into %s because it does not exist", operationName, destId);
+ }
+ // New table. Validate the shape of the name.
+ IdUtils.validateId(operationName + " dataSource", tableName);
+ return null;
+ }
+ throw e;
+ }
+ }
+
+ private void validateSegmentGranularity(
+ final String operationName,
+ final DruidSqlIngest ingestNode,
+ final DatasourceFacade tableMetadata
+ )
+ {
+ final Granularity definedGranularity = tableMetadata == null ? null : tableMetadata.segmentGranularity();
+ if (definedGranularity != null) {
+ // Should already have been checked when creating the catalog entry
+ DruidSqlParserUtils.throwIfUnsupportedGranularityInPartitionedBy(definedGranularity);
+ }
+ final Granularity ingestionGranularity = DruidSqlParserUtils.convertSqlNodeToGranularity(ingestNode.getPartitionedBy());
+ if (ingestionGranularity != null) {
+ DruidSqlParserUtils.throwIfUnsupportedGranularityInPartitionedBy(ingestionGranularity);
+ }
+ final Granularity finalGranularity;
+ if (definedGranularity == null && ingestionGranularity == null) {
+ // Neither have a value: error
+ throw new IAE(
+ "%s statements must specify a PARTITIONED BY clause explicitly",
+ operationName
+ );
+ } else if (ingestionGranularity == null) {
+ // The query has no granularity: just apply the catalog granularity.
+ finalGranularity = definedGranularity;
+ } else if (definedGranularity == null) {
+ // The catalog has no granularity: apply the query value
+ finalGranularity = ingestionGranularity;
+ } else if (definedGranularity.equals(ingestionGranularity)) {
+ // Both have a setting. They have to be the same.
+ finalGranularity = definedGranularity;
+ } else {
+ throw new IAE(
+ "PARTITIONED BY mismatch. Catalog: [%s], query: [%s]",
+ granularityToSqlString(definedGranularity),
+ granularityToSqlString(ingestionGranularity)
+ );
+ }
+
+ // Note: though this is the validator, we cheat a bit and write the target
+ // granularity into the query context. Perhaps this step should be done
+ // during conversion, however, we've just worked out the granularity, so we
+ // do it here instead.
+ try {
+ validatorContext.queryContextMap().put(
+ DruidSqlInsert.SQL_INSERT_SEGMENT_GRANULARITY,
+ validatorContext.jsonMapper().writeValueAsString(finalGranularity)
+ );
+ }
+ catch (JsonProcessingException e) {
+ throw new IAE("Invalid PARTITIONED BY granularity");
+ }
+ }
+
+ private String granularityToSqlString(Granularity gran)
+ {
+ if (gran == null) {
+ return "NULL";
+ }
+ if (Granularities.ALL == gran) {
+ return "ALL TIME";
+ }
+ if (!(gran instanceof PeriodGranularity)) {
+ return gran.toString();
+ }
+ return ((PeriodGranularity) gran).getPeriod().toString();
+ }
+
+ private SqlSelect validateInsertSelect(
+ final SqlInsert insert,
+ final String operationName
+ )
+ {
+ SqlNode source = insert.getSource();
+
+ // The source SELECT cannot include an ORDER BY clause. Ordering is given
+ // by the CLUSTERED BY clause, if any.
+ // Check that an ORDER BY clause is not provided by the underlying query
+ SqlNodeList orderByList;
+ if (source instanceof SqlOrderBy) {
+ throw new IAE(
+ "Cannot have ORDER BY on %s %s statement, use CLUSTERED BY instead.",
+ "INSERT".equals(operationName) ? "an" : "a",
+ operationName
+ );
+ }
+
+ // Validate the query: but we only know how to do SELECT.
+ // As we do, get the scope for the query.
+ if (!source.isA(SqlKind.QUERY)) {
+ throw new IAE("Cannot execute %s.", source.getKind());
+ }
+ if (!(source instanceof SqlSelect)) {
+ throw new IAE(
+ "%s %s must include a SELECT statement",
+ "INSERT".equals(operationName) ? "An" : "A",
+ operationName
+ );
+ }
+ SqlSelect select = (SqlSelect) source;
+ orderByList = select.getOrderList();
+ if (orderByList != null && orderByList.size() != 0) {
+ throw new IAE(
+ "Cannot have ORDER BY on %s %s statement, use CLUSTERED BY instead.",
+ "INSERT".equals(operationName) ? "an" : "a",
+ operationName
+ );
+ }
+ return select;
+ }
+
+ private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+ if (CollectionUtils.isNullOrEmpty(keyCols)) {
+ return null;
+ }
+ SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+ for (ClusterKeySpec keyCol : keyCols) {
+ SqlIdentifier colIdent = new SqlIdentifier(
+ Collections.singletonList(keyCol.expr()),
+ null, SqlParserPos.ZERO,
+ Collections.singletonList(SqlParserPos.ZERO)
+ );
+ SqlNode keyNode;
+ if (keyCol.desc()) {
+ keyNode = SqlStdOperatorTable.DESC.createCall(SqlParserPos.ZERO, colIdent);
+ } else {
+ keyNode = colIdent;
+ }
+ keyNodes.add(keyNode);
+ }
+ return keyNodes;
+ }
+
+ /**
+ * 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(SqlSelect select, DruidSqlIngest ingestNode, SqlNodeList catalogClustering)
+ {
+ SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+ if (clusteredBy == null || clusteredBy.getList().isEmpty()) {
+ if (catalogClustering == null || catalogClustering.getList().isEmpty()) {
+ return;
+ }
+ clusteredBy = catalogClustering;
+ }
+
+ // 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.
+ select.setOrderBy(clusteredBy);
+ OrderByScope orderScope = ValidatorShim.newOrderByScope(scopes.get(select), clusteredBy, select);
+ try {
+ @SuppressWarnings("unchecked")
+ Map<SqlSelect, SqlValidatorScope> orderScopes = (Map<SqlSelect, SqlValidatorScope>) FieldUtils.getDeclaredField(SqlValidatorImpl.class, "orderScopes", true).get(this);
+ orderScopes.put(select, orderScope);
+ }
+ catch (Exception e) {
+ throw new ISE(e, "orderScopes is not accessible");
+ }
+ }
+
+ private void validateTimeColumn(RelRecordType sourceType, int timeColumnIndex)
+ {
+ RelDataTypeField timeCol = sourceType.getFieldList().get(timeColumnIndex);
+ RelDataType timeColType = timeCol.getType();
+ if (timeColType instanceof BasicSqlType) {
+ BasicSqlType timeColSqlType = (BasicSqlType) timeColType;
+ SqlTypeName timeColSqlTypeName = timeColSqlType.getSqlTypeName();
+ if (timeColSqlTypeName == SqlTypeName.BIGINT || timeColSqlTypeName == SqlTypeName.TIMESTAMP) {
+ return;
+ }
+ }
+ throw new IAE(
+ "Invalid %s column type %s: must be BIGINT or TIMESTAMP",
+ timeCol.getName(),
+ timeColType.toString()
+ );
+ }
+
+ /**
+ * Verify clustering which can come from the query, the catalog or both. If both,
+ * the two must match. In either case, the cluster keys must be present in the SELECT
+ * clause. The {@code __time} column cannot be included.
+ * @param source
+ */
+ private void validateClustering(
+ final SqlSelect source,
+ final SqlValidatorScope selectScope,
+ final RelRecordType sourceType,
+ final int timeColumnIndex,
+ final DruidSqlIngest ingestNode,
+ final SqlNodeList catalogClustering
+ )
+ {
+ final SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+
+ // Validate both the catalog and query definitions if present. This ensures
+ // that things are sane if we later check that the two are identical.
+ if (clusteredBy != null) {
+ validateClusteredBy(source, sourceType, timeColumnIndex, selectScope, clusteredBy);
+ }
+ if (catalogClustering != null) {
+ // Catalog defines the key columns. Verify that they are present in the query.
+ validateClusteredBy(source, sourceType, timeColumnIndex, selectScope, catalogClustering);
+ }
+ if (clusteredBy != null && catalogClustering != null) {
+ // Both the query and catalog have keys.
+ verifyQueryClusterByMatchesCatalog(sourceType, catalogClustering, clusteredBy);
+ }
+ }
+
+ /**
+ * Validate the CLUSTERED BY list. Members can be any of the following:
+ * <p>
+ * {@code CLUSTERED BY [<ordinal> | <id> | <expr>] DESC?}
+ * <p>
+ * Ensure that each id exists. Ensure each column is included only once.
+ * For an expression, just ensure it is valid; we don't check for duplicates.
+ * <p>
+ * Once the keys are validated, update the underlying {@code SELECT} statement
+ * to include the {@code CLUSTERED BY} as the {@code ORDER BY} clause.
+ */
+ private void validateClusteredBy(
+ final SqlSelect source,
Review Comment:
## Useless parameter
The parameter 'source' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/3690)
##########
extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sql/LiveCatalogResolver.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.catalog.sql;
+
+import org.apache.druid.catalog.model.ColumnSpec;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.ResolvedTable;
+import org.apache.druid.catalog.model.TableId;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.table.DatasourceDefn;
+import org.apache.druid.catalog.sync.MetadataCatalog;
+import org.apache.druid.query.TableDataSource;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveColumnMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.EffectiveMetadata;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+
+import javax.inject.Inject;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+public class LiveCatalogResolver implements CatalogResolver
+{
+ public static final String TYPE = "catalog";
+
+ private final MetadataCatalog catalog;
+
+ @Inject
+ public LiveCatalogResolver(final MetadataCatalog catalog)
+ {
+ this.catalog = catalog;
+ }
+
+ private DatasourceFacade datasourceSpec(String name)
+ {
+ TableId tableId = TableId.datasource(name);
+ ResolvedTable table = catalog.resolveTable(tableId);
+ if (table == null) {
+ return null;
+ }
+ if (!DatasourceDefn.isDatasource(table)) {
+ return null;
+ }
+ return new DatasourceFacade(table);
+ }
+
+ /**
+ * Create a {@link DruidTable} based on the physical segments, catalog entry, or both.
+ */
+ @Override
+ public DruidTable resolveDatasource(String name, PhysicalDatasourceMetadata dsMetadata)
+ {
+ DatasourceFacade dsSpec = datasourceSpec(name);
+
+ // No catalog metadata. If there is no physical metadata, then the
+ // datasource does not exist. Else, if there is physical metadata, the
+ // datasource is based entirely on the physical information.
+ if (dsSpec == null) {
+ return dsMetadata == null ? null : new DatasourceTable(dsMetadata);
+ }
+ if (dsMetadata == null) {
+ // Datasource exists only in the catalog: no physical segments.
+ return emptyDatasource(name, dsSpec);
+ } else {
+ // Datasource exists as both segments and a catalog entry.
+ return mergeDatasource(name, dsMetadata, dsSpec);
+ }
+ }
+
+ private DruidTable emptyDatasource(String name, DatasourceFacade dsSpec)
+ {
+ RowSignature.Builder builder = RowSignature.builder();
+ Map<String, EffectiveColumnMetadata> columns = new HashMap<>();
+ boolean hasTime = false;
+ for (ColumnSpec col : dsSpec.columns()) {
+ EffectiveColumnMetadata colMetadata = columnFromCatalog(col, null);
+ if (colMetadata.name().equals(Columns.TIME_COLUMN)) {
+ hasTime = true;
+ }
+ builder.add(col.name(), colMetadata.druidType());
+ columns.put(col.name(), colMetadata);
+ }
+ if (!hasTime) {
+ columns.put(Columns.TIME_COLUMN, new EffectiveColumnMetadata(
+ Columns.TIME_COLUMN,
+ ColumnType.LONG
+ ));
+ builder = RowSignature.builder()
+ .add(Columns.TIME_COLUMN, ColumnType.LONG)
+ .addAll(builder.build());
+ }
+
+ final PhysicalDatasourceMetadata mergedMetadata = new PhysicalDatasourceMetadata(
+ new TableDataSource(name),
+ builder.build(),
+ true, // Can join to an empty table
+ false // Cannot broadcast an empty table
+ );
+ return new DatasourceTable(
+ mergedMetadata.rowSignature(),
+ mergedMetadata,
+ new EffectiveMetadata(dsSpec, columns, true)
+ );
+ }
+
+ private EffectiveColumnMetadata columnFromCatalog(ColumnSpec col, ColumnType physicalType)
+ {
+ ColumnType type = Columns.druidType(col.sqlType());
+ if (type != null) {
+ // Use the type that the user provided.
+ } else if (physicalType == null) {
+ // Corner case: the user has defined a column in the catalog, has
+ // not specified a type (meaning the user wants Druid to decide), but
+ // there is no data at this moment. Guess String as the type for the
+ // null values. If new segments appear between now and execution, we'll
+ // convert the values to string, which is always safe.
+ type = ColumnType.STRING;
+ } else {
+ type = physicalType;
+ }
+ return new EffectiveColumnMetadata(col.name(), type);
+ }
+
+ private DruidTable mergeDatasource(
+ final String name,
Review Comment:
## Useless parameter
The parameter 'name' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/3689)
##########
extensions-core/druid-catalog/src/test/java/org/apache/druid/catalog/sql/LiveCatalogTest.java:
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.catalog.sql;
+
+import org.apache.druid.catalog.CatalogException.DuplicateKeyException;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.TableMetadata;
+import org.apache.druid.catalog.model.table.TableBuilder;
+import org.apache.druid.catalog.storage.CatalogStorage;
+import org.apache.druid.catalog.storage.CatalogTests;
+import org.apache.druid.catalog.sync.LocalMetadataCatalog;
+import org.apache.druid.catalog.sync.MetadataCatalog;
+import org.apache.druid.metadata.TestDerbyConnector;
+import org.apache.druid.query.TableDataSource;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata;
+import org.apache.druid.sql.calcite.table.DruidTable;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.fail;
+
+/**
+ * Test for the datasource resolution aspects of the live catalog resolver.
+ * Too tedious to test the insert resolution in its current state.
+ */
+public class LiveCatalogTest
+{
+ @Rule
+ public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule = new TestDerbyConnector.DerbyConnectorRule();
+
+ private CatalogTests.DbFixture dbFixture;
+ private CatalogStorage storage;
+ private CatalogResolver resolver;
+
+ @Before
+ public void setUp()
+ {
+ dbFixture = new CatalogTests.DbFixture(derbyConnectorRule);
+ storage = dbFixture.storage;
+ MetadataCatalog catalog = new LocalMetadataCatalog(storage, storage.schemaRegistry());
+ resolver = new LiveCatalogResolver(catalog);
+ }
+
+ @After
+ public void tearDown()
+ {
+ CatalogTests.tearDown(dbFixture);
+ }
+
+ private void createTableMetadata(TableMetadata table)
+ {
+ try {
+ storage.tables().create(table);
+ }
+ catch (DuplicateKeyException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Populate the catalog with a few items using the REST resource.
+ * @throws DuplicateKeyException
Review Comment:
## Javadoc has impossible 'throws' tag
Javadoc for populateCatalog claims to throw DuplicateKeyException but this is impossible.
[Show more details](https://github.com/apache/druid/security/code-scanning/3688)
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java:
##########
@@ -19,24 +19,702 @@
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.sql.SqlBasicCall;
+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.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.SqlConformance;
+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.curator.shaded.com.google.common.base.Objects;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.facade.DatasourceFacade.ColumnFacade;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.common.utils.IdUtils;
+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.table.DatasourceTable;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
/**
- * Druid extended SQL validator. (At present, it doesn't actually
- * have any extensions yet, but it will soon.)
+ * Druid extended SQL validator.
*/
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;
+
protected DruidSqlValidator(
- SqlOperatorTable opTab,
- CalciteCatalogReader catalogReader,
- JavaTypeFactory typeFactory,
- SqlConformance conformance)
+ final SqlOperatorTable opTab,
+ final CalciteCatalogReader catalogReader,
+ final JavaTypeFactory typeFactory,
+ final SqlConformance conformance,
+ final ValidatorContext validatorContext
+ )
{
super(opTab, catalogReader, typeFactory, conformance);
+ 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(SqlInsert insert)
+ {
+ 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.
+ 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);
+ IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+
+ // The target is a new or existing datasource.
+ DatasourceTable table = validateInsertTarget(targetNamespace, insertNs, operationName);
+
+ // An existing datasource may have metadata.
+ 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
+ SqlSelect select = validateInsertSelect(insert, operationName);
+
+ // Convert CLUSTERED BY, or the catalog equivalent, to an ORDER BY clause
+ SqlNodeList catalogClustering = convertCatalogClustering(tableMetadata);
+ rewriteClusteringToOrderBy(select, ingestNode, catalogClustering);
+
+ // Validate the source statement. Validates the ORDER BY pushed down in the above step.
+ validateSelect(select, unknownType);
+ SqlValidatorScope selectScope = getSelectScope(select);
+
+ SqlValidatorNamespace sourceNamespace = namespaces.get(select);
+ 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(select, selectScope, sourceType, timeColumnIndex, ingestNode, catalogClustering);
+
+ // Determine the output (target) schema.
+ RelDataType targetType = validateTargetType(insert, sourceType, tableMetadata);
+
+ // Set the type for the INSERT/REPLACE node
+ setValidatedNodeType(insert, targetType);
+
+ // Segment size
+ if (tableMetadata != null) {
+ 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
+ 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.
+ int n = destId.names.size();
+ if (n > 2) {
+ throw new IAE("Table name is undefined: %s", destId.toString());
+ }
+ 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 it is not a Druid datasource.",
+ operationName,
+ destId
+ );
+ }
+ try {
+ // Try to resolve the table. Will fail if this is an INSERT into a new table.
+ validateNamespace(targetNamespace, unknownType);
+ SqlValidatorTable target = insertNs.resolve().getTable();
+ try {
+ return target.unwrap(DatasourceTable.class);
+ }
+ catch (Exception e) {
+ throw new IAE("Cannot %s into %s: it is not a datasource", operationName, destId);
+ }
+ }
+ catch (CalciteContextException e) {
+ // Something failed. Let's make sure it was the table lookup.
+ // The check is kind of a hack, but its the best we can do given that Calcite
+ // didn't expect this non-SQL use case.
+ if (e.getCause() instanceof SqlValidatorException && e.getMessage().contains(StringUtils.format("Object '%s' not found", tableName))) {
+ // The catalog implementation may be "strict": and require that the target
+ // table already exists, rather than the default "lenient" mode that can
+ // create a new table.
+ if (validatorContext.catalog().ingestRequiresExistingTable()) {
+ throw new IAE("Cannot %s into %s because it does not exist", operationName, destId);
+ }
+ // New table. Validate the shape of the name.
+ IdUtils.validateId(operationName + " dataSource", tableName);
+ return null;
+ }
+ throw e;
+ }
+ }
+
+ private void validateSegmentGranularity(
+ final String operationName,
+ final DruidSqlIngest ingestNode,
+ final DatasourceFacade tableMetadata
+ )
+ {
+ final Granularity definedGranularity = tableMetadata == null ? null : tableMetadata.segmentGranularity();
+ if (definedGranularity != null) {
+ // Should already have been checked when creating the catalog entry
+ DruidSqlParserUtils.throwIfUnsupportedGranularityInPartitionedBy(definedGranularity);
+ }
+ final Granularity ingestionGranularity = DruidSqlParserUtils.convertSqlNodeToGranularity(ingestNode.getPartitionedBy());
+ if (ingestionGranularity != null) {
+ DruidSqlParserUtils.throwIfUnsupportedGranularityInPartitionedBy(ingestionGranularity);
+ }
+ final Granularity finalGranularity;
+ if (definedGranularity == null && ingestionGranularity == null) {
+ // Neither have a value: error
+ throw new IAE(
+ "%s statements must specify a PARTITIONED BY clause explicitly",
+ operationName
+ );
+ } else if (ingestionGranularity == null) {
+ // The query has no granularity: just apply the catalog granularity.
+ finalGranularity = definedGranularity;
+ } else if (definedGranularity == null) {
+ // The catalog has no granularity: apply the query value
+ finalGranularity = ingestionGranularity;
+ } else if (definedGranularity.equals(ingestionGranularity)) {
+ // Both have a setting. They have to be the same.
+ finalGranularity = definedGranularity;
+ } else {
+ throw new IAE(
+ "PARTITIONED BY mismatch. Catalog: [%s], query: [%s]",
+ granularityToSqlString(definedGranularity),
+ granularityToSqlString(ingestionGranularity)
+ );
+ }
+
+ // Note: though this is the validator, we cheat a bit and write the target
+ // granularity into the query context. Perhaps this step should be done
+ // during conversion, however, we've just worked out the granularity, so we
+ // do it here instead.
+ try {
+ validatorContext.queryContextMap().put(
+ DruidSqlInsert.SQL_INSERT_SEGMENT_GRANULARITY,
+ validatorContext.jsonMapper().writeValueAsString(finalGranularity)
+ );
+ }
+ catch (JsonProcessingException e) {
+ throw new IAE("Invalid PARTITIONED BY granularity");
+ }
+ }
+
+ private String granularityToSqlString(Granularity gran)
+ {
+ if (gran == null) {
+ return "NULL";
+ }
+ if (Granularities.ALL == gran) {
+ return "ALL TIME";
+ }
+ if (!(gran instanceof PeriodGranularity)) {
+ return gran.toString();
+ }
+ return ((PeriodGranularity) gran).getPeriod().toString();
+ }
+
+ private SqlSelect validateInsertSelect(
+ final SqlInsert insert,
+ final String operationName
+ )
+ {
+ SqlNode source = insert.getSource();
+
+ // The source SELECT cannot include an ORDER BY clause. Ordering is given
+ // by the CLUSTERED BY clause, if any.
+ // Check that an ORDER BY clause is not provided by the underlying query
+ SqlNodeList orderByList;
+ if (source instanceof SqlOrderBy) {
+ throw new IAE(
+ "Cannot have ORDER BY on %s %s statement, use CLUSTERED BY instead.",
+ "INSERT".equals(operationName) ? "an" : "a",
+ operationName
+ );
+ }
+
+ // Validate the query: but we only know how to do SELECT.
+ // As we do, get the scope for the query.
+ if (!source.isA(SqlKind.QUERY)) {
+ throw new IAE("Cannot execute %s.", source.getKind());
+ }
+ if (!(source instanceof SqlSelect)) {
+ throw new IAE(
+ "%s %s must include a SELECT statement",
+ "INSERT".equals(operationName) ? "An" : "A",
+ operationName
+ );
+ }
+ SqlSelect select = (SqlSelect) source;
+ orderByList = select.getOrderList();
+ if (orderByList != null && orderByList.size() != 0) {
+ throw new IAE(
+ "Cannot have ORDER BY on %s %s statement, use CLUSTERED BY instead.",
+ "INSERT".equals(operationName) ? "an" : "a",
+ operationName
+ );
+ }
+ return select;
+ }
+
+ private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+ {
+ if (tableMetadata == null) {
+ return null;
+ }
+ List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+ if (CollectionUtils.isNullOrEmpty(keyCols)) {
+ return null;
+ }
+ SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+ for (ClusterKeySpec keyCol : keyCols) {
+ SqlIdentifier colIdent = new SqlIdentifier(
+ Collections.singletonList(keyCol.expr()),
+ null, SqlParserPos.ZERO,
+ Collections.singletonList(SqlParserPos.ZERO)
+ );
+ SqlNode keyNode;
+ if (keyCol.desc()) {
+ keyNode = SqlStdOperatorTable.DESC.createCall(SqlParserPos.ZERO, colIdent);
+ } else {
+ keyNode = colIdent;
+ }
+ keyNodes.add(keyNode);
+ }
+ return keyNodes;
+ }
+
+ /**
+ * 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(SqlSelect select, DruidSqlIngest ingestNode, SqlNodeList catalogClustering)
+ {
+ SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+ if (clusteredBy == null || clusteredBy.getList().isEmpty()) {
+ if (catalogClustering == null || catalogClustering.getList().isEmpty()) {
+ return;
+ }
+ clusteredBy = catalogClustering;
+ }
+
+ // 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.
+ select.setOrderBy(clusteredBy);
+ OrderByScope orderScope = ValidatorShim.newOrderByScope(scopes.get(select), clusteredBy, select);
+ try {
+ @SuppressWarnings("unchecked")
+ Map<SqlSelect, SqlValidatorScope> orderScopes = (Map<SqlSelect, SqlValidatorScope>) FieldUtils.getDeclaredField(SqlValidatorImpl.class, "orderScopes", true).get(this);
+ orderScopes.put(select, orderScope);
+ }
+ catch (Exception e) {
+ throw new ISE(e, "orderScopes is not accessible");
+ }
+ }
+
+ private void validateTimeColumn(RelRecordType sourceType, int timeColumnIndex)
+ {
+ RelDataTypeField timeCol = sourceType.getFieldList().get(timeColumnIndex);
+ RelDataType timeColType = timeCol.getType();
+ if (timeColType instanceof BasicSqlType) {
+ BasicSqlType timeColSqlType = (BasicSqlType) timeColType;
+ SqlTypeName timeColSqlTypeName = timeColSqlType.getSqlTypeName();
+ if (timeColSqlTypeName == SqlTypeName.BIGINT || timeColSqlTypeName == SqlTypeName.TIMESTAMP) {
+ return;
+ }
+ }
+ throw new IAE(
+ "Invalid %s column type %s: must be BIGINT or TIMESTAMP",
+ timeCol.getName(),
+ timeColType.toString()
+ );
+ }
+
+ /**
+ * Verify clustering which can come from the query, the catalog or both. If both,
+ * the two must match. In either case, the cluster keys must be present in the SELECT
+ * clause. The {@code __time} column cannot be included.
+ * @param source
+ */
+ private void validateClustering(
+ final SqlSelect source,
+ final SqlValidatorScope selectScope,
+ final RelRecordType sourceType,
+ final int timeColumnIndex,
+ final DruidSqlIngest ingestNode,
+ final SqlNodeList catalogClustering
+ )
+ {
+ final SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+
+ // Validate both the catalog and query definitions if present. This ensures
+ // that things are sane if we later check that the two are identical.
+ if (clusteredBy != null) {
+ validateClusteredBy(source, sourceType, timeColumnIndex, selectScope, clusteredBy);
+ }
+ if (catalogClustering != null) {
+ // Catalog defines the key columns. Verify that they are present in the query.
+ validateClusteredBy(source, sourceType, timeColumnIndex, selectScope, catalogClustering);
+ }
+ if (clusteredBy != null && catalogClustering != null) {
+ // Both the query and catalog have keys.
+ verifyQueryClusterByMatchesCatalog(sourceType, catalogClustering, clusteredBy);
+ }
+ }
+
+ /**
+ * Validate the CLUSTERED BY list. Members can be any of the following:
+ * <p>
+ * {@code CLUSTERED BY [<ordinal> | <id> | <expr>] DESC?}
+ * <p>
+ * Ensure that each id exists. Ensure each column is included only once.
+ * For an expression, just ensure it is valid; we don't check for duplicates.
+ * <p>
+ * Once the keys are validated, update the underlying {@code SELECT} statement
+ * to include the {@code CLUSTERED BY} as the {@code ORDER BY} clause.
+ */
+ private void validateClusteredBy(
+ final SqlSelect source,
+ final RelRecordType sourceType,
+ final int timeColumnIndex,
+ final SqlValidatorScope selectScope,
Review Comment:
## Useless parameter
The parameter 'selectScope' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/3691)
--
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