gianm commented on code in PR #19830: URL: https://github.com/apache/druid/pull/19830#discussion_r3858573245
########## extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/http/TableEditor.java: ########## Review Comment: This could probably use pattern matching to be cooler. ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogDdlHandler.java: ########## @@ -0,0 +1,690 @@ +/* + * 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()); Review Comment: It's sketchy to execute the operation in `plan()`. It should happen only when `PlannerResult#run` is called. Please also include a test that when the DDL operation is unauthorized, no methods are called on the catalog client. ########## extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogSqlTableWriter.java: ########## @@ -0,0 +1,216 @@ +/* + * 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.sync; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; +import org.apache.druid.catalog.CatalogException; +import org.apache.druid.catalog.http.TableEditRequest; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.DatasourceBaseTableMetadata; +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.error.DruidException; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.rpc.HttpResponseException; +import org.apache.druid.sql.calcite.planner.CatalogTableWriter; + +import javax.annotation.Nullable; +import javax.inject.Inject; +import java.util.List; +import java.util.Map; + +/** + * Applies catalog DDL by calling the Coordinator, which owns catalog metadata. + * <p> + * The Coordinator is authoritative for validation, so the errors it reports are unwrapped and presented as the + * statement's own error rather than as a failed HTTP call. + */ +public class CatalogSqlTableWriter implements CatalogTableWriter +{ + private static final Logger LOG = new Logger(CatalogSqlTableWriter.class); + + private final CatalogClient client; + private final CachedMetadataCatalog cache; + private final ObjectMapper jsonMapper; + + @Inject + public CatalogSqlTableWriter( + final CatalogClient client, + final CachedMetadataCatalog cache, + final ObjectMapper jsonMapper + ) + { + this.client = client; + this.cache = cache; + this.jsonMapper = jsonMapper; + } + + @Override + public void createTable(TableId tableId, TableSpec spec, boolean ifNotExists, boolean replace) + { + execute(tableId, () -> client.createTable(tableId, spec, ifNotExists, replace)); + } + + @Override + public void addColumns(TableId tableId, List<ColumnSpec> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AddColumns(columns))); + } + + @Override + public void alterColumns(TableId tableId, List<ColumnSpec> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AlterColumns(columns))); + } + + @Override + public void dropColumns(TableId tableId, List<String> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropColumns(columns))); + } + + @Override + public void updateProperties(TableId tableId, Map<String, Object> properties) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.UpdateProperties(properties))); + } + + @Override + public void addProjection(TableId tableId, DatasourceProjectionMetadata projection, boolean ifNotExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AddProjection(projection, ifNotExists))); + } + + @Override + public void dropProjection(TableId tableId, String projectionName, boolean ifExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropProjection(projectionName, ifExists))); + } + + @Override + public void setBaseTable(TableId tableId, DatasourceBaseTableMetadata baseTable, boolean ifNotExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.SetBaseTable(baseTable, ifNotExists))); + } + + @Override + public void dropBaseTable(TableId tableId, boolean ifExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropBaseTable(ifExists))); + } + + @Nullable + @Override + public TableMetadata readTable(TableId tableId) + { + return client.table(tableId); + } + + /** + * Run a catalog write, translate any failure into a statement error, then refresh this Broker's cache. + * <p> + * The refresh matters because the Coordinator's update notification is asynchronous: without it, a CREATE TABLE + * followed immediately by an INSERT on the same connection could plan against the pre-DDL schema. Other Brokers + * still converge through the normal notification and polling path. + */ + private void execute(TableId tableId, Runnable operation) + { + try { + operation.run(); + } + catch (Exception e) { + throw translateError(tableId, e); + } + refreshCache(tableId); + } + + private void refreshCache(TableId tableId) + { + try { + final TableMetadata table = client.table(tableId); + cache.updated( Review Comment: Is this just needed so the Broker can immediately see its own writes? I guess it should eventually update its cache even without this code. ########## extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/http/TableEditor.java: ########## @@ -111,14 +135,37 @@ public long go() throws CatalogException } } + /** + * Validate the revised spec as a whole before it is written back. Every edit is a read-modify-write of one part of Review Comment: I don't think there's really a need to justify the concept of validating the entire table as a unit. ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/CatalogTableWriter.java: ########## @@ -0,0 +1,187 @@ +/* + * 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 org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.DatasourceBaseTableMetadata; +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.error.DruidException; + +import javax.annotation.Nullable; +import java.util.List; +import java.util.Map; + +/** + * Write side of the catalog, the counterpart to the read-only {@link CatalogResolver}. Catalog metadata is owned by + * the Coordinator, so an implementation of this interface makes a remote call; it is bound by the {@code druid-catalog} + * extension and defaults to {@link #NOT_AVAILABLE} when that extension is absent. + * <p> + * The methods are deliberately semantic rather than a generic "apply this edit request", so that the extension's edit + * request types need not be visible here. Each corresponds to exactly one atomic Coordinator operation, which is why Review Comment: I'm not sure this is actually better than moving the extension's edit types into core. With the approach currently in the PR, the edit types don't need to be here, but we have a 1-1 coupling anyway where there needs to be one method here for each edit type. Seems like a judgment call though. Thoughts? ########## extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/sync/CatalogSqlTableWriter.java: ########## @@ -0,0 +1,216 @@ +/* + * 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.sync; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; +import org.apache.druid.catalog.CatalogException; +import org.apache.druid.catalog.http.TableEditRequest; +import org.apache.druid.catalog.model.ColumnSpec; +import org.apache.druid.catalog.model.DatasourceBaseTableMetadata; +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.error.DruidException; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.rpc.HttpResponseException; +import org.apache.druid.sql.calcite.planner.CatalogTableWriter; + +import javax.annotation.Nullable; +import javax.inject.Inject; +import java.util.List; +import java.util.Map; + +/** + * Applies catalog DDL by calling the Coordinator, which owns catalog metadata. + * <p> + * The Coordinator is authoritative for validation, so the errors it reports are unwrapped and presented as the + * statement's own error rather than as a failed HTTP call. + */ +public class CatalogSqlTableWriter implements CatalogTableWriter +{ + private static final Logger LOG = new Logger(CatalogSqlTableWriter.class); + + private final CatalogClient client; + private final CachedMetadataCatalog cache; + private final ObjectMapper jsonMapper; + + @Inject + public CatalogSqlTableWriter( + final CatalogClient client, + final CachedMetadataCatalog cache, + final ObjectMapper jsonMapper + ) + { + this.client = client; + this.cache = cache; + this.jsonMapper = jsonMapper; + } + + @Override + public void createTable(TableId tableId, TableSpec spec, boolean ifNotExists, boolean replace) + { + execute(tableId, () -> client.createTable(tableId, spec, ifNotExists, replace)); + } + + @Override + public void addColumns(TableId tableId, List<ColumnSpec> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AddColumns(columns))); + } + + @Override + public void alterColumns(TableId tableId, List<ColumnSpec> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AlterColumns(columns))); + } + + @Override + public void dropColumns(TableId tableId, List<String> columns) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropColumns(columns))); + } + + @Override + public void updateProperties(TableId tableId, Map<String, Object> properties) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.UpdateProperties(properties))); + } + + @Override + public void addProjection(TableId tableId, DatasourceProjectionMetadata projection, boolean ifNotExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.AddProjection(projection, ifNotExists))); + } + + @Override + public void dropProjection(TableId tableId, String projectionName, boolean ifExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropProjection(projectionName, ifExists))); + } + + @Override + public void setBaseTable(TableId tableId, DatasourceBaseTableMetadata baseTable, boolean ifNotExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.SetBaseTable(baseTable, ifNotExists))); + } + + @Override + public void dropBaseTable(TableId tableId, boolean ifExists) + { + execute(tableId, () -> client.editTable(tableId, new TableEditRequest.DropBaseTable(ifExists))); + } + + @Nullable + @Override + public TableMetadata readTable(TableId tableId) + { + return client.table(tableId); + } + + /** + * Run a catalog write, translate any failure into a statement error, then refresh this Broker's cache. + * <p> + * The refresh matters because the Coordinator's update notification is asynchronous: without it, a CREATE TABLE + * followed immediately by an INSERT on the same connection could plan against the pre-DDL schema. Other Brokers + * still converge through the normal notification and polling path. + */ + private void execute(TableId tableId, Runnable operation) + { + try { + operation.run(); + } + catch (Exception e) { + throw translateError(tableId, e); + } + refreshCache(tableId); + } + + private void refreshCache(TableId tableId) + { + try { + final TableMetadata table = client.table(tableId); + cache.updated( + new UpdateEvent( + table == null ? UpdateEvent.EventType.DELETE : UpdateEvent.EventType.UPDATE, Review Comment: I think doing every change as an `UPDATE` is going to confuse the `CachedMetadataCatalog`. It will cause `computeUpdate` to run, which expects to see a pre-existing object and does a `LOG.error` if it doesn't see one. In some cases it won't see one, like `CREATE TABLE`. ########## extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/storage/sql/SQLCatalogManager.java: ########## @@ -453,44 +436,47 @@ public TableMetadata withHandle(Handle handle) throws CatalogException { handle.begin(); try { - final Query<Map<String, Object>> query = handle - .createQuery(statement(SELECT_COLUMNS_STMT)) + final ResultIterator<TableMetadata> resultIterator = handle + .createQuery(statement(selectStmt)) .setFetchSize(connector.getStreamingFetchSize()) .bind(SCHEMA_NAME_COL, id.schema()) - .bind(TABLE_NAME_COL, id.name()); - - final ResultIterator<TableSpec> resultIterator = query - .map((index, r, ctx) -> - tableSpecFromBytes( - jsonMapper, - r.getString(1), - null, - r.getBytes(2) - ) - ) - .iterator(); - final TableSpec tableSpec; + .bind(TABLE_NAME_COL, id.name()) + .map((index, r, ctx) -> + TableMetadata.forUpdate( + id, + r.getLong(4), + tableSpecFromBytes(jsonMapper, r.getString(1), r.getBytes(2), r.getBytes(3)) + ) + ) + .iterator(); + final TableMetadata existing; if (resultIterator.hasNext()) { - tableSpec = resultIterator.next(); + existing = resultIterator.next(); } else { throw tableNotFound(id); } - final TableSpec revised = transform.apply(TableMetadata.of(id, tableSpec)); + final TableSpec revised = transform.apply(existing); if (revised == null) { handle.rollback(); return null; } - final long updateTime = System.currentTimeMillis(); + // The version is also the compare-and-set token, so it must actually change on every write: a commit + // landing in the same millisecond as the write that produced the version it read would otherwise + // leave the token as it was, letting a second writer's predicate match after this one commits. + final long updateTime = Math.max(System.currentTimeMillis(), existing.updateTime() + 1); Review Comment: Should become a helper function that gets applied to all the `UPDATE_TIME_COL` usages. There's a few others. ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java: ########## @@ -147,6 +152,16 @@ private SqlStatementHandler createHandler(final SqlNode node) } SqlStatementHandler.HandlerContext handlerContext = new HandlerContextImpl(); + + if (query instanceof DruidSqlCreateTable || query instanceof DruidSqlAlterTable) { + // The grammar does not admit EXPLAIN of a DDL statement; this guards the case anyway, since a DDL statement Review Comment: Please include a unit test for this, if it does not already exist. (I haven't checked yet.) ########## sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java: ########## @@ -147,6 +152,16 @@ private SqlStatementHandler createHandler(final SqlNode node) } SqlStatementHandler.HandlerContext handlerContext = new HandlerContextImpl(); + + if (query instanceof DruidSqlCreateTable || query instanceof DruidSqlAlterTable) { + // The grammar does not admit EXPLAIN of a DDL statement; this guards the case anyway, since a DDL statement + // has no query to explain. + if (explain != null) { + throw InvalidSqlInput.exception("EXPLAIN is not supported for [%s]", query.getKind()); + } + return createDdlHandler(handlerContext, query); Review Comment: I *think* structuring it this way will cause DDL to be accepted by all engines. It makes me wonder what we want. I *think* what we want is for the Task engine, which is meant to be async/offline, to reject DDL. I think we want the interactive ones (native, Dart) to accept it. Consider adding `EngineFeature#CAN_DDL` (and maybe consolidate `CAN_INSERT` and `CAN_REPLACE` into `CAN_DML` while you're at it) so we can do this. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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]
