lvyanquan commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4080146385
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java:
##########
@@ -114,6 +119,57 @@ public Set<SchemaChangeEventType>
getSupportedSchemaEvolutionTypes() {
SchemaChangeEventType.ALTER_COLUMN_TYPE);
}
+ @Override
+ public Optional<ExistingTableSchemaExpansionSupport>
getExistingTableSchemaExpansionSupport() {
+ return Optional.of(this);
+ }
+
+ @Override
+ public Optional<Schema> getExistingTableSchema(TableId tableId) {
Review Comment:
Consolidated in fe288eec0: `getExistingTableSchema`,
`isColumnNameCaseSensitive` and `applySchemaChange` all go through a single
private `getCatalog()`, and `close()` now also nulls the cached instance so the
applier stays reusable after being closed. Added
`testCloseIsIdempotentAndRecreatesCatalogOnReuse`, which triggers lazy creation
through `applySchemaChange`, closes twice, applies again and asserts the
recreated catalog sees the table.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,767 @@
+/*
+ * 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.cdc.runtime.operators.schema.common;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
+import org.apache.flink.cdc.common.pipeline.ExistingTableSchemaExpansionMode;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
+import org.apache.flink.cdc.common.schema.Column;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.ExistingTableSchemaExpansionSupport;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.types.BinaryType;
+import org.apache.flink.cdc.common.types.CharType;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypeFamily;
+import org.apache.flink.cdc.common.types.DataTypeRoot;
+import org.apache.flink.cdc.common.types.DecimalType;
+import org.apache.flink.cdc.common.types.LocalZonedTimestampType;
+import org.apache.flink.cdc.common.types.TimeType;
+import org.apache.flink.cdc.common.types.TimestampType;
+import org.apache.flink.cdc.common.types.VarBinaryType;
+import org.apache.flink.cdc.common.types.VarCharType;
+import org.apache.flink.cdc.common.types.ZonedTimestampType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Handles the initial {@link CreateTableEvent} for an existing target table
according to the
+ * configured {@link ExistingTableSchemaExpansionMode}.
+ */
+@Internal
+public class ExistingTableSchemaExpander {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ExistingTableSchemaExpander.class);
+
+ private final MetadataApplier metadataApplier;
+ private final ExistingTableSchemaExpansionSupport expansionSupport;
+ private final SchemaChangeBehavior schemaChangeBehavior;
+ private final ExistingTableSchemaExpansionMode mode;
+
+ public ExistingTableSchemaExpander(
+ MetadataApplier metadataApplier,
+ ExistingTableSchemaExpansionSupport expansionSupport,
+ SchemaChangeBehavior schemaChangeBehavior) {
+ this(
+ metadataApplier,
+ expansionSupport,
+ schemaChangeBehavior,
+ ExistingTableSchemaExpansionMode.TRY_EXPAND);
+ }
+
+ public ExistingTableSchemaExpander(
+ MetadataApplier metadataApplier,
+ ExistingTableSchemaExpansionSupport expansionSupport,
+ SchemaChangeBehavior schemaChangeBehavior,
+ ExistingTableSchemaExpansionMode mode) {
+ this.metadataApplier = metadataApplier;
+ this.expansionSupport = expansionSupport;
+ this.schemaChangeBehavior = schemaChangeBehavior;
+ this.mode = mode;
+ }
+
+ /**
+ * Handles the initial {@link CreateTableEvent} for an existing target
table.
+ *
+ * @return whether the caller should proceed to apply the original {@link
CreateTableEvent} to
+ * the sink. {@code CHECK} mode returns {@code false} after a
successful check so that no
+ * external DDL is issued by the pipeline.
+ */
+ public boolean expand(CreateTableEvent createTableEvent) {
+ if (mode == ExistingTableSchemaExpansionMode.CHECK) {
+ // CHECK guards the initial table state and runs regardless of
schema.change.behavior.
+ checkCompatibility(createTableEvent);
+ return false;
+ }
+ if (schemaChangeBehavior == SchemaChangeBehavior.IGNORE
+ || schemaChangeBehavior == SchemaChangeBehavior.EXCEPTION) {
+ // Keep the original rule: TRY_EXPAND/EXPAND skip framework-side
handling here.
+ return true;
+ }
+ switch (mode) {
+ case TRY_EXPAND:
+ tryExpand(createTableEvent);
+ return true;
+ case EXPAND:
+ expandStrictly(createTableEvent);
+ return true;
+ case OFF:
+ default:
+ return true;
+ }
+ }
+
+ private void checkCompatibility(CreateTableEvent createTableEvent) {
+ Optional<Schema> targetSchema =
queryTargetSchema(createTableEvent.tableId());
+ if (!targetSchema.isPresent()) {
+ throw new SchemaEvolveException(
+ createTableEvent,
+ String.format(
+ "Existing target table %s does not exist. CHECK
mode never creates tables; create the target table externally first.",
+ createTableEvent.tableId()));
+ }
+ ExpansionPlan plan = analyze(createTableEvent, targetSchema.get());
+ // CHECK never issues DDL, so every difference - including missing
columns and narrow
+ // column types that EXPAND could repair - makes the target table
unable to contain the
+ // upstream schema.
+ if (!plan.incompatibilities.isEmpty()
+ || !plan.columnsToAdd.isEmpty()
+ || !plan.columnsToWiden.isEmpty()) {
+ throw incompatibleException(createTableEvent, plan);
+ }
+ LOG.info(
+ "Existing target table {} passed the schema compatibility
check.",
+ createTableEvent.tableId());
+ }
+
+ private void tryExpand(CreateTableEvent createTableEvent) {
+ try {
+ if (!supportsAnyExpansionDdl()) {
+ LOG.info(
+ "Neither ADD_COLUMN nor ALTER_COLUMN_TYPE is enabled
or supported for target table {}. Delegating schema handling to the sink.",
+ createTableEvent.tableId());
+ return;
+ }
+ Optional<Schema> targetSchema =
queryTargetSchema(createTableEvent.tableId());
+ if (!targetSchema.isPresent()) {
+ LOG.info(
+ "Target table {} does not exist. Delegating table
creation to the sink.",
+ createTableEvent.tableId());
+ return;
+ }
+ ExpansionPlan plan = analyze(createTableEvent, targetSchema.get());
+ for (String incompatibility : plan.incompatibilities) {
+ LOG.warn(
+ "Target table {} has an unsupported difference: {}.
Delegating it to the sink.",
+ createTableEvent.tableId(),
+ incompatibility);
+ }
+ applyPlan(createTableEvent, plan);
+ verifyExpansion(createTableEvent, plan, false);
+ } catch (Exception e) {
+ LOG.warn(
+ "Best-effort schema expansion failed for existing target
table {}. Delegating schema handling to the sink.",
+ createTableEvent.tableId(),
+ e);
+ }
+ }
+
+ private void expandStrictly(CreateTableEvent createTableEvent) {
+ Optional<Schema> targetSchema =
queryTargetSchema(createTableEvent.tableId());
+ if (!targetSchema.isPresent()) {
+ LOG.info(
+ "Target table {} does not exist. Delegating table creation
to the sink.",
+ createTableEvent.tableId());
+ return;
+ }
+ ExpansionPlan plan = analyze(createTableEvent, targetSchema.get());
+ if (!plan.incompatibilities.isEmpty()) {
+ throw incompatibleException(createTableEvent, plan);
+ }
+ // A fully compatible target table needs no DDL, so a missing DDL
capability is only an
+ // error when differences actually require repair; applyPlan enforces
that per event type.
+ applyPlan(createTableEvent, plan);
+ verifyExpansion(createTableEvent, plan, true);
+ }
+
+ private boolean supportsAnyExpansionDdl() {
+ return supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)
+ ||
supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE);
+ }
+
+ private SchemaEvolveException incompatibleException(
+ CreateTableEvent createTableEvent, ExpansionPlan plan) {
+ StringBuilder differences = new StringBuilder();
+ for (String incompatibility : plan.incompatibilities) {
+ differences.append("\n - ").append(incompatibility);
+ }
+ for (Column columnToAdd : plan.columnsToAdd) {
+ differences
+ .append("\n - target table is missing column \"")
+ .append(columnToAdd.getName())
+ .append("\"");
+ }
+ for (Map.Entry<String, DataType> columnToWiden :
plan.columnsToWiden.entrySet()) {
+ differences
+ .append("\n - target column \"")
+ .append(columnToWiden.getKey())
+ .append("\" is narrower than pipeline type ")
+ .append(columnToWiden.getValue());
+ }
+ String message =
+ String.format(
+ "Existing target table %s cannot contain the pipeline
schema:%s",
+ createTableEvent.tableId(), differences);
+ String repairSuggestions = renderRepairSuggestions(createTableEvent,
plan);
+ if (!repairSuggestions.isEmpty()) {
+ message +=
+ String.format(
+ "\nSuggested repair statements (adjust to the
target system's DDL dialect):%s",
+ repairSuggestions);
+ }
+ return new SchemaEvolveException(createTableEvent, message);
+ }
+
+ /**
+ * Renders lightweight, review-oriented ALTER TABLE suggestions for the
safely repairable
+ * differences. Differences that cannot be fixed safely never get a
suggested statement.
+ */
+ private String renderRepairSuggestions(CreateTableEvent createTableEvent,
ExpansionPlan plan) {
Review Comment:
Added (fe288eec0): the suggestion block is now introduced by `Suggested
repair SQL templates (review and adjust to the target connector's DDL dialect
before execution):` inside the aggregated `SchemaEvolveException` message, so
it reaches operators in the failure text rather than only in the docs; the docs
echo the same caveat. Differences that cannot be repaired safely still get no
suggested statement.
--
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]
