AHeise commented on code in PR #29838: URL: https://github.com/apache/flink/pull/29838#discussion_r4537097668
########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/MaterializedPipelineImpl.java: ########## @@ -0,0 +1,124 @@ +/* + * 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.flink.table.api.internal.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.MaterializedPipeline; +import org.apache.flink.table.api.MaterializedTable; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable.TableKind; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.materializedtable.MaterializedTableOperationFactory; +import org.apache.flink.table.operations.materializedtable.MaterializedTableOperationFactory.Mode; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.util.Optional; + +/** + * Implementation of {@link MaterializedPipeline}. It serializes the query when it is constructed + * and builds the operation against the catalog state at execute time. + */ +@Internal +public final class MaterializedPipelineImpl implements MaterializedPipeline { + + private final TableEnvironmentInternal tableEnvironment; + private final QueryOperation queryOperation; + private final String serializedQuery; + private final ObjectIdentifier identifier; + private final @Nullable IntervalFreshness freshness; + + public MaterializedPipelineImpl( + TableEnvironmentInternal tableEnvironment, + QueryOperation queryOperation, + ObjectIdentifier identifier, + @Nullable IntervalFreshness freshness) { + this.tableEnvironment = + Preconditions.checkNotNull(tableEnvironment, "Table environment must not be null."); + this.queryOperation = + Preconditions.checkNotNull(queryOperation, "Query operation must not be null."); + this.identifier = Preconditions.checkNotNull(identifier, "Identifier must not be null."); + this.freshness = freshness; + this.serializedQuery = serializeQuery(queryOperation, tableEnvironment.getCatalogManager()); + } + + private static String serializeQuery( + QueryOperation queryOperation, CatalogManager catalogManager) { + try { + return queryOperation.asSerializableString(catalogManager.getSqlFactory()); + } catch (UnsupportedOperationException | TableException | ValidationException e) { + final String cause = e.getMessage() != null ? e.getMessage() : e.getClass().getName(); + throw new ValidationException( + "The query of a materialized table must be expressible in SQL. The refresh job" + + " re-runs the stored query text, which rules out, for example," + + " functions that are not registered in a catalog, tables created from" + + " a TableDescriptor and tables backed by a DataStream. Cause: " + + cause, + e); + } + } Review Comment: FLIP-616 ("Query restrictions") and FLINK-41214 both say temporary functions, tables and views are rejected with a `ValidationException` when the declaration is built. `asSerializableString` doesn't catch them: temporary catalog functions and temporary tables serialize to their identifiers. So this only rejects what can't be serialized, and the docs (`table-api.md:71-72`) explicitly say they're not rejected. Either walk the `QueryOperation` tree here (temporary `ContextResolvedTable`s, temporary function identifiers), or update the FLIP/JIRA and file a follow-up. I'd prefer the former, since a refresh job that can't resolve the query only fails much later. ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/QueryMaterializedTableMergeContext.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.flink.table.api.internal.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.StartMode; +import org.apache.flink.table.catalog.TableDistribution; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.materializedtable.MaterializedTableMergeContext; +import org.apache.flink.table.types.DataType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType; + +/** + * The {@link MaterializedTableMergeContext} of {@code Table#materializeInto}, with the semantics of + * the equivalent {@code CREATE OR ALTER MATERIALIZED TABLE} statement. + */ +@Internal +final class QueryMaterializedTableMergeContext implements MaterializedTableMergeContext { + + private final QueryOperation queryOperation; + private final String serializedQuery; + private final @Nullable IntervalFreshness freshness; + private final ResolvedSchema querySchema; + + QueryMaterializedTableMergeContext( + QueryOperation queryOperation, + String serializedQuery, + @Nullable IntervalFreshness freshness) { + this.queryOperation = queryOperation; + this.serializedQuery = serializedQuery; + this.freshness = freshness; + this.querySchema = toPhysicalSchema(queryOperation.getResolvedSchema()); Review Comment: The stored schema comes from the `QueryOperation`, while the refresh job runs `serializedQuery` re-parsed. In continuous mode a mismatch fails at submit inside `execute()`, but silent differences (nullability, precision, time attributes) would only show up in the stored schema. Could `materializeInto` parse `serializedQuery` once and check its schema against `querySchema`? That would also catch serializer bugs early. ########## docs/content/docs/sql/materialized-table/table-api.md: ########## @@ -0,0 +1,72 @@ +--- +title: Table API +weight: 5 +type: docs +--- +<!-- +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. +--> + +`Table#materializeInto` declares the query of a `Table` as the definition of a materialized table, +and `execute()` submits the declaration with the semantics of +[`CREATE OR ALTER MATERIALIZED TABLE ... AS <query>`]({{< ref "docs/sql/materialized-table/statements" >}}#or-alter). + +# Setup + +Continuous refresh jobs need a session cluster as the execution target of the `TableEnvironment`, +and redeploying a running materialized table needs a savepoint directory. See +[Using Table API]({{< ref "docs/sql/materialized-table/deployment" >}}#using-table-api) for the +supported targets. The catalog must supply the storage of the materialized table. + +# Creating a Materialized Table + +```java +MaterializedTable revenue = tEnv.from("orders") + .groupBy($("region")) + .select($("region"), $("amount").sum().as("revenue")) + .materializeInto("revenue", Duration.ofMinutes(1)) + .execute(); +``` + +The schema is derived from the query, and the path is resolved against the current catalog and +database. The freshness is optional and defaults to +[materialized-table.default-freshness.continuous]({{< ref "docs/dev/table/config" >}}#materialized-table-default-freshness-continuous). +The refresh mode is derived from the freshness (see [Refresh Mode]({{< ref "docs/sql/materialized-table/overview" >}}#refresh-mode)). +Full refresh mode needs a workflow scheduler and cannot be used from a plain `TableEnvironment`. + +# Redeploying + +`execute()` on an existing materialized table alters it, and a changed query follows the +schema-evolution rules of +[ALTER MATERIALIZED TABLE AS]({{< ref "docs/sql/materialized-table/statements" >}}#as-select_statement-1). +A running continuous refresh job is stopped with a savepoint and restarted without restoring it, +even if the declaration is unchanged, so streaming state is reset. + +If the changes are rejected, for example because the new query drops a column, `execute()` fails +and leaves the materialized table suspended with its previous definition. Resume it with +`ALTER MATERIALIZED TABLE ... RESUME` before you redeploy, because redeploying a suspended +materialized table clears its savepoint. + +# Limitations + +- The query of a refresh job is stored as SQL text, so it must be expressible in SQL. Inline + functions (unless a custom `SqlFactory` serializes them), tables created from a `TableDescriptor` + or a `DataStream`, and operations without a SQL form, such as session group windows, fail at + `materializeInto`. +- As in SQL, referring to temporary tables, views, functions or models is not supported. They are + not rejected, but a refresh job started from another session can't resolve them. Review Comment: Contradicts FLIP-616, see the comment on `MaterializedPipelineImpl#serializeQuery`. ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/QueryMaterializedTableMergeContext.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.flink.table.api.internal.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.StartMode; +import org.apache.flink.table.catalog.TableDistribution; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.materializedtable.MaterializedTableMergeContext; +import org.apache.flink.table.types.DataType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType; + +/** + * The {@link MaterializedTableMergeContext} of {@code Table#materializeInto}, with the semantics of + * the equivalent {@code CREATE OR ALTER MATERIALIZED TABLE} statement. + */ +@Internal +final class QueryMaterializedTableMergeContext implements MaterializedTableMergeContext { + + private final QueryOperation queryOperation; + private final String serializedQuery; + private final @Nullable IntervalFreshness freshness; + private final ResolvedSchema querySchema; + + QueryMaterializedTableMergeContext( + QueryOperation queryOperation, + String serializedQuery, + @Nullable IntervalFreshness freshness) { + this.queryOperation = queryOperation; + this.serializedQuery = serializedQuery; + this.freshness = freshness; + this.querySchema = toPhysicalSchema(queryOperation.getResolvedSchema()); + } + + @Override + public boolean hasSchemaDefinition() { + return false; + } + + @Override + public boolean hasConstraintDefinition() { + return false; + } + + @Override + public Schema getMergedSchema() { + return Schema.newBuilder().fromResolvedSchema(querySchema).build(); + } + + @Override + public ResolvedSchema getQuerySchema() { + return querySchema; + } + + @Override + public Map<String, String> getOptions() { + return Map.of(); + } + + @Override + public List<String> getPartitionKeys() { + return List.of(); + } + + @Override + public Optional<TableDistribution> getDistribution() { + return Optional.empty(); + } + + @Override + public @Nullable String getComment() { + return null; + } + + @Override + public @Nullable IntervalFreshness getFreshness() { + return freshness; + } + + @Override + public LogicalRefreshMode getLogicalRefreshMode() { + return LogicalRefreshMode.AUTOMATIC; + } + + @Override + public @Nullable StartMode getStartMode() { + return null; + } + + @Override + public String getOriginalQuery() { + return serializedQuery; + } + + @Override + public String getExpandedQuery() { + return serializedQuery; + } + + @Override + public QueryOperation getQueryOperation() { + return queryOperation; + } + + @Override + public QueryOperation alignQuery(ResolvedCatalogMaterializedTable table) { + // The table's schema is the query schema, so the query already lines up with the sink. + return queryOperation; + } + + /** Produces physical columns with default conversion classes. */ + private static ResolvedSchema toPhysicalSchema(ResolvedSchema schema) { + final List<DataType> dataTypes = + schema.getColumnDataTypes().stream() + .map(dataType -> fromLogicalToDataType(dataType.getLogicalType())) + .collect(Collectors.toList()); + return ResolvedSchema.physical(schema.getColumnNames(), dataTypes); + } Review Comment: `fromLogicalToDataType` keeps the full logical type, including `TimestampKind.ROWTIME` on time attributes. Is that intended for the stored physical schema? The `constrained_source` ITCase checks names and the absence of watermarks, but not types. ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/materializedtable/QueryMaterializedTableMergeContext.java: ########## @@ -0,0 +1,148 @@ +/* + * 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.flink.table.api.internal.materializedtable; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; +import org.apache.flink.table.catalog.IntervalFreshness; +import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.StartMode; +import org.apache.flink.table.catalog.TableDistribution; +import org.apache.flink.table.operations.QueryOperation; +import org.apache.flink.table.operations.materializedtable.MaterializedTableMergeContext; +import org.apache.flink.table.types.DataType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType; + +/** + * The {@link MaterializedTableMergeContext} of {@code Table#materializeInto}, with the semantics of + * the equivalent {@code CREATE OR ALTER MATERIALIZED TABLE} statement. + */ +@Internal +final class QueryMaterializedTableMergeContext implements MaterializedTableMergeContext { + + private final QueryOperation queryOperation; + private final String serializedQuery; + private final @Nullable IntervalFreshness freshness; + private final ResolvedSchema querySchema; + + QueryMaterializedTableMergeContext( + QueryOperation queryOperation, + String serializedQuery, + @Nullable IntervalFreshness freshness) { + this.queryOperation = queryOperation; + this.serializedQuery = serializedQuery; + this.freshness = freshness; + this.querySchema = toPhysicalSchema(queryOperation.getResolvedSchema()); + } + + @Override + public boolean hasSchemaDefinition() { + return false; + } + + @Override + public boolean hasConstraintDefinition() { + return false; + } + + @Override + public Schema getMergedSchema() { + return Schema.newBuilder().fromResolvedSchema(querySchema).build(); + } + + @Override + public ResolvedSchema getQuerySchema() { + return querySchema; + } + + @Override + public Map<String, String> getOptions() { + return Map.of(); + } + + @Override + public List<String> getPartitionKeys() { + return List.of(); + } + + @Override + public Optional<TableDistribution> getDistribution() { + return Optional.empty(); + } Review Comment: On an existing materialized table this resets every option and drops the distribution (`getOptionsTableChanges(old, Map.of())` / `dropDistribution()`). SQL `CREATE OR ALTER` without `WITH` behaves the same, but there the user can restate the clauses; through the Table API they can't. A materialized table created in SQL with `WITH`/`DISTRIBUTED BY` therefore silently loses them on the first `materializeInto` redeploy. Either keep them from the existing table when undeclared (as is done for freshness, comment and partition keys), reject the redeploy, or at least document it in the Redeploying section. ########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/MaterializedPipeline.java: ########## @@ -0,0 +1,56 @@ +/* + * 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.flink.table.api; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.table.catalog.ObjectIdentifier; + +/** + * A pending declaration of one materialized table, created by {@link + * Table#materializeInto(String)}. + */ +@PublicEvolving +public interface MaterializedPipeline { + + /** + * The fully qualified identifier this declaration will create or alter. + * + * <p>Resolved against the current catalog and database at {@link Table#materializeInto}. + */ + ObjectIdentifier getIdentifier(); + + /** + * Submits this declaration with the semantics of {@code CREATE OR ALTER MATERIALIZED TABLE}. + * + * <p>If the materialized table exists and its continuous refresh job is running, the job is + * stopped with a savepoint and a new refresh job is started, which does not restore that + * savepoint. This also happens if the declaration is unchanged. The stop requires a savepoint + * directory ({@code execution.checkpointing.savepoint-dir}); without one, this fails before the + * stop. If the changes are rejected or cannot be written after the stop, the materialized table + * is left suspended with its previous definition and the savepoint. + * + * <p>A suspended materialized table is altered and stays suspended, and its stored savepoint is + * cleared, so a later {@code RESUME} starts the refresh job without state. A materialized table + * in full refresh mode is altered without changing its refresh workflow. Review Comment: Two things worth stating here and in the Redeploying docs: - The `Duration` passed to `materializeInto(path, freshness)` is ignored when the table exists. That's per FLIP-616, but a user will expect it to apply. - Options and distribution are reset (see the comment on `QueryMaterializedTableMergeContext`). FYI #29843 adds another ALTER outcome: if the new job can't be cancelled after a failed catalog write, the table is rolled back to SUSPENDED and the old job isn't restarted. Whichever PR lands second should add it here and in `table-api.md`. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/materializedtable/SqlCreateOrAlterMaterializedTableConverter.java: ########## @@ -18,506 +18,38 @@ package org.apache.flink.table.planner.operations.converters.materializedtable; -import org.apache.flink.sql.parser.ddl.SqlTableColumn.SqlRegularColumn; import org.apache.flink.sql.parser.ddl.materializedtable.SqlCreateOrAlterMaterializedTable; -import org.apache.flink.table.api.Schema; -import org.apache.flink.table.api.ValidationException; -import org.apache.flink.table.api.config.TableConfigOptions; -import org.apache.flink.table.catalog.CatalogBaseTable.TableKind; -import org.apache.flink.table.catalog.CatalogMaterializedTable; -import org.apache.flink.table.catalog.CatalogMaterializedTable.LogicalRefreshMode; -import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshMode; -import org.apache.flink.table.catalog.CatalogMaterializedTable.RefreshStatus; -import org.apache.flink.table.catalog.IntervalFreshness; import org.apache.flink.table.catalog.ObjectIdentifier; -import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; -import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable; -import org.apache.flink.table.catalog.ResolvedCatalogTable; -import org.apache.flink.table.catalog.ResolvedSchema; -import org.apache.flink.table.catalog.SchemaResolver; -import org.apache.flink.table.catalog.StartMode; -import org.apache.flink.table.catalog.TableChange; -import org.apache.flink.table.catalog.TableDistribution; -import org.apache.flink.table.catalog.UniqueConstraint; -import org.apache.flink.table.catalog.WatermarkSpec; +import org.apache.flink.table.catalog.UnresolvedIdentifier; import org.apache.flink.table.operations.Operation; -import org.apache.flink.table.operations.materializedtable.ConvertTableToMaterializedTableOperation; -import org.apache.flink.table.operations.materializedtable.CreateMaterializedTableOperation; -import org.apache.flink.table.operations.materializedtable.FullAlterMaterializedTableOperation; -import org.apache.flink.table.operations.materializedtable.MaterializedTableChangeHandler; -import org.apache.flink.table.planner.operations.PlannerQueryOperation; -import org.apache.flink.table.planner.operations.converters.MergeTableAsUtil; -import org.apache.flink.table.planner.utils.MaterializedTableUtils; - -import org.apache.calcite.sql.SqlNodeList; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Objects; -import java.util.Optional; +import org.apache.flink.table.operations.materializedtable.MaterializedTableOperationFactory; +import org.apache.flink.table.operations.materializedtable.MaterializedTableOperationFactory.Mode; +import org.apache.flink.table.planner.operations.converters.SqlNodeConverter; /** A converter for {@link SqlCreateOrAlterMaterializedTable}. */ public class SqlCreateOrAlterMaterializedTableConverter - extends AbstractCreateMaterializedTableConverter<SqlCreateOrAlterMaterializedTable> { + implements SqlNodeConverter<SqlCreateOrAlterMaterializedTable> { @Override public Operation convertSqlNode( SqlCreateOrAlterMaterializedTable sqlCreateOrAlterMaterializedTable, ConvertContext context) { final ObjectIdentifier identifier = - getIdentifier(sqlCreateOrAlterMaterializedTable, context); - - if (createOrAlterOperation(sqlCreateOrAlterMaterializedTable)) { - return handleCreateOrAlter(sqlCreateOrAlterMaterializedTable, context, identifier); - } - return handleCreate(sqlCreateOrAlterMaterializedTable, context, identifier); - } - - private Operation handleCreateOrAlter( - final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterMaterializedTable, - final ConvertContext context, - final ObjectIdentifier identifier) { - final Optional<ResolvedCatalogBaseTable<?>> resolvedBaseTable = - context.getCatalogManager().getCatalogBaseTable(identifier); - if (resolvedBaseTable.isEmpty()) { - return handleCreate(sqlCreateOrAlterMaterializedTable, context, identifier); - } - final ResolvedCatalogBaseTable<?> oldBaseTable = resolvedBaseTable.get(); - final TableKind oldBaseTableKind = oldBaseTable.getTableKind(); - switch (oldBaseTableKind) { - case MATERIALIZED_TABLE: - return handleAlter( - sqlCreateOrAlterMaterializedTable, - (ResolvedCatalogMaterializedTable) oldBaseTable, - context, - identifier); - case TABLE: - return handleConvert( - sqlCreateOrAlterMaterializedTable, - context, + context.getCatalogManager() + .qualifyIdentifier( + UnresolvedIdentifier.of( + sqlCreateOrAlterMaterializedTable.getFullName())); + final Mode mode = + sqlCreateOrAlterMaterializedTable.getOperator() + == SqlCreateOrAlterMaterializedTable.CREATE_OR_ALTER_OPERATOR + ? Mode.CREATE_OR_ALTER + : Mode.CREATE; + return new MaterializedTableOperationFactory( + context.getCatalogManager(), context.getTableConfig()) + .build( identifier, - (ResolvedCatalogTable) oldBaseTable); - default: - throw new ValidationException( - String.format( - "Catalog object %s of kind %s does not support the CREATE OR ALTER MATERIALIZED TABLE operation.", - identifier.asSummaryString(), oldBaseTableKind)); - } - } - - private static boolean createOrAlterOperation( - final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterMaterializedTable) { - return sqlCreateOrAlterMaterializedTable.getOperator() - == SqlCreateOrAlterMaterializedTable.CREATE_OR_ALTER_OPERATOR; - } - - private Operation handleAlter( - final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterTable, - final ResolvedCatalogMaterializedTable oldTable, - final ConvertContext context, - final ObjectIdentifier identifier) { - final SchemaResolver schemaResolver = context.getCatalogManager().getSchemaResolver(); - final MergeContext mergeContext = getMergeContext(sqlCreateOrAlterTable, context); - final PlannerQueryOperation asQuery = mergeContext.getAsQueryOperation(); - - return new FullAlterMaterializedTableOperation( - identifier, - currentTable -> buildTableChanges(currentTable, mergeContext, schemaResolver), - oldTable, - currentTable -> buildNewTable(currentTable, mergeContext, schemaResolver), - asQuery); - } - - private Operation handleConvert( - final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterMaterializedTable, - final ConvertContext context, - final ObjectIdentifier identifier, - final ResolvedCatalogTable oldBaseTable) { - final boolean conversionEnabled = - context.getTableConfig() - .getRootConfiguration() - .get(TableConfigOptions.MATERIALIZED_TABLE_CONVERSION_FROM_TABLE_ENABLED); - if (!conversionEnabled) { - throw new ValidationException( - "Regular table does not support create or alter operation."); - } - final MergeContext baseMergeContext = - getMergeContext(sqlCreateOrAlterMaterializedTable, context); - - final CatalogMaterializedTable newMaterializedTable = - CatalogMaterializedTable.newBuilder() - .schema(baseMergeContext.getMergedSchema()) - .comment(baseMergeContext.getMergedComment()) - .partitionKeys(baseMergeContext.getMergedPartitionKeys()) - .options(baseMergeContext.getMergedTableOptions()) - .originalQuery(baseMergeContext.getMergedOriginalQuery()) - .expandedQuery(baseMergeContext.getMergedExpandedQuery()) - .distribution(baseMergeContext.getMergedTableDistribution().orElse(null)) - .freshness(baseMergeContext.getMergedFreshness()) - .logicalRefreshMode(baseMergeContext.getMergedLogicalRefreshMode()) - .refreshMode(baseMergeContext.getMergedRefreshMode()) - .refreshStatus(RefreshStatus.INITIALIZING) - .startMode(baseMergeContext.getMergedStartMode()) - .build(); - final ResolvedCatalogMaterializedTable resolvedNewMaterializedTable = - context.getCatalogManager().resolveCatalogMaterializedTable(newMaterializedTable); - - final PlannerQueryOperation asQueryOperation = - new MergeTableAsUtil(context) - .maybeRewriteQuery( - baseMergeContext.getAsQueryOperation(), - sqlCreateOrAlterMaterializedTable.getAsQuery(), - resolvedNewMaterializedTable); - - return new ConvertTableToMaterializedTableOperation( - identifier, - oldBaseTable, - resolvedNewMaterializedTable, - resolvedCatalogMaterializedTable -> - buildConversionTableChanges( - oldBaseTable, - resolvedCatalogMaterializedTable, - baseMergeContext.hasSchemaDefinition(), - baseMergeContext.hasConstraintDefinition()), - asQueryOperation); - } - - private List<TableChange> buildConversionTableChanges( - final ResolvedCatalogTable oldTable, - final ResolvedCatalogMaterializedTable newTable, - final boolean hasSchemaDefinition, - final boolean hasConstraintDefinition) { - final ResolvedSchema oldSchema = oldTable.getResolvedSchema(); - final ResolvedSchema newSchema = newTable.getResolvedSchema(); - final List<TableChange> changes = - new ArrayList<>( - MaterializedTableUtils.validateAndExtractColumnChanges( - oldSchema, newSchema, hasSchemaDefinition)); - - getConstraintChange(oldSchema, newSchema, hasConstraintDefinition).ifPresent(changes::add); - getWatermarkChange(oldSchema, newSchema, hasSchemaDefinition).ifPresent(changes::add); - - changes.addAll( - getQueryTableChanges( - null, null, newTable.getOriginalQuery(), newTable.getExpandedQuery())); - changes.addAll(getOptionsTableChanges(oldTable.getOptions(), newTable.getOptions())); - changes.addAll( - getDistributionTableChanges( - oldTable.getDistribution().orElse(null), - newTable.getDistribution().orElse(null))); - - newTable.getStartMode() - .ifPresent(newStartMode -> changes.add(TableChange.modifyStartMode(newStartMode))); - - return changes; - } - - private CatalogMaterializedTable buildNewTable( - final ResolvedCatalogMaterializedTable currentTable, - final MergeContext mergeContext, - final SchemaResolver schemaResolver) { - return CatalogMaterializedTable.newBuilder() - .schema( - MaterializedTableChangeHandler.getHandlerWithChanges( - currentTable, - getSchemaTableChanges( - mergeContext, schemaResolver, currentTable)) - .retrieveSchema()) - .comment(mergeContext.getMergedComment()) - .partitionKeys(mergeContext.getMergedPartitionKeys()) - .options(mergeContext.getMergedTableOptions()) - .originalQuery(mergeContext.getMergedOriginalQuery()) - .expandedQuery(mergeContext.getMergedExpandedQuery()) - .distribution(mergeContext.getMergedTableDistribution().orElse(null)) - .freshness(mergeContext.getMergedFreshness()) - .logicalRefreshMode(mergeContext.getMergedLogicalRefreshMode()) - .refreshMode(mergeContext.getMergedRefreshMode()) - .refreshStatus(currentTable.getRefreshStatus()) - .refreshHandlerDescription(currentTable.getRefreshHandlerDescription().orElse(null)) - .serializedRefreshHandler(currentTable.getSerializedRefreshHandler()) - .startMode(mergeContext.getMergedStartMode()) - .build(); - } - - private Operation handleCreate( - final SqlCreateOrAlterMaterializedTable sqlCreateOrAlterTable, - final ConvertContext context, - final ObjectIdentifier identifier) { - final MergeContext mergeContext = getMergeContext(sqlCreateOrAlterTable, context); - final ResolvedCatalogMaterializedTable resolvedTable = - getResolvedCatalogMaterializedTable(mergeContext, sqlCreateOrAlterTable, context); - final PlannerQueryOperation asQueryOperation = - new MergeTableAsUtil(context) - .maybeRewriteQuery( - mergeContext.getAsQueryOperation(), - sqlCreateOrAlterTable.getAsQuery(), - resolvedTable); - - return new CreateMaterializedTableOperation(identifier, resolvedTable, asQueryOperation); - } - - private List<TableChange> buildTableChanges( - final ResolvedCatalogMaterializedTable oldTable, - final MergeContext mergeContext, - final SchemaResolver schemaResolver) { - final List<TableChange> changes = - getSchemaTableChanges(mergeContext, schemaResolver, oldTable); - - changes.addAll( - getQueryTableChanges( - oldTable.getOriginalQuery(), - oldTable.getExpandedQuery(), - mergeContext.getMergedOriginalQuery(), - mergeContext.getMergedExpandedQuery())); - changes.addAll( - getOptionsTableChanges( - oldTable.getOptions(), mergeContext.getMergedTableOptions())); - changes.addAll( - getDistributionTableChanges( - oldTable.getDistribution().orElse(null), - mergeContext.getMergedTableDistribution().orElse(null))); - - final RefreshMode oldRefreshMode = oldTable.getRefreshMode(); - final RefreshMode newRefreshMode = mergeContext.getMergedRefreshMode(); - if (oldRefreshMode != newRefreshMode && newRefreshMode != null) { - throw new ValidationException("Changing of REFRESH MODE is unsupported"); - } - - final StartMode newStartMode = mergeContext.getMergedStartMode(); - if (newStartMode != null) { - final StartMode oldStartMode = - oldTable.getStartMode() - .orElseThrow( - () -> - new ValidationException( - "Start mode must be set on materialized table.")); - if (!Objects.equals(oldStartMode, newStartMode)) { - changes.add(TableChange.modifyStartMode(newStartMode)); - } - } - - return changes; - } - - private List<TableChange> getDistributionTableChanges( - final TableDistribution oldDistribution, final TableDistribution newDistribution) { - if (!Objects.equals(oldDistribution, newDistribution)) { - if (oldDistribution == null) { - return List.of(TableChange.add(newDistribution)); - } else if (newDistribution == null) { - return List.of(TableChange.dropDistribution()); - } else { - return List.of(TableChange.modify(newDistribution)); - } - } - return List.of(); - } - - private List<TableChange> getOptionsTableChanges( - final Map<String, String> oldOptions, final Map<String, String> newOptions) { - final List<TableChange> changes = new ArrayList<>(); - - for (Map.Entry<String, String> newOptionEntry : newOptions.entrySet()) { - if (!newOptionEntry.getValue().equals(oldOptions.get(newOptionEntry.getKey()))) { - changes.add(TableChange.set(newOptionEntry.getKey(), newOptionEntry.getValue())); - } - } - - for (Map.Entry<String, String> oldOptionEntry : oldOptions.entrySet()) { - if (newOptions.get(oldOptionEntry.getKey()) == null) { - changes.add(TableChange.reset(oldOptionEntry.getKey())); - } - } - return changes; - } - - private List<TableChange> getQueryTableChanges( - final String oldOriginalQuery, - final String oldExpandedQuery, - final String newOriginalQuery, - final String newExpandedQuery) { - if (!Objects.equals(oldOriginalQuery, newOriginalQuery) - || !Objects.equals(oldExpandedQuery, newExpandedQuery)) { - return List.of(TableChange.modifyDefinitionQuery(newOriginalQuery, newExpandedQuery)); - } - return List.of(); - } - - private List<TableChange> getSchemaTableChanges( - final MergeContext mergeContext, - final SchemaResolver schemaResolver, - final ResolvedCatalogMaterializedTable oldTable) { - final ResolvedSchema oldSchema = oldTable.getResolvedSchema(); - final ResolvedSchema newSchema = schemaResolver.resolve(mergeContext.getMergedSchema()); - final boolean hasSchemaDefinition = mergeContext.hasSchemaDefinition(); - final List<TableChange> changes = - new ArrayList<>( - MaterializedTableUtils.validateAndExtractColumnChanges( - oldSchema, newSchema, hasSchemaDefinition)); - - getConstraintChange(oldSchema, newSchema, mergeContext.hasConstraintDefinition()) - .ifPresent(changes::add); - getWatermarkChange(oldSchema, newSchema, hasSchemaDefinition).ifPresent(changes::add); - return changes; - } - - private Optional<TableChange> getConstraintChange( - final ResolvedSchema oldSchema, - final ResolvedSchema newSchema, - final boolean hasConstraintDefinition) { - final UniqueConstraint oldConstraint = oldSchema.getPrimaryKey().orElse(null); - final UniqueConstraint newConstraint = newSchema.getPrimaryKey().orElse(null); - if (hasConstraintDefinition && !Objects.equals(oldConstraint, newConstraint)) { - if (newConstraint == null) { - return Optional.of(TableChange.dropConstraint(oldConstraint.getName())); - } else if (oldConstraint == null) { - return Optional.of(TableChange.add(newConstraint)); - } else { - return Optional.of(TableChange.modify(newConstraint)); - } - } - return Optional.empty(); - } - - private Optional<TableChange> getWatermarkChange( - final ResolvedSchema oldSchema, - final ResolvedSchema newSchema, - boolean hasSchemaDefinition) { - final WatermarkSpec oldWatermarkSpec = - oldSchema.getWatermarkSpecs().isEmpty() - ? null - : oldSchema.getWatermarkSpecs().get(0); - final WatermarkSpec newWatermarkSpec = - newSchema.getWatermarkSpecs().isEmpty() - ? null - : newSchema.getWatermarkSpecs().get(0); - if (hasSchemaDefinition && !Objects.equals(oldWatermarkSpec, newWatermarkSpec)) { - if (newWatermarkSpec == null) { - return Optional.of(TableChange.dropWatermark()); - } else if (oldWatermarkSpec == null) { - return Optional.of(TableChange.add(newWatermarkSpec)); - } else { - return Optional.of(TableChange.modify(newWatermarkSpec)); - } - } - return Optional.empty(); - } - - @Override - protected MergeContext getMergeContext( - final SqlCreateOrAlterMaterializedTable sqlCreateMaterializedTable, - final ConvertContext context) { - return new MergeContext() { - private final MergeTableAsUtil mergeTableAsUtil = new MergeTableAsUtil(context); - - // Cache original query. If we call getDerivedExpandedQuery() first, without storing the - // original query, it the SqlNode will be changed and the getAsQuery() will always - // return the expanded query. - private final String originalQuery = - SqlCreateOrAlterMaterializedTableConverter.this.getDerivedOriginalQuery( - sqlCreateMaterializedTable, context); - - private final PlannerQueryOperation asQueryOperation = - SqlCreateOrAlterMaterializedTableConverter.this.getAsQueryOperation( - sqlCreateMaterializedTable, context); - - private final ResolvedSchema querySchema = asQueryOperation.getResolvedSchema(); - - @Override - public boolean hasSchemaDefinition() { - final SqlNodeList sqlNodeList = sqlCreateMaterializedTable.getColumnList(); - return !sqlNodeList.getList().isEmpty() - && sqlNodeList.getList().get(0) instanceof SqlRegularColumn; - } - - @Override - public boolean hasConstraintDefinition() { - return !sqlCreateMaterializedTable.getTableConstraints().isEmpty() - || hasSchemaDefinition(); - } - - @Override - public Schema getMergedSchema() { - final SqlNodeList sqlNodeList = sqlCreateMaterializedTable.getColumnList(); - MaterializedTableUtils.validatePersistedColumnsUsedByQuery( - sqlNodeList, - querySchema, - sqlCreateMaterializedTable.getOperator().getName()); - if (sqlCreateMaterializedTable.isSchemaWithColumnsIdentifiersOnly()) { - // If only column identifiers are provided, then these are used to - // order the columns in the schema. - return mergeTableAsUtil.reorderSchema(sqlNodeList, querySchema); - } else { - return mergeTableAsUtil.mergeSchemas( - sqlNodeList, - sqlCreateMaterializedTable.getWatermark().orElse(null), - sqlCreateMaterializedTable.getFullConstraints(), - querySchema); - } - } - - @Override - public Map<String, String> getMergedTableOptions() { - return sqlCreateMaterializedTable.getProperties(); - } - - @Override - public List<String> getMergedPartitionKeys() { - return sqlCreateMaterializedTable.getPartitionKeyList(); - } - - @Override - public Optional<TableDistribution> getMergedTableDistribution() { - return SqlCreateOrAlterMaterializedTableConverter.this.getDerivedTableDistribution( - sqlCreateMaterializedTable); - } - - @Override - public String getMergedOriginalQuery() { - return this.originalQuery; - } - - @Override - public String getMergedExpandedQuery() { - return SqlCreateOrAlterMaterializedTableConverter.this.getDerivedExpandedQuery( - sqlCreateMaterializedTable, context); - } - - @Override - public ResolvedSchema getMergedQuerySchema() { - return asQueryOperation.getResolvedSchema(); - } - - @Override - public PlannerQueryOperation getAsQueryOperation() { - return asQueryOperation; - } - - @Override - public RefreshMode getMergedRefreshMode() { - return getDerivedRefreshMode(getMergedLogicalRefreshMode()); - } - - @Override - public LogicalRefreshMode getMergedLogicalRefreshMode() { - return getDerivedLogicalRefreshMode(sqlCreateMaterializedTable); - } - - @Override - public StartMode getMergedStartMode() { - return getStartMode(sqlCreateMaterializedTable, context); - } - - @Override - public String getMergedComment() { - return getComment(sqlCreateMaterializedTable); - } - - @Override - public IntervalFreshness getMergedFreshness() { - return getDerivedFreshness(sqlCreateMaterializedTable); - } - }; + mode, + new SqlMaterializedTableMergeContext( + sqlCreateOrAlterMaterializedTable, context)); Review Comment: Small behavior change for SQL: the merge context validates the query in its constructor, before the factory looks up what's at the path. With an invalid query and a view at the path (or conversion disabled), the user now gets the query error instead of "Catalog object … does not support …". Probably fine, but the PR says SQL behavior is unchanged. Either construct the context lazily or mention it. -- 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]
