FrankChen021 commented on code in PR #19830: URL: https://github.com/apache/druid/pull/19830#discussion_r3690442095
########## sql/src/main/java/org/apache/druid/sql/calcite/planner/ProjectionSpecTranslator.java: ########## @@ -0,0 +1,546 @@ +/* + * 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.sql.calcite.planner; + +import com.google.common.collect.ImmutableMap; +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlSelect; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.util.SqlBasicVisitor; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.data.input.impl.AggregateProjectionSpec; +import org.apache.druid.data.input.impl.DimensionSchema; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidSqlInput; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.math.expr.ExprMacroTable; +import org.apache.druid.query.DataSource; +import org.apache.druid.query.Query; +import org.apache.druid.query.QueryContexts; +import org.apache.druid.query.TableDataSource; +import org.apache.druid.query.aggregation.AggregatorFactory; +import org.apache.druid.query.dimension.DefaultDimensionSpec; +import org.apache.druid.query.dimension.DimensionSpec; +import org.apache.druid.query.filter.AndDimFilter; +import org.apache.druid.query.filter.DimFilter; +import org.apache.druid.query.filter.RangeFilter; +import org.apache.druid.query.groupby.GroupByQuery; +import org.apache.druid.query.scan.ScanQuery; +import org.apache.druid.query.timeseries.TimeseriesQuery; +import org.apache.druid.segment.VirtualColumn; +import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.server.security.AuthorizationResult; +import org.apache.druid.server.security.NoopEscalator; +import org.apache.druid.sql.calcite.rel.DruidQuery; +import org.apache.druid.sql.calcite.rel.Grouping; +import org.apache.druid.sql.calcite.table.DatasourceTable; +import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata; +import org.apache.druid.sql.calcite.table.DruidTable; +import org.joda.time.Interval; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * Turns the SQL body of a projection definition into the {@link AggregateProjectionSpec} the catalog stores. + * <p> + * The body is planned through the normal pipeline against the columns the enclosing statement declares, and the + * specification is lifted out of the resulting native query. Going through the planner is the point: a projection is + * only useful if it matches the queries the planner generates at query time, and that agreement is guaranteed when + * the same machinery produces both. It also means aggregators contributed by extensions work without a second + * registry. + */ +public class ProjectionSpecTranslator +{ + /** + * The reserved projection name that describes the table's own physical layout. + */ + public static final String BASE_PROJECTION_NAME = "__base"; + + /** + * Planning is deterministic in the shapes the lift understands: no timeseries or topN rewrite to hide the grouping + * columns, and no approximation choices that depend on unrelated configuration. + */ + private static final Map<String, Object> CONTEXT = ImmutableMap.of( + PlannerContext.CTX_SQL_USE_GRANULARITY, false, + QueryContexts.TIME_BOUNDARY_PLANNING_KEY, false, + PlannerConfig.CTX_KEY_USE_APPROXIMATE_TOPN, false + ); + + private final PlannerFactory plannerFactory; + + public ProjectionSpecTranslator(PlannerFactory plannerFactory) + { + this.plannerFactory = plannerFactory; + } + + /** + * Translate one projection definition. + * + * @param tableName the table the projection belongs to + * @param columns the table's declared columns, which are the only ones the body may reference + */ + public AggregateProjectionSpec translate( + final String tableName, + final List<ColumnSpec> columns, + final String projectionName, + final SqlSelect body + ) + { + rejectSubqueries(projectionName, body); + + final DruidQuery druidQuery = planBody(tableName, columns, projectionName, body); + return lift(projectionName, tableName, druidQuery); + } + + /** + * Translate the reserved base-table projection, which describes the physical layout of the table itself rather + * than an additional aggregate. + * <p> + * The body enumerates the table's columns in the order segments store them, so it must name every declared column, + * in declared order. An item written as {@code <expr> AS <name>} makes that column computed at ingest time: the + * expression becomes a virtual column materializing the declared column, which is why the declared type has to + * match what the expression produces. + * + * @param clusteredBy the columns segments are clustered on, which must be the leading prefix of the column list + */ + public ClusteredValueGroupsBaseTableMetadata translateBaseTable( + final String tableName, + final List<ColumnSpec> columns, + final SqlSelect body, + @Nullable final SqlNodeList clusteredBy + ) + { + if (body.getWhere() != null || body.getGroup() != null) { + throw invalid( + BASE_PROJECTION_NAME, + "its body filters or groups. The base table stores every ingested row, so it can do neither" + ); + } + rejectSubqueries(BASE_PROJECTION_NAME, body); + + final DruidQuery druidQuery = planBody(tableName, columns, BASE_PROJECTION_NAME, body); + final ClusteredValueGroupsBaseTableMetadata metadata = new ClusteredValueGroupsBaseTableMetadata( + clusteringColumns(clusteredBy), + liftComputedColumns(columns, druidQuery), + null + ); + + // Derive the physical spec now. The catalog does this too when the write lands, but doing it here attributes + // layout problems to the statement that caused them rather than to a Coordinator round trip. + try { + metadata.createSpec(columns); + } + catch (DruidException e) { + throw contextualize(BASE_PROJECTION_NAME, e); + } + return metadata; + } + + private static List<String> clusteringColumns(@Nullable final SqlNodeList clusteredBy) + { + if (clusteredBy == null) { + return Collections.emptyList(); + } + final List<String> names = new ArrayList<>(clusteredBy.size()); + for (SqlNode node : clusteredBy) { + if (!(node instanceof SqlIdentifier) || !((SqlIdentifier) node).isSimple()) { + throw invalid( + BASE_PROJECTION_NAME, + "its CLUSTERED BY names [" + node + "], which is not a column. Segments are clustered on stored columns;" + + " to cluster on a computed value, declare it as a column of the table" + ); + } + names.add(((SqlIdentifier) node).getSimple()); + } + return names; + } + + /** + * Pair the planned output with the declared columns and lift the virtual columns behind the computed ones. + * <p> + * The planner names its virtual columns {@code v0}, {@code v1}, ...; each is renamed to the declared column it + * fills, which is what makes it a materialized column rather than an anonymous intermediate. + */ + private static VirtualColumns liftComputedColumns( + final List<ColumnSpec> columns, + final DruidQuery druidQuery + ) + { + final Query<?> query = druidQuery.getQuery(); + if (!(query instanceof ScanQuery)) { + throw invalid( + BASE_PROJECTION_NAME, + "its body does not select rows directly. The base table stores every ingested row as it arrives" + ); + } + final List<String> selected = ((ScanQuery) query).getColumns(); + final List<String> outputNames = druidQuery.getOutputRowType().getFieldNames(); + + if (outputNames.size() != columns.size()) { + throw invalid( + BASE_PROJECTION_NAME, + StringUtils.format( + "it selects %d column(s) but the table declares %d. The body lists the columns in the order segments" + + " store them, so it must name every declared column", + outputNames.size(), + columns.size() + ) + ); + } + + final VirtualColumns planned = ((ScanQuery) query).getVirtualColumns(); + final List<VirtualColumn> materialized = new ArrayList<>(); + for (int i = 0; i < columns.size(); i++) { + final String declared = columns.get(i).name(); + if (!declared.equals(outputNames.get(i))) { + throw invalid( + BASE_PROJECTION_NAME, + StringUtils.format( + "its column %d is [%s] but the table declares [%s] there. The body lists the columns in the order" + + " segments store them", + i + 1, + outputNames.get(i), + declared + ) + ); + } + final VirtualColumn virtualColumn = planned.getVirtualColumn(selected.get(i)); + if (virtualColumn == null) { + // A plain reference: the column is ingested as it arrives. + continue; + } + if (!(virtualColumn instanceof ExpressionVirtualColumn)) { + throw invalid( + BASE_PROJECTION_NAME, + "column [" + declared + "] is computed by an expression the base table cannot store" + ); + } + final ExpressionVirtualColumn expression = (ExpressionVirtualColumn) virtualColumn; + materialized.add( + new ExpressionVirtualColumn( + declared, + expression.getExpression(), + expression.getOutputType(), + ExprMacroTable.nil() + ) + ); + } + return VirtualColumns.create(materialized); + } + + /** + * Plan the body against a table built from the declared columns. The table is synthesized rather than looked up + * because for {@code CREATE TABLE} it does not exist yet, and for {@code ALTER TABLE} the statement's own columns + * are what the projection must agree with, not whatever a possibly stale cache holds. + */ + private DruidQuery planBody( + final String tableName, + final List<ColumnSpec> columns, + final String projectionName, + final SqlSelect body + ) + { + final SqlSelect query = (SqlSelect) body.clone(body.getParserPosition()); + query.setFrom(new SqlIdentifier(tableName, SqlParserPos.ZERO)); + + final ProjectionSqlEngine engine = new ProjectionSqlEngine(); + final String sql = query.toString(); + try (DruidPlanner planner = plannerFactory.createPlannerForTable( + engine, + sql, + query, + CONTEXT, Review Comment: [P2] Preserve statement context when planning projections The nested planner receives only the hard-coded CONTEXT and discards the enclosing statement context, even though SET clauses are explicitly supported before DDL. For example, SET sqlTimeZone = 'America/Los_Angeles' followed by a projection using TIME_FLOOR stores a UTC expression, so the equivalent query under the same context plans differently and cannot match the projection. Merge relevant outer PlannerContext values before applying the deterministic overrides. ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java: ########## @@ -0,0 +1,736 @@ +/* + * 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.sql.calcite.planner; + +import com.google.common.base.Supplier; +import com.google.common.base.Suppliers; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import org.apache.calcite.jdbc.CalciteSchema; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.catalog.model.DatasourceProjectionMetadata; +import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.TableMetadata; +import org.apache.druid.catalog.model.TableSpec; +import org.apache.druid.catalog.model.table.ClusterKeySpec; +import org.apache.druid.catalog.model.table.DatasourceDefn; +import org.apache.druid.common.utils.IdUtils; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidSqlInput; +import org.apache.druid.java.util.common.IAE; +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.java.util.common.guava.Sequences; +import org.apache.druid.query.explain.ExplainAttributes; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.projections.Projections; +import org.apache.druid.server.QueryResponse; +import org.apache.druid.server.security.Action; +import org.apache.druid.server.security.Resource; +import org.apache.druid.server.security.ResourceAction; +import org.apache.druid.server.security.ResourceType; +import org.apache.druid.sql.calcite.parser.DruidSqlAlterTable; +import org.apache.druid.sql.calcite.parser.DruidSqlColumnDeclaration; +import org.apache.druid.sql.calcite.parser.DruidSqlCreateTable; +import org.apache.druid.sql.calcite.parser.DruidSqlParser; +import org.apache.druid.sql.calcite.parser.DruidSqlPropertyAssignment; +import org.apache.druid.sql.calcite.parser.SqlGranularityLiteral; +import org.apache.druid.sql.calcite.parser.SqlProjectionSpec; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Handles the catalog DDL statements: {@code CREATE TABLE} and {@code ALTER TABLE}. + * <p> + * These statements are metadata operations, not queries. They are validated here, converted to a catalog + * {@link TableSpec} or column/property edit, and applied through {@link CatalogTableWriter}, which forwards them to + * the Coordinator. No Calcite validation or query planning takes place, and no rows are returned. + * <p> + * Validation is deliberately split. This class checks what it can attribute to a position in the statement (type + * spellings, duplicate columns, granularity, clustering) so that the error names the offending SQL. The Coordinator + * remains authoritative: {@code DatasourceDefn.validate} runs on write, and its message is surfaced verbatim. + */ +public abstract class CatalogDdlHandler extends SqlStatementHandler.BaseStatementHandler +{ + /** + * DDL produces no rows. A single-column type still has to be declared, because JDBC clients ask for a result set + * signature when preparing the statement. + */ + private static final RelDataType RESULT_TYPE = resultType(); + + protected final SqlIdentifier tableIdentifier; + protected TableId tableId; + + protected CatalogDdlHandler(SqlStatementHandler.HandlerContext handlerContext, SqlIdentifier tableIdentifier) + { + super(handlerContext); + this.tableIdentifier = tableIdentifier; + } + + /** + * The runtime property that gates these statements. Read from {@link PlannerConfig} rather than the query context + * so that a user cannot turn the feature on for their own statement. + */ + public static final String ENABLE_CATALOG_DDL_PROPERTY = "druid.sql.planner.enableCatalogDdl"; + + /** + * The reserved name of the base-table projection, which describes the physical layout of the table itself. Handled + * as a separate catalog property, not as one of the aggregate projections. + */ + public static final String BASE_PROJECTION_NAME = "__base"; + + @Override + public void validate() + { + if (!handlerContext.plannerContext().getPlannerConfig().isEnableCatalogDdl()) { + throw DruidException.forPersona(DruidException.Persona.ADMIN) + .ofCategory(DruidException.Category.UNSUPPORTED) + .build( + "Catalog DDL statements are disabled. Set [%s] to true on the Broker to enable [%s].", + ENABLE_CATALOG_DDL_PROPERTY, + operationName() + ); + } + if (!handlerContext.plannerContext().getParameters().isEmpty()) { + throw InvalidSqlInput.exception("Dynamic parameters are not supported for [%s]", operationName()); + } + tableId = TableId.datasource(resolveTableName()); + resourceActions = Collections.singleton( + new ResourceAction(new Resource(tableId.name(), ResourceType.DATASOURCE), Action.WRITE) + ); + validateStatement(); + } + + /** + * Statement-specific validation, which also prepares whatever {@link #execute} will apply. + */ + protected abstract void validateStatement(); + + protected abstract void execute(CatalogTableWriter writer); + + protected abstract String operationName(); + + @Override + public void prepare() + { + // Nothing to prepare: there is no query to plan. + } + + @Override + public PrepareResult prepareResult() + { + return new PrepareResult(RESULT_TYPE, RESULT_TYPE, DruidTypeSystem.TYPE_FACTORY.createStructType( + Collections.emptyList(), + Collections.emptyList() + )); + } + + @Override + public PlannerResult plan() + { + execute(handlerContext.plannerContext().getPlannerToolbox().catalogTableWriter()); + final Supplier<QueryResponse<Object[]>> resultsSupplier = Suppliers.ofInstance( + QueryResponse.withEmptyContext(Sequences.empty()) + ); + return new PlannerResult(resultsSupplier, RESULT_TYPE); + } + + @Override + public ExplainAttributes explainAttributes() + { + throw InvalidSqlInput.exception("EXPLAIN is not supported for [%s]", operationName()); + } + + /** + * Resolve the table name, which may be unqualified or qualified by the Druid schema. Other schemas are rejected: + * only datasources have catalog specs that DDL can write. + */ + private String resolveTableName() + { + final String tableName; + if (tableIdentifier.names.size() == 1) { + tableName = tableIdentifier.names.get(0); + } else if (tableIdentifier.names.size() == 2) { + final String defaultSchemaName = + Iterables.getOnlyElement(CalciteSchema.from(handlerContext.defaultSchema()).path(null)); + if (!defaultSchemaName.equals(tableIdentifier.names.get(0))) { + throw InvalidSqlInput.exception( + "Table [%s] does not support operation [%s] because it is not a Druid datasource", + tableIdentifier, + operationName() + ); + } + tableName = tableIdentifier.names.get(1); + } else { + throw InvalidSqlInput.exception( + "Table name [%s] is not valid for operation [%s]", + tableIdentifier, + operationName() + ); + } + IdUtils.validateId("table", tableName); + return tableName; + } + + /** + * Convert a parsed column declaration into its catalog form, checking that the type is one Druid can store. + */ + protected static ColumnSpec toColumnSpec(DruidSqlColumnDeclaration declaration) + { + final String name = simpleName(declaration.getName(), "Column"); + final String type; + try { + type = CatalogColumnTypes.forCatalogColumn(name, declaration.getDataType()); + } + catch (IAE e) { + throw InvalidSqlInput.exception(e, "%s", e.getMessage()); + } + if (Columns.isTimeColumn(name) && !ColumnType.LONG.equals(Columns.druidTypeFromString(type))) { + throw InvalidSqlInput.exception( + "Column [%s] must have type [%s] or [%s], but was [%s]", + Columns.TIME_COLUMN, + Columns.SQL_TIMESTAMP, + Columns.SQL_BIGINT, + type + ); + } + return new ColumnSpec(name, type, null); + } + + /** + * Translate one projection definition into the catalog form. + * <p> + * {@code __base} is reserved for the base-table projection, which is a different catalog entity: it describes the + * physical layout of the table itself rather than an additional aggregate. It is rejected here rather than being + * translated as an ordinary projection. + */ + protected static DatasourceProjectionMetadata translateProjection( + final SqlStatementHandler.HandlerContext handlerContext, + final String tableName, + final List<ColumnSpec> columns, + final SqlProjectionSpec projection + ) + { + final String name = simpleName(projection.getName(), "Projection"); + try { + Projections.validateProjectionName(name); + } + catch (DruidException e) { + throw InvalidSqlInput.exception(e, "%s", e.getMessage()); + } + if (projection.getClusteredBy() != null) { + throw InvalidSqlInput.exception( + "Projection [%s] cannot use CLUSTERED BY: an aggregate projection is ordered by its grouping columns." + + " Only the [%s] projection, which describes the table's own layout, chooses a clustering", + name, + BASE_PROJECTION_NAME + ); + } + return new DatasourceProjectionMetadata( + new ProjectionSpecTranslator(handlerContext.plannerFactory()) + .translate(tableName, columns, name, projection.getBody()) + ); + } + + /** + * Translate the reserved {@code __base} projection, which describes the physical layout of the table rather than an + * additional aggregate, and so becomes the {@code baseTable} property instead of one of the projections. + */ + protected static ClusteredValueGroupsBaseTableMetadata translateBaseTable( + final SqlStatementHandler.HandlerContext handlerContext, + final String tableName, + final List<ColumnSpec> columns, + final SqlProjectionSpec projection + ) + { + return new ProjectionSpecTranslator(handlerContext.plannerFactory()) + .translateBaseTable(tableName, columns, projection.getBody(), projection.getClusteredBy()); + } + + /** + * A base table layout derives the physical segment schema from the declared columns, so a column that is not + * declared cannot be stored. The catalog enforces this too, but saying it here names the clause that is missing. + */ + protected static void requireSealed(boolean sealed) + { + if (!sealed) { + throw InvalidSqlInput.exception( + "A table with a [%s] projection must be declared SEALED: its columns define the physical segment schema," + + " so columns that are not declared cannot be ingested", + BASE_PROJECTION_NAME + ); + } + } + + protected static String simpleName(SqlIdentifier identifier, String what) + { + if (!identifier.isSimple()) { + throw InvalidSqlInput.exception("%s name [%s] must be a simple name", what, identifier); + } + return identifier.getSimple(); + } + + /** + * The catalog stores a segment granularity as either {@code ALL} or an ISO period string. + */ + protected static String toGranularityString(SqlGranularityLiteral partitionedBy) + { + final Granularity granularity = partitionedBy.getGranularity(); + if (Granularities.ALL.equals(granularity)) { + return DatasourceDefn.ALL_GRANULARITY; + } + if (granularity instanceof PeriodGranularity) { + return ((PeriodGranularity) granularity).getPeriod().toString(); + } + throw InvalidSqlInput.exception("Granularity [%s] is not supported by the catalog", partitionedBy); + } + + /** + * The catalog's clustering keys are plain ascending column references. Expressions, ordinals and DESC have no + * catalog representation, so they are rejected here rather than silently dropped. + */ + protected static List<ClusterKeySpec> toClusterKeys(SqlNodeList clusteredBy) + { + final List<ClusterKeySpec> keys = new ArrayList<>(clusteredBy.size()); + for (SqlNode node : clusteredBy) { + if (!(node instanceof SqlIdentifier) || !((SqlIdentifier) node).isSimple()) { + throw InvalidSqlInput.exception( + "CLUSTERED BY column [%s] must be a column name; expressions, ordinals and DESC are not supported when" + + " defining a table", + node + ); + } + keys.add(new ClusterKeySpec(((SqlIdentifier) node).getSimple(), false)); + } + return keys; + } + + private static RelDataType resultType() + { + final RelDataTypeFactory typeFactory = DruidTypeSystem.TYPE_FACTORY; + return typeFactory.createStructType( + ImmutableList.of(Calcites.createSqlType(typeFactory, SqlTypeName.VARCHAR)), + ImmutableList.of("RESULT") + ); + } + + /** + * {@code CREATE [OR REPLACE] TABLE [IF NOT EXISTS] ...}. + */ + public static class CreateTableHandler extends CatalogDdlHandler + { + private final DruidSqlCreateTable createTable; + private TableSpec tableSpec; + + public CreateTableHandler(SqlStatementHandler.HandlerContext handlerContext, DruidSqlCreateTable createTable) + { + super(handlerContext, createTable.getName()); + this.createTable = createTable; + } + + @Override + protected void validateStatement() + { + if (createTable.getReplace() && createTable.isIfNotExists()) { + throw InvalidSqlInput.exception("Cannot specify both OR REPLACE and IF NOT EXISTS"); + } + + final List<ColumnSpec> columns = new ArrayList<>(createTable.getColumnList().size()); + final Set<String> seen = new HashSet<>(); + for (SqlNode node : createTable.getColumnList()) { + final ColumnSpec column = toColumnSpec((DruidSqlColumnDeclaration) node); + if (!seen.add(column.name())) { + throw InvalidSqlInput.exception("Column [%s] is declared more than once", column.name()); + } + columns.add(column); + } + + final Map<String, Object> properties = new LinkedHashMap<>(); + if (createTable.getPartitionedBy() != null) { + properties.put( + DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY, + toGranularityString(createTable.getPartitionedBy()) + ); + } + if (createTable.getClusteredBy() != null) { + properties.put(DatasourceDefn.CLUSTER_KEYS_PROPERTY, toClusterKeys(createTable.getClusteredBy())); + } + if (createTable.isSealed()) { + properties.put(DatasourceDefn.SEALED_PROPERTY, true); + } + if (!createTable.getProjectionList().isEmpty()) { + final List<DatasourceProjectionMetadata> projections = + new ArrayList<>(createTable.getProjectionList().size()); + final Set<String> seenProjections = new HashSet<>(); + for (SqlNode node : createTable.getProjectionList()) { + final SqlProjectionSpec projection = (SqlProjectionSpec) node; + final String name = simpleName(projection.getName(), "Projection"); + if (!seenProjections.add(name)) { + throw InvalidSqlInput.exception("Projection [%s] is declared more than once", name); + } + if (BASE_PROJECTION_NAME.equals(name)) { + requireSealed(createTable.isSealed()); + properties.put( + DatasourceDefn.BASE_TABLE_PROPERTY, + translateBaseTable(handlerContext, tableId.name(), columns, projection) + ); + } else { + projections.add(translateProjection(handlerContext, tableId.name(), columns, projection)); + } + } + if (!projections.isEmpty()) { + properties.put(DatasourceDefn.PROJECTIONS_KEYS_PROPERTY, projections); + } + } + + tableSpec = new TableSpec(DatasourceDefn.TABLE_TYPE, properties, columns); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.createTable(tableId, tableSpec, createTable.isIfNotExists(), createTable.getReplace()); + } + + @Override + protected String operationName() + { + return "CREATE TABLE"; + } + } + + /** + * {@code ALTER TABLE ... ADD COLUMN}. The Coordinator merges columns by name, so an existing column would be + * silently updated; this checks first so that {@code ADD} means add. + */ + public static class AddColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.AddColumn alterTable; + private ColumnSpec column; + + public AddColumnHandler(SqlStatementHandler.HandlerContext handlerContext, DruidSqlAlterTable.AddColumn alterTable) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = toColumnSpec(alterTable.getColumn()); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + final TableMetadata existing = writer.readTable(tableId); + if (existing == null) { + throw InvalidSqlInput.exception("Table [%s] does not have a catalog entry", tableId.name()); + } + if (existing.spec().columns() != null + && existing.spec().columns().stream().anyMatch(c -> column.name().equals(c.name()))) { + throw InvalidSqlInput.exception( + "Column [%s] already exists in table [%s]; use ALTER COLUMN to change its type", + column.name(), + tableId.name() + ); + } + writer.updateColumns(tableId, Collections.singletonList(column)); + } + + @Override + protected String operationName() + { + return "ALTER TABLE ADD COLUMN"; + } + } + + /** + * {@code ALTER TABLE ... DROP COLUMN}. + */ + public static class DropColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.DropColumn alterTable; + private String column; + + public DropColumnHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.DropColumn alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = simpleName(alterTable.getColumn(), "Column"); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.dropColumns(tableId, Collections.singletonList(column)); + } + + @Override + protected String operationName() + { + return "ALTER TABLE DROP COLUMN"; + } + } + + /** + * {@code ALTER TABLE ... ALTER COLUMN ... SET DATA TYPE}. Merging a column by name is exactly what changing its + * type requires, so this reuses the same update as ADD COLUMN without the existence check. + */ + public static class AlterColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.AlterColumn alterTable; + private ColumnSpec column; + + public AlterColumnHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.AlterColumn alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = toColumnSpec(alterTable.getColumn()); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.updateColumns(tableId, Collections.singletonList(column)); + } + + @Override + protected String operationName() + { + return "ALTER TABLE ALTER COLUMN"; + } + } + + /** + * {@code ALTER TABLE ... ADD PROJECTION}. The body is translated against the table's current declared columns, so + * the table must already have a catalog entry. + */ + public static class AddProjectionHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.AddProjection alterTable; + private String projectionName; + + public AddProjectionHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.AddProjection alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + projectionName = simpleName(alterTable.getProjection().getName(), "Projection"); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + final TableMetadata existing = writer.readTable(tableId); + if (existing == null) { + throw InvalidSqlInput.exception("Table [%s] does not have a catalog entry", tableId.name()); + } + final List<ColumnSpec> columns = + existing.spec().columns() == null ? Collections.emptyList() : existing.spec().columns(); + + if (BASE_PROJECTION_NAME.equals(projectionName)) { + // The base table is a property of the table, not one of its projections, so it is set rather than appended. + if (existing.spec().properties().get(DatasourceDefn.BASE_TABLE_PROPERTY) != null) { + if (alterTable.isIfNotExists()) { + return; + } + throw InvalidSqlInput.exception( + "Table [%s] already has a [%s] projection; drop it before defining another", + tableId.name(), + BASE_PROJECTION_NAME + ); + } + requireSealed(Boolean.TRUE.equals(existing.spec().properties().get(DatasourceDefn.SEALED_PROPERTY))); + writer.updateProperties( + tableId, + Collections.singletonMap( + DatasourceDefn.BASE_TABLE_PROPERTY, + translateBaseTable(handlerContext, tableId.name(), columns, alterTable.getProjection()) + ) + ); + return; + } + + writer.addProjection( + tableId, + translateProjection(handlerContext, tableId.name(), columns, alterTable.getProjection()), + alterTable.isIfNotExists() + ); + } + + @Override + protected String operationName() + { + return "ALTER TABLE ADD PROJECTION"; + } + } + + /** + * {@code ALTER TABLE ... DROP PROJECTION}. Segments already built keep whatever projections they were built with; + * this only stops future ingestion from building it. + */ + public static class DropProjectionHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.DropProjection alterTable; + private String projectionName; + + public DropProjectionHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.DropProjection alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + projectionName = simpleName(alterTable.getProjectionName(), "Projection"); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + if (BASE_PROJECTION_NAME.equals(projectionName)) { + // Removing the layout leaves the declared columns alone; only future segments are affected. + final TableMetadata existing = writer.readTable(tableId); + final boolean present = existing != null + && existing.spec().properties().get(DatasourceDefn.BASE_TABLE_PROPERTY) != null; + if (!present) { + if (alterTable.isIfExists()) { + return; + } + throw InvalidSqlInput.exception( + "Table [%s] does not have a [%s] projection", + tableId.name(), + BASE_PROJECTION_NAME + ); + } + writer.updateProperties( + tableId, + Collections.singletonMap(DatasourceDefn.BASE_TABLE_PROPERTY, null) + ); + return; + } + writer.dropProjection(tableId, projectionName, alterTable.isIfExists()); + } + + @Override + protected String operationName() + { + return "ALTER TABLE DROP PROJECTION"; + } + } + + /** + * {@code ALTER TABLE ... SET PROPERTIES}. A NULL value removes the property. The set of legal keys is not checked + * here: the Coordinator's table definition registry is what knows them. + */ + public static class SetPropertiesHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.SetProperties alterTable; + private Map<String, Object> properties; + + public SetPropertiesHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.SetProperties alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + properties = new LinkedHashMap<>(); + for (SqlNode node : alterTable.getProperties()) { + final DruidSqlPropertyAssignment assignment = (DruidSqlPropertyAssignment) node; + final String key = simpleName(assignment.getKey(), "Property"); + if (properties.containsKey(key)) { + throw InvalidSqlInput.exception("Property [%s] is assigned more than once", key); + } + properties.put(key, propertyValue(key, assignment.getValue())); + } + } + + private static Object propertyValue(String key, SqlNode value) + { + if (!(value instanceof SqlLiteral)) { + throw InvalidSqlInput.exception("Value for property [%s] must be a literal", key); + } + // A NULL literal coerces to null, which the catalog treats as "remove this property". + return DruidSqlParser.sqlLiteralToJavaValue((SqlLiteral) value, "property " + key); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.updateProperties(tableId, properties); Review Comment: [P1] Revalidate the complete table after property edits This uses the property-only edit endpoint, whose transaction loads and validates properties without columns, so DatasourceDefn.validate(ResolvedTable) and its cross-field checks never run. For example, after defining a DAY projection, SET PROPERTIES can change segmentGranularity to PT1H even though full validation rejects a projection coarser than its segments; it can likewise clear sealed while __base remains. The catalog then contains an invalid specification and subsequent ingestion fails. Load and validate the complete revised TableSpec inside the Coordinator transaction. ########## sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlCreateTable.java: ########## @@ -0,0 +1,215 @@ +/* + * 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.sql.calcite.parser; + +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlCreate; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlOperator; +import org.apache.calcite.sql.SqlSpecialOperator; +import org.apache.calcite.sql.SqlWriter; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.util.ImmutableNullableList; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.List; + +/** + * {@code CREATE [OR REPLACE] TABLE [IF NOT EXISTS] <name> (<columns>) [PARTITIONED BY <granularity>] + * [CLUSTERED BY <columns>]}, which defines a table in the Druid catalog. + * <p> + * This statement writes catalog metadata only; it neither creates segments nor otherwise touches data. See + * {@link org.apache.druid.sql.calcite.planner.CatalogDdlHandler} for the execution side. + */ +public class DruidSqlCreateTable extends SqlCreate +{ + public static final SqlOperator OPERATOR = new DruidSqlCreateTableOperator(); + + private final SqlIdentifier name; + private final SqlNodeList columnList; + private final SqlNodeList projectionList; + @Nullable + private final SqlGranularityLiteral partitionedBy; + @Nullable + private final SqlNodeList clusteredBy; + private final boolean sealed; + + public DruidSqlCreateTable( + SqlParserPos pos, + boolean replace, + boolean ifNotExists, + SqlIdentifier name, + SqlNodeList columnList, + SqlNodeList projectionList, + @Nullable SqlGranularityLiteral partitionedBy, + @Nullable SqlNodeList clusteredBy, + boolean sealed + ) + { + super(OPERATOR, pos, replace, ifNotExists); + this.sealed = sealed; + this.name = name; + this.columnList = columnList; + this.projectionList = projectionList; + this.partitionedBy = partitionedBy; + this.clusteredBy = clusteredBy; + } + + public SqlIdentifier getName() + { + return name; + } + + /** + * The declared columns, each a {@link DruidSqlColumnDeclaration}. Order is significant: it is the order columns are + * recorded in the catalog table spec. + */ + public SqlNodeList getColumnList() + { + return columnList; + } + + /** + * The declared projections, each a {@link SqlProjectionSpec}. + */ + public SqlNodeList getProjectionList() + { + return projectionList; + } + + @Nullable + public SqlGranularityLiteral getPartitionedBy() + { + return partitionedBy; + } + + @Nullable + public SqlNodeList getClusteredBy() + { + return clusteredBy; + } + + public boolean isIfNotExists() + { + return ifNotExists; + } + + /** + * Whether the statement declared SEALED, which requires every ingested column to be declared. + */ + public boolean isSealed() + { + return sealed; + } + + @Nonnull + @Override + public List<SqlNode> getOperandList() + { + // The replace and ifNotExists flags travel as operands so that createCall() can rebuild an equivalent node. + return ImmutableNullableList.of( + name, + columnList, + projectionList, + partitionedBy, + clusteredBy, + SqlLiteral.createBoolean(getReplace(), SqlParserPos.ZERO), + SqlLiteral.createBoolean(ifNotExists, SqlParserPos.ZERO), + SqlLiteral.createBoolean(sealed, SqlParserPos.ZERO) + ); + } + + @Override + public void unparse(SqlWriter writer, int leftPrec, int rightPrec) + { + writer.keyword("CREATE"); + if (getReplace()) { + writer.keyword("OR REPLACE"); + } + writer.keyword("TABLE"); + if (ifNotExists) { + writer.keyword("IF NOT EXISTS"); + } + name.unparse(writer, leftPrec, rightPrec); + + final SqlWriter.Frame frame = writer.startList("(", ")"); Review Comment: [P2] Do not unparse omitted columns as empty parentheses The grammar permits CREATE TABLE tbl PARTITIONED BY DAY with no parenthesized element list, but unparse always emits CREATE TABLE tbl () PARTITIONED BY DAY. Empty parentheses cannot be parsed because AddDruidTableElement is mandatory once '(' is present, so a valid AST does not round-trip. Omit the frame when both lists are empty or teach the grammar to accept (). ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java: ########## @@ -0,0 +1,736 @@ +/* + * 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.sql.calcite.planner; + +import com.google.common.base.Supplier; +import com.google.common.base.Suppliers; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import org.apache.calcite.jdbc.CalciteSchema; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.type.SqlTypeName; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.catalog.model.DatasourceProjectionMetadata; +import org.apache.druid.catalog.model.TableId; +import org.apache.druid.catalog.model.TableMetadata; +import org.apache.druid.catalog.model.TableSpec; +import org.apache.druid.catalog.model.table.ClusterKeySpec; +import org.apache.druid.catalog.model.table.DatasourceDefn; +import org.apache.druid.common.utils.IdUtils; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidSqlInput; +import org.apache.druid.java.util.common.IAE; +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.java.util.common.guava.Sequences; +import org.apache.druid.query.explain.ExplainAttributes; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.projections.Projections; +import org.apache.druid.server.QueryResponse; +import org.apache.druid.server.security.Action; +import org.apache.druid.server.security.Resource; +import org.apache.druid.server.security.ResourceAction; +import org.apache.druid.server.security.ResourceType; +import org.apache.druid.sql.calcite.parser.DruidSqlAlterTable; +import org.apache.druid.sql.calcite.parser.DruidSqlColumnDeclaration; +import org.apache.druid.sql.calcite.parser.DruidSqlCreateTable; +import org.apache.druid.sql.calcite.parser.DruidSqlParser; +import org.apache.druid.sql.calcite.parser.DruidSqlPropertyAssignment; +import org.apache.druid.sql.calcite.parser.SqlGranularityLiteral; +import org.apache.druid.sql.calcite.parser.SqlProjectionSpec; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Handles the catalog DDL statements: {@code CREATE TABLE} and {@code ALTER TABLE}. + * <p> + * These statements are metadata operations, not queries. They are validated here, converted to a catalog + * {@link TableSpec} or column/property edit, and applied through {@link CatalogTableWriter}, which forwards them to + * the Coordinator. No Calcite validation or query planning takes place, and no rows are returned. + * <p> + * Validation is deliberately split. This class checks what it can attribute to a position in the statement (type + * spellings, duplicate columns, granularity, clustering) so that the error names the offending SQL. The Coordinator + * remains authoritative: {@code DatasourceDefn.validate} runs on write, and its message is surfaced verbatim. + */ +public abstract class CatalogDdlHandler extends SqlStatementHandler.BaseStatementHandler +{ + /** + * DDL produces no rows. A single-column type still has to be declared, because JDBC clients ask for a result set + * signature when preparing the statement. + */ + private static final RelDataType RESULT_TYPE = resultType(); + + protected final SqlIdentifier tableIdentifier; + protected TableId tableId; + + protected CatalogDdlHandler(SqlStatementHandler.HandlerContext handlerContext, SqlIdentifier tableIdentifier) + { + super(handlerContext); + this.tableIdentifier = tableIdentifier; + } + + /** + * The runtime property that gates these statements. Read from {@link PlannerConfig} rather than the query context + * so that a user cannot turn the feature on for their own statement. + */ + public static final String ENABLE_CATALOG_DDL_PROPERTY = "druid.sql.planner.enableCatalogDdl"; + + /** + * The reserved name of the base-table projection, which describes the physical layout of the table itself. Handled + * as a separate catalog property, not as one of the aggregate projections. + */ + public static final String BASE_PROJECTION_NAME = "__base"; + + @Override + public void validate() + { + if (!handlerContext.plannerContext().getPlannerConfig().isEnableCatalogDdl()) { + throw DruidException.forPersona(DruidException.Persona.ADMIN) + .ofCategory(DruidException.Category.UNSUPPORTED) + .build( + "Catalog DDL statements are disabled. Set [%s] to true on the Broker to enable [%s].", + ENABLE_CATALOG_DDL_PROPERTY, + operationName() + ); + } + if (!handlerContext.plannerContext().getParameters().isEmpty()) { + throw InvalidSqlInput.exception("Dynamic parameters are not supported for [%s]", operationName()); + } + tableId = TableId.datasource(resolveTableName()); + resourceActions = Collections.singleton( + new ResourceAction(new Resource(tableId.name(), ResourceType.DATASOURCE), Action.WRITE) + ); + validateStatement(); + } + + /** + * Statement-specific validation, which also prepares whatever {@link #execute} will apply. + */ + protected abstract void validateStatement(); + + protected abstract void execute(CatalogTableWriter writer); + + protected abstract String operationName(); + + @Override + public void prepare() + { + // Nothing to prepare: there is no query to plan. + } + + @Override + public PrepareResult prepareResult() + { + return new PrepareResult(RESULT_TYPE, RESULT_TYPE, DruidTypeSystem.TYPE_FACTORY.createStructType( + Collections.emptyList(), + Collections.emptyList() + )); + } + + @Override + public PlannerResult plan() + { + execute(handlerContext.plannerContext().getPlannerToolbox().catalogTableWriter()); + final Supplier<QueryResponse<Object[]>> resultsSupplier = Suppliers.ofInstance( + QueryResponse.withEmptyContext(Sequences.empty()) + ); + return new PlannerResult(resultsSupplier, RESULT_TYPE); + } + + @Override + public ExplainAttributes explainAttributes() + { + throw InvalidSqlInput.exception("EXPLAIN is not supported for [%s]", operationName()); + } + + /** + * Resolve the table name, which may be unqualified or qualified by the Druid schema. Other schemas are rejected: + * only datasources have catalog specs that DDL can write. + */ + private String resolveTableName() + { + final String tableName; + if (tableIdentifier.names.size() == 1) { + tableName = tableIdentifier.names.get(0); + } else if (tableIdentifier.names.size() == 2) { + final String defaultSchemaName = + Iterables.getOnlyElement(CalciteSchema.from(handlerContext.defaultSchema()).path(null)); + if (!defaultSchemaName.equals(tableIdentifier.names.get(0))) { + throw InvalidSqlInput.exception( + "Table [%s] does not support operation [%s] because it is not a Druid datasource", + tableIdentifier, + operationName() + ); + } + tableName = tableIdentifier.names.get(1); + } else { + throw InvalidSqlInput.exception( + "Table name [%s] is not valid for operation [%s]", + tableIdentifier, + operationName() + ); + } + IdUtils.validateId("table", tableName); + return tableName; + } + + /** + * Convert a parsed column declaration into its catalog form, checking that the type is one Druid can store. + */ + protected static ColumnSpec toColumnSpec(DruidSqlColumnDeclaration declaration) + { + final String name = simpleName(declaration.getName(), "Column"); + final String type; + try { + type = CatalogColumnTypes.forCatalogColumn(name, declaration.getDataType()); + } + catch (IAE e) { + throw InvalidSqlInput.exception(e, "%s", e.getMessage()); + } + if (Columns.isTimeColumn(name) && !ColumnType.LONG.equals(Columns.druidTypeFromString(type))) { + throw InvalidSqlInput.exception( + "Column [%s] must have type [%s] or [%s], but was [%s]", + Columns.TIME_COLUMN, + Columns.SQL_TIMESTAMP, + Columns.SQL_BIGINT, + type + ); + } + return new ColumnSpec(name, type, null); + } + + /** + * Translate one projection definition into the catalog form. + * <p> + * {@code __base} is reserved for the base-table projection, which is a different catalog entity: it describes the + * physical layout of the table itself rather than an additional aggregate. It is rejected here rather than being + * translated as an ordinary projection. + */ + protected static DatasourceProjectionMetadata translateProjection( + final SqlStatementHandler.HandlerContext handlerContext, + final String tableName, + final List<ColumnSpec> columns, + final SqlProjectionSpec projection + ) + { + final String name = simpleName(projection.getName(), "Projection"); + try { + Projections.validateProjectionName(name); + } + catch (DruidException e) { + throw InvalidSqlInput.exception(e, "%s", e.getMessage()); + } + if (projection.getClusteredBy() != null) { + throw InvalidSqlInput.exception( + "Projection [%s] cannot use CLUSTERED BY: an aggregate projection is ordered by its grouping columns." + + " Only the [%s] projection, which describes the table's own layout, chooses a clustering", + name, + BASE_PROJECTION_NAME + ); + } + return new DatasourceProjectionMetadata( + new ProjectionSpecTranslator(handlerContext.plannerFactory()) + .translate(tableName, columns, name, projection.getBody()) + ); + } + + /** + * Translate the reserved {@code __base} projection, which describes the physical layout of the table rather than an + * additional aggregate, and so becomes the {@code baseTable} property instead of one of the projections. + */ + protected static ClusteredValueGroupsBaseTableMetadata translateBaseTable( + final SqlStatementHandler.HandlerContext handlerContext, + final String tableName, + final List<ColumnSpec> columns, + final SqlProjectionSpec projection + ) + { + return new ProjectionSpecTranslator(handlerContext.plannerFactory()) + .translateBaseTable(tableName, columns, projection.getBody(), projection.getClusteredBy()); + } + + /** + * A base table layout derives the physical segment schema from the declared columns, so a column that is not + * declared cannot be stored. The catalog enforces this too, but saying it here names the clause that is missing. + */ + protected static void requireSealed(boolean sealed) + { + if (!sealed) { + throw InvalidSqlInput.exception( + "A table with a [%s] projection must be declared SEALED: its columns define the physical segment schema," + + " so columns that are not declared cannot be ingested", + BASE_PROJECTION_NAME + ); + } + } + + protected static String simpleName(SqlIdentifier identifier, String what) + { + if (!identifier.isSimple()) { + throw InvalidSqlInput.exception("%s name [%s] must be a simple name", what, identifier); + } + return identifier.getSimple(); + } + + /** + * The catalog stores a segment granularity as either {@code ALL} or an ISO period string. + */ + protected static String toGranularityString(SqlGranularityLiteral partitionedBy) + { + final Granularity granularity = partitionedBy.getGranularity(); + if (Granularities.ALL.equals(granularity)) { + return DatasourceDefn.ALL_GRANULARITY; + } + if (granularity instanceof PeriodGranularity) { + return ((PeriodGranularity) granularity).getPeriod().toString(); + } + throw InvalidSqlInput.exception("Granularity [%s] is not supported by the catalog", partitionedBy); + } + + /** + * The catalog's clustering keys are plain ascending column references. Expressions, ordinals and DESC have no + * catalog representation, so they are rejected here rather than silently dropped. + */ + protected static List<ClusterKeySpec> toClusterKeys(SqlNodeList clusteredBy) + { + final List<ClusterKeySpec> keys = new ArrayList<>(clusteredBy.size()); + for (SqlNode node : clusteredBy) { + if (!(node instanceof SqlIdentifier) || !((SqlIdentifier) node).isSimple()) { + throw InvalidSqlInput.exception( + "CLUSTERED BY column [%s] must be a column name; expressions, ordinals and DESC are not supported when" + + " defining a table", + node + ); + } + keys.add(new ClusterKeySpec(((SqlIdentifier) node).getSimple(), false)); + } + return keys; + } + + private static RelDataType resultType() + { + final RelDataTypeFactory typeFactory = DruidTypeSystem.TYPE_FACTORY; + return typeFactory.createStructType( + ImmutableList.of(Calcites.createSqlType(typeFactory, SqlTypeName.VARCHAR)), + ImmutableList.of("RESULT") + ); + } + + /** + * {@code CREATE [OR REPLACE] TABLE [IF NOT EXISTS] ...}. + */ + public static class CreateTableHandler extends CatalogDdlHandler + { + private final DruidSqlCreateTable createTable; + private TableSpec tableSpec; + + public CreateTableHandler(SqlStatementHandler.HandlerContext handlerContext, DruidSqlCreateTable createTable) + { + super(handlerContext, createTable.getName()); + this.createTable = createTable; + } + + @Override + protected void validateStatement() + { + if (createTable.getReplace() && createTable.isIfNotExists()) { + throw InvalidSqlInput.exception("Cannot specify both OR REPLACE and IF NOT EXISTS"); + } + + final List<ColumnSpec> columns = new ArrayList<>(createTable.getColumnList().size()); + final Set<String> seen = new HashSet<>(); + for (SqlNode node : createTable.getColumnList()) { + final ColumnSpec column = toColumnSpec((DruidSqlColumnDeclaration) node); + if (!seen.add(column.name())) { + throw InvalidSqlInput.exception("Column [%s] is declared more than once", column.name()); + } + columns.add(column); + } + + final Map<String, Object> properties = new LinkedHashMap<>(); + if (createTable.getPartitionedBy() != null) { + properties.put( + DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY, + toGranularityString(createTable.getPartitionedBy()) + ); + } + if (createTable.getClusteredBy() != null) { + properties.put(DatasourceDefn.CLUSTER_KEYS_PROPERTY, toClusterKeys(createTable.getClusteredBy())); + } + if (createTable.isSealed()) { + properties.put(DatasourceDefn.SEALED_PROPERTY, true); + } + if (!createTable.getProjectionList().isEmpty()) { + final List<DatasourceProjectionMetadata> projections = + new ArrayList<>(createTable.getProjectionList().size()); + final Set<String> seenProjections = new HashSet<>(); + for (SqlNode node : createTable.getProjectionList()) { + final SqlProjectionSpec projection = (SqlProjectionSpec) node; + final String name = simpleName(projection.getName(), "Projection"); + if (!seenProjections.add(name)) { + throw InvalidSqlInput.exception("Projection [%s] is declared more than once", name); + } + if (BASE_PROJECTION_NAME.equals(name)) { + requireSealed(createTable.isSealed()); + properties.put( + DatasourceDefn.BASE_TABLE_PROPERTY, + translateBaseTable(handlerContext, tableId.name(), columns, projection) + ); + } else { + projections.add(translateProjection(handlerContext, tableId.name(), columns, projection)); + } + } + if (!projections.isEmpty()) { + properties.put(DatasourceDefn.PROJECTIONS_KEYS_PROPERTY, projections); + } + } + + tableSpec = new TableSpec(DatasourceDefn.TABLE_TYPE, properties, columns); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.createTable(tableId, tableSpec, createTable.isIfNotExists(), createTable.getReplace()); + } + + @Override + protected String operationName() + { + return "CREATE TABLE"; + } + } + + /** + * {@code ALTER TABLE ... ADD COLUMN}. The Coordinator merges columns by name, so an existing column would be + * silently updated; this checks first so that {@code ADD} means add. + */ + public static class AddColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.AddColumn alterTable; + private ColumnSpec column; + + public AddColumnHandler(SqlStatementHandler.HandlerContext handlerContext, DruidSqlAlterTable.AddColumn alterTable) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = toColumnSpec(alterTable.getColumn()); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + final TableMetadata existing = writer.readTable(tableId); + if (existing == null) { + throw InvalidSqlInput.exception("Table [%s] does not have a catalog entry", tableId.name()); + } + if (existing.spec().columns() != null + && existing.spec().columns().stream().anyMatch(c -> column.name().equals(c.name()))) { + throw InvalidSqlInput.exception( + "Column [%s] already exists in table [%s]; use ALTER COLUMN to change its type", + column.name(), + tableId.name() + ); + } + writer.updateColumns(tableId, Collections.singletonList(column)); + } + + @Override + protected String operationName() + { + return "ALTER TABLE ADD COLUMN"; + } + } + + /** + * {@code ALTER TABLE ... DROP COLUMN}. + */ + public static class DropColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.DropColumn alterTable; + private String column; + + public DropColumnHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.DropColumn alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = simpleName(alterTable.getColumn(), "Column"); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.dropColumns(tableId, Collections.singletonList(column)); + } + + @Override + protected String operationName() + { + return "ALTER TABLE DROP COLUMN"; + } + } + + /** + * {@code ALTER TABLE ... ALTER COLUMN ... SET DATA TYPE}. Merging a column by name is exactly what changing its + * type requires, so this reuses the same update as ADD COLUMN without the existence check. + */ + public static class AlterColumnHandler extends CatalogDdlHandler + { + private final DruidSqlAlterTable.AlterColumn alterTable; + private ColumnSpec column; + + public AlterColumnHandler( + SqlStatementHandler.HandlerContext handlerContext, + DruidSqlAlterTable.AlterColumn alterTable + ) + { + super(handlerContext, alterTable.getName()); + this.alterTable = alterTable; + } + + @Override + protected void validateStatement() + { + column = toColumnSpec(alterTable.getColumn()); + } + + @Override + protected void execute(CatalogTableWriter writer) + { + writer.updateColumns(tableId, Collections.singletonList(column)); Review Comment: [P1] Enforce column operation predicates atomically UpdateColumns appends a column when its name is absent, so ALTER TABLE t ALTER COLUMN typo SET DATA TYPE BIGINT silently adds typo instead of rejecting the nonexistent target. ADD COLUMN has the inverse predicate checked by a separate Broker read, allowing concurrent ADDs to both pass and the later merge to overwrite the first type. Add/alter existence semantics need dedicated checks inside the Coordinator's column-update transaction. ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/ProjectionSpecTranslator.java: ########## @@ -0,0 +1,546 @@ +/* + * 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.sql.calcite.planner; + +import com.google.common.collect.ImmutableMap; +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlSelect; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.util.SqlBasicVisitor; +import org.apache.druid.catalog.model.ClusteredValueGroupsBaseTableMetadata; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.Columns; +import org.apache.druid.data.input.impl.AggregateProjectionSpec; +import org.apache.druid.data.input.impl.DimensionSchema; +import org.apache.druid.error.DruidException; +import org.apache.druid.error.InvalidSqlInput; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.math.expr.ExprMacroTable; +import org.apache.druid.query.DataSource; +import org.apache.druid.query.Query; +import org.apache.druid.query.QueryContexts; +import org.apache.druid.query.TableDataSource; +import org.apache.druid.query.aggregation.AggregatorFactory; +import org.apache.druid.query.dimension.DefaultDimensionSpec; +import org.apache.druid.query.dimension.DimensionSpec; +import org.apache.druid.query.filter.AndDimFilter; +import org.apache.druid.query.filter.DimFilter; +import org.apache.druid.query.filter.RangeFilter; +import org.apache.druid.query.groupby.GroupByQuery; +import org.apache.druid.query.scan.ScanQuery; +import org.apache.druid.query.timeseries.TimeseriesQuery; +import org.apache.druid.segment.VirtualColumn; +import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.server.security.AuthorizationResult; +import org.apache.druid.server.security.NoopEscalator; +import org.apache.druid.sql.calcite.rel.DruidQuery; +import org.apache.druid.sql.calcite.rel.Grouping; +import org.apache.druid.sql.calcite.table.DatasourceTable; +import org.apache.druid.sql.calcite.table.DatasourceTable.PhysicalDatasourceMetadata; +import org.apache.druid.sql.calcite.table.DruidTable; +import org.joda.time.Interval; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * Turns the SQL body of a projection definition into the {@link AggregateProjectionSpec} the catalog stores. + * <p> + * The body is planned through the normal pipeline against the columns the enclosing statement declares, and the + * specification is lifted out of the resulting native query. Going through the planner is the point: a projection is + * only useful if it matches the queries the planner generates at query time, and that agreement is guaranteed when + * the same machinery produces both. It also means aggregators contributed by extensions work without a second + * registry. + */ +public class ProjectionSpecTranslator +{ + /** + * The reserved projection name that describes the table's own physical layout. + */ + public static final String BASE_PROJECTION_NAME = "__base"; + + /** + * Planning is deterministic in the shapes the lift understands: no timeseries or topN rewrite to hide the grouping + * columns, and no approximation choices that depend on unrelated configuration. + */ + private static final Map<String, Object> CONTEXT = ImmutableMap.of( + PlannerContext.CTX_SQL_USE_GRANULARITY, false, + QueryContexts.TIME_BOUNDARY_PLANNING_KEY, false, + PlannerConfig.CTX_KEY_USE_APPROXIMATE_TOPN, false + ); + + private final PlannerFactory plannerFactory; + + public ProjectionSpecTranslator(PlannerFactory plannerFactory) + { + this.plannerFactory = plannerFactory; + } + + /** + * Translate one projection definition. + * + * @param tableName the table the projection belongs to + * @param columns the table's declared columns, which are the only ones the body may reference + */ + public AggregateProjectionSpec translate( + final String tableName, + final List<ColumnSpec> columns, + final String projectionName, + final SqlSelect body + ) + { + rejectSubqueries(projectionName, body); + + final DruidQuery druidQuery = planBody(tableName, columns, projectionName, body); + return lift(projectionName, tableName, druidQuery); + } + + /** + * Translate the reserved base-table projection, which describes the physical layout of the table itself rather + * than an additional aggregate. + * <p> + * The body enumerates the table's columns in the order segments store them, so it must name every declared column, + * in declared order. An item written as {@code <expr> AS <name>} makes that column computed at ingest time: the + * expression becomes a virtual column materializing the declared column, which is why the declared type has to + * match what the expression produces. + * + * @param clusteredBy the columns segments are clustered on, which must be the leading prefix of the column list + */ + public ClusteredValueGroupsBaseTableMetadata translateBaseTable( + final String tableName, + final List<ColumnSpec> columns, + final SqlSelect body, + @Nullable final SqlNodeList clusteredBy + ) + { + if (body.getWhere() != null || body.getGroup() != null) { + throw invalid( + BASE_PROJECTION_NAME, + "its body filters or groups. The base table stores every ingested row, so it can do neither" + ); + } + rejectSubqueries(BASE_PROJECTION_NAME, body); + + final DruidQuery druidQuery = planBody(tableName, columns, BASE_PROJECTION_NAME, body); + final ClusteredValueGroupsBaseTableMetadata metadata = new ClusteredValueGroupsBaseTableMetadata( + clusteringColumns(clusteredBy), + liftComputedColumns(columns, druidQuery), + null + ); + + // Derive the physical spec now. The catalog does this too when the write lands, but doing it here attributes + // layout problems to the statement that caused them rather than to a Coordinator round trip. + try { + metadata.createSpec(columns); + } + catch (DruidException e) { + throw contextualize(BASE_PROJECTION_NAME, e); + } + return metadata; + } + + private static List<String> clusteringColumns(@Nullable final SqlNodeList clusteredBy) + { + if (clusteredBy == null) { + return Collections.emptyList(); + } + final List<String> names = new ArrayList<>(clusteredBy.size()); + for (SqlNode node : clusteredBy) { + if (!(node instanceof SqlIdentifier) || !((SqlIdentifier) node).isSimple()) { + throw invalid( + BASE_PROJECTION_NAME, + "its CLUSTERED BY names [" + node + "], which is not a column. Segments are clustered on stored columns;" + + " to cluster on a computed value, declare it as a column of the table" + ); + } + names.add(((SqlIdentifier) node).getSimple()); + } + return names; + } + + /** + * Pair the planned output with the declared columns and lift the virtual columns behind the computed ones. + * <p> + * The planner names its virtual columns {@code v0}, {@code v1}, ...; each is renamed to the declared column it + * fills, which is what makes it a materialized column rather than an anonymous intermediate. + */ + private static VirtualColumns liftComputedColumns( + final List<ColumnSpec> columns, + final DruidQuery druidQuery + ) + { + final Query<?> query = druidQuery.getQuery(); + if (!(query instanceof ScanQuery)) { + throw invalid( + BASE_PROJECTION_NAME, + "its body does not select rows directly. The base table stores every ingested row as it arrives" + ); + } + final List<String> selected = ((ScanQuery) query).getColumns(); + final List<String> outputNames = druidQuery.getOutputRowType().getFieldNames(); + + if (outputNames.size() != columns.size()) { + throw invalid( + BASE_PROJECTION_NAME, + StringUtils.format( + "it selects %d column(s) but the table declares %d. The body lists the columns in the order segments" + + " store them, so it must name every declared column", + outputNames.size(), + columns.size() + ) + ); + } + + final VirtualColumns planned = ((ScanQuery) query).getVirtualColumns(); + final List<VirtualColumn> materialized = new ArrayList<>(); + for (int i = 0; i < columns.size(); i++) { + final String declared = columns.get(i).name(); + if (!declared.equals(outputNames.get(i))) { + throw invalid( + BASE_PROJECTION_NAME, + StringUtils.format( + "its column %d is [%s] but the table declares [%s] there. The body lists the columns in the order" + + " segments store them", + i + 1, + outputNames.get(i), + declared + ) + ); + } + final VirtualColumn virtualColumn = planned.getVirtualColumn(selected.get(i)); Review Comment: [P2] Handle direct aliases in base projections ScanQuery deduplicates its column list, while outputNames retains every SELECT item. A valid declared layout [id, copy] with SELECT id, id AS copy therefore has two outputs but only one selected entry, and the second iteration throws IndexOutOfBoundsException; a direct alias also has no virtual column to materialize copy. Preserve the select-to-source mapping or reject this form with a user-facing validation error. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
