leonardBang commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4091165266
########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,808 @@ +/* + * 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. + * + * <p>This is a two-phase contract: the expander first tries to check/expand the existing target + * table, then signals whether the sink should still apply the original {@link + * CreateTableEvent}. + * + * @return {@code true} if the caller should proceed to apply the original {@link + * CreateTableEvent} to the sink; {@code false} means the framework has fully handled this + * event and the original {@link CreateTableEvent} must not be applied. {@code CHECK} mode + * returns {@code false} after a successful check so that no external DDL is issued by the + * pipeline. + */ + public boolean handleExistingTableCreation(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 DISABLED: + 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) { + // Phase 1 - probe the target table. Failures here are "cannot expand safely" cases: the + // connector lacks DDL capability, or the target table does not exist yet. Delegating to the + // sink's original behavior is safe because no derived DDL has been issued. + 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; + try { + targetSchema = queryTargetSchema(createTableEvent.tableId()); + } catch (Exception e) { + LOG.warn( + "Failed to read the existing target table {} before expansion. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + return; + } + 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); + } + + // The plan needs repair DDL, but the sink cannot perform the specific required event type. + // This is a capability gap, not a transient failure, so delegate to the sink instead of + // failing the job (mirrors applyPlan's capability checks, which throw for EXPAND mode). + if (!plan.columnsToAdd.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + LOG.warn( + "Target table {} is missing columns {}, but ADD_COLUMN is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + getColumnNames(plan.columnsToAdd)); + return; + } + if (!plan.columnsToWiden.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + LOG.warn( + "Target table {} has narrow columns {}, but ALTER_COLUMN_TYPE is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + plan.columnsToWiden.keySet()); + return; + } + + // Phase 2 - apply and verify. At this point the expander has identified supportable + // differences and is about to issue derived DDL. A failure here (transient network/catalog + // error, DDL execution failure, or read-back verification failure) must propagate so the + // job fails over and the idempotent expander retries; otherwise the original + // CreateTableEvent + // would be applied to an existing table (typically a no-op) and the missing columns would + // be + // silently dropped forever. + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); + } + + 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 SQL templates (review and adjust to the target connector's DDL dialect before execution):%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) { + StringBuilder suggestions = new StringBuilder(); + for (Column columnToAdd : plan.columnsToAdd) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ADD COLUMN ") + .append(columnToAdd.getName()) + .append(" ") + .append(columnToAdd.getType()) + .append(";"); + } + for (Map.Entry<String, DataType> columnToWiden : plan.columnsToWiden.entrySet()) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ALTER COLUMN ") + .append(columnToWiden.getKey()) + .append(" TYPE ") + .append(columnToWiden.getValue()) + .append(";"); + } + return suggestions.toString(); + } + + /** + * Analyzes the pipeline schema against the current target schema and derives the safe DDL plan. + * Differences that cannot be fixed by safe DDL are collected in {@link + * ExpansionPlan#incompatibilities}. + */ + private ExpansionPlan analyze(CreateTableEvent createTableEvent, Schema currentTargetSchema) { + ExpansionPlan plan = new ExpansionPlan(); + Schema pipelineSchema = createTableEvent.getSchema(); + boolean columnNameCaseSensitive = expansionSupport.isColumnNameCaseSensitive(); + ColumnIndex targetColumns = indexColumns(currentTargetSchema, columnNameCaseSensitive); + plan.targetColumns = targetColumns; + Set<String> ambiguousColumnNames = + new HashSet<>( + indexColumns(pipelineSchema, columnNameCaseSensitive) + .getAmbiguousColumnNames()); + ambiguousColumnNames.addAll(targetColumns.getAmbiguousColumnNames()); + Set<String> keyColumns = + getKeyColumns(pipelineSchema, currentTargetSchema, columnNameCaseSensitive); + + for (Column pipelineColumn : pipelineSchema.getColumns()) { + String columnName = pipelineColumn.getName(); + String comparisonName = normalizeColumnName(columnName, columnNameCaseSensitive); + if (ambiguousColumnNames.contains(comparisonName)) { + plan.incompatibilities.add( + String.format( + "column \"%s\" is ambiguous under the target system's case-sensitivity rule", + columnName)); + continue; + } + + Column targetColumn = targetColumns.get(columnName); + if (targetColumn == null) { + if (pipelineColumn.isPhysical() && !keyColumns.contains(comparisonName)) { + plan.columnsToAdd.add(pipelineColumn.copy(pipelineColumn.getType().nullable())); + } else { + plan.incompatibilities.add( + String.format( + "target table is missing the non-addable column \"%s\"", + columnName)); + } + continue; + } + + Optional<DataType> normalizedPipelineTypeOptional = + normalizeType( + createTableEvent.tableId(), + columnName, + pipelineColumn.getType(), + currentTargetSchema); + if (!normalizedPipelineTypeOptional.isPresent()) { + plan.incompatibilities.add( + String.format( + "pipeline type %s of column \"%s\" cannot be normalized to the target type system", + pipelineColumn.getType(), columnName)); + continue; + } + DataType normalizedPipelineType = normalizedPipelineTypeOptional.get().nullable(); + DataType targetType = targetColumn.getType().nullable(); + + if (pipelineColumn.getType().isNullable() && !targetColumn.getType().isNullable()) { + plan.incompatibilities.add( + String.format( + "column \"%s\" is nullable in the pipeline but NOT NULL in the target table", + columnName)); + continue; + } + + if (canContain(targetType, normalizedPipelineType)) { + continue; + } + if (keyColumns.contains(comparisonName)) { + plan.incompatibilities.add( + String.format( + "key column \"%s\" with target type %s cannot contain pipeline type %s", + columnName, targetType, normalizedPipelineType)); + continue; + } + + Optional<DataType> widenedType = getSafeWidenedType(targetType, normalizedPipelineType); + if (!widenedType.isPresent()) { + plan.incompatibilities.add( + String.format( + "column \"%s\" with target type %s cannot safely contain pipeline type %s", + columnName, targetType, normalizedPipelineType)); + continue; + } + + Optional<DataType> normalizedWidenedTypeOptional = + normalizeType( + createTableEvent.tableId(), + columnName, + widenedType.get(), + currentTargetSchema); + if (!normalizedWidenedTypeOptional.isPresent()) { + plan.incompatibilities.add( + String.format( + "proposed widened type %s for column \"%s\" cannot be normalized to the target type system", + widenedType.get(), columnName)); + continue; + } + DataType normalizedWidenedType = normalizedWidenedTypeOptional.get().nullable(); + if (!canContain(normalizedWidenedType, targetType) + || !canContain(normalizedWidenedType, normalizedPipelineType)) { + plan.incompatibilities.add( + String.format( + "target system normalizes proposed widened type %s for column \"%s\" to %s, which is not a safe widening", + widenedType.get(), columnName, normalizedWidenedType)); + continue; + } + + plan.columnsToWiden.put( + targetColumn.getName(), + widenedType.get().copy(targetColumn.getType().isNullable())); + } + return plan; + } + + private void applyPlan(CreateTableEvent createTableEvent, ExpansionPlan plan) { + if (!plan.columnsToAdd.isEmpty()) { + if (!supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + throw new SchemaEvolveException( + createTableEvent, + String.format( + "Target table %s is missing columns %s, but ADD_COLUMN is not enabled or supported by the sink.", + createTableEvent.tableId(), getColumnNames(plan.columnsToAdd))); + } + AddColumnEvent addColumnEvent = + new AddColumnEvent( + createTableEvent.tableId(), + plan.columnsToAdd.stream() + .map(AddColumnEvent.ColumnWithPosition::new) + .collect(Collectors.toList())); + applySchemaChange(addColumnEvent, plan.columnsToAdd); + } + if (!plan.columnsToWiden.isEmpty()) { + if (!supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + throw new SchemaEvolveException( + createTableEvent, + String.format( + "Target table %s has narrow columns %s, but ALTER_COLUMN_TYPE is not enabled or supported by the sink.", + createTableEvent.tableId(), plan.columnsToWiden.keySet())); + } + AlterColumnTypeEvent alterColumnTypeEvent = + new AlterColumnTypeEvent( + createTableEvent.tableId(), + plan.columnsToWiden, + plan.columnsToWiden.keySet().stream() + .collect( + Collectors.toMap( + columnName -> columnName, + columnName -> + plan.targetColumns + .get(columnName) + .getType()))); + applySchemaChange(alterColumnTypeEvent, plan.columnsToWiden); + } + } + + /** Re-reads the target schema after applying derived DDL to detect no-op or failed DDL. */ + private void verifyExpansion( + CreateTableEvent createTableEvent, ExpansionPlan plan, boolean strict) { + if (plan.columnsToAdd.isEmpty() && plan.columnsToWiden.isEmpty()) { + return; + } + Optional<Schema> updatedTargetSchema = queryTargetSchema(createTableEvent.tableId()); + if (!updatedTargetSchema.isPresent()) { + throw new SchemaEvolveException( + createTableEvent, + String.format( + "Failed to read back target table %s after expansion.", + createTableEvent.tableId())); + } + ExpansionPlan remaining = analyze(createTableEvent, updatedTargetSchema.get()); + if (!remaining.incompatibilities.isEmpty() + || !remaining.columnsToAdd.isEmpty() + || !remaining.columnsToWiden.isEmpty()) { + String message = + String.format( + "Target table %s still has unresolved differences after expansion: %s", + createTableEvent.tableId(), remaining.incompatibilities); + if (strict) { + throw new SchemaEvolveException(createTableEvent, message); Review Comment: Confirmed resolved - `describeDifferences(plan)` renders all three kinds now, and the `doesNotContain("after expansion: []")` assertion in `testUnresolvedDifferencesAfterExpansionAreListed` pins the behaviour. Thanks. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,808 @@ +/* + * 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. + * + * <p>This is a two-phase contract: the expander first tries to check/expand the existing target + * table, then signals whether the sink should still apply the original {@link + * CreateTableEvent}. + * + * @return {@code true} if the caller should proceed to apply the original {@link + * CreateTableEvent} to the sink; {@code false} means the framework has fully handled this + * event and the original {@link CreateTableEvent} must not be applied. {@code CHECK} mode + * returns {@code false} after a successful check so that no external DDL is issued by the + * pipeline. + */ + public boolean handleExistingTableCreation(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 DISABLED: + 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) { + // Phase 1 - probe the target table. Failures here are "cannot expand safely" cases: the + // connector lacks DDL capability, or the target table does not exist yet. Delegating to the + // sink's original behavior is safe because no derived DDL has been issued. + 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; + try { + targetSchema = queryTargetSchema(createTableEvent.tableId()); + } catch (Exception e) { + LOG.warn( + "Failed to read the existing target table {} before expansion. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + return; + } + 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); + } + + // The plan needs repair DDL, but the sink cannot perform the specific required event type. + // This is a capability gap, not a transient failure, so delegate to the sink instead of + // failing the job (mirrors applyPlan's capability checks, which throw for EXPAND mode). + if (!plan.columnsToAdd.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + LOG.warn( + "Target table {} is missing columns {}, but ADD_COLUMN is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + getColumnNames(plan.columnsToAdd)); + return; + } + if (!plan.columnsToWiden.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + LOG.warn( + "Target table {} has narrow columns {}, but ALTER_COLUMN_TYPE is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + plan.columnsToWiden.keySet()); + return; + } + + // Phase 2 - apply and verify. At this point the expander has identified supportable + // differences and is about to issue derived DDL. A failure here (transient network/catalog + // error, DDL execution failure, or read-back verification failure) must propagate so the + // job fails over and the idempotent expander retries; otherwise the original + // CreateTableEvent + // would be applied to an existing table (typically a no-op) and the missing columns would + // be + // silently dropped forever. + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); + } + + 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 SQL templates (review and adjust to the target connector's DDL dialect before execution):%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) { + StringBuilder suggestions = new StringBuilder(); + for (Column columnToAdd : plan.columnsToAdd) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ADD COLUMN ") + .append(columnToAdd.getName()) + .append(" ") + .append(columnToAdd.getType()) + .append(";"); + } + for (Map.Entry<String, DataType> columnToWiden : plan.columnsToWiden.entrySet()) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ALTER COLUMN ") + .append(columnToWiden.getKey()) + .append(" TYPE ") + .append(columnToWiden.getValue()) + .append(";"); + } + return suggestions.toString(); + } + + /** + * Analyzes the pipeline schema against the current target schema and derives the safe DDL plan. + * Differences that cannot be fixed by safe DDL are collected in {@link + * ExpansionPlan#incompatibilities}. + */ + private ExpansionPlan analyze(CreateTableEvent createTableEvent, Schema currentTargetSchema) { + ExpansionPlan plan = new ExpansionPlan(); + Schema pipelineSchema = createTableEvent.getSchema(); + boolean columnNameCaseSensitive = expansionSupport.isColumnNameCaseSensitive(); + ColumnIndex targetColumns = indexColumns(currentTargetSchema, columnNameCaseSensitive); + plan.targetColumns = targetColumns; + Set<String> ambiguousColumnNames = + new HashSet<>( + indexColumns(pipelineSchema, columnNameCaseSensitive) + .getAmbiguousColumnNames()); + ambiguousColumnNames.addAll(targetColumns.getAmbiguousColumnNames()); + Set<String> keyColumns = + getKeyColumns(pipelineSchema, currentTargetSchema, columnNameCaseSensitive); + + for (Column pipelineColumn : pipelineSchema.getColumns()) { + String columnName = pipelineColumn.getName(); + String comparisonName = normalizeColumnName(columnName, columnNameCaseSensitive); + if (ambiguousColumnNames.contains(comparisonName)) { + plan.incompatibilities.add( + String.format( + "column \"%s\" is ambiguous under the target system's case-sensitivity rule", + columnName)); + continue; Review Comment: Confirmed resolved. Gating on `normalizeType(...)` while still handing the pipeline type to the sink is the behaviour I expected: the unsupported type is caught during the probe, so `TRY_EXPAND` delegates with no DDL issued and `EXPAND` reports it as an incompatibility. Thanks. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,808 @@ +/* + * 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. + * + * <p>This is a two-phase contract: the expander first tries to check/expand the existing target + * table, then signals whether the sink should still apply the original {@link + * CreateTableEvent}. + * + * @return {@code true} if the caller should proceed to apply the original {@link + * CreateTableEvent} to the sink; {@code false} means the framework has fully handled this + * event and the original {@link CreateTableEvent} must not be applied. {@code CHECK} mode + * returns {@code false} after a successful check so that no external DDL is issued by the + * pipeline. + */ + public boolean handleExistingTableCreation(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 DISABLED: + 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) { + // Phase 1 - probe the target table. Failures here are "cannot expand safely" cases: the + // connector lacks DDL capability, or the target table does not exist yet. Delegating to the + // sink's original behavior is safe because no derived DDL has been issued. + 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; + try { + targetSchema = queryTargetSchema(createTableEvent.tableId()); + } catch (Exception e) { + LOG.warn( + "Failed to read the existing target table {} before expansion. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + return; + } + 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); + } + + // The plan needs repair DDL, but the sink cannot perform the specific required event type. + // This is a capability gap, not a transient failure, so delegate to the sink instead of + // failing the job (mirrors applyPlan's capability checks, which throw for EXPAND mode). + if (!plan.columnsToAdd.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + LOG.warn( + "Target table {} is missing columns {}, but ADD_COLUMN is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + getColumnNames(plan.columnsToAdd)); + return; + } + if (!plan.columnsToWiden.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + LOG.warn( + "Target table {} has narrow columns {}, but ALTER_COLUMN_TYPE is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + plan.columnsToWiden.keySet()); + return; + } + + // Phase 2 - apply and verify. At this point the expander has identified supportable + // differences and is about to issue derived DDL. A failure here (transient network/catalog + // error, DDL execution failure, or read-back verification failure) must propagate so the + // job fails over and the idempotent expander retries; otherwise the original + // CreateTableEvent + // would be applied to an existing table (typically a no-op) and the missing columns would + // be + // silently dropped forever. + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); + } + + 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 SQL templates (review and adjust to the target connector's DDL dialect before execution):%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) { + StringBuilder suggestions = new StringBuilder(); + for (Column columnToAdd : plan.columnsToAdd) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ADD COLUMN ") + .append(columnToAdd.getName()) + .append(" ") + .append(columnToAdd.getType()) + .append(";"); + } + for (Map.Entry<String, DataType> columnToWiden : plan.columnsToWiden.entrySet()) { + suggestions + .append("\n - ALTER TABLE ") + .append(createTableEvent.tableId()) + .append(" ALTER COLUMN ") + .append(columnToWiden.getKey()) + .append(" TYPE ") + .append(columnToWiden.getValue()) + .append(";"); + } + return suggestions.toString(); + } + + /** + * Analyzes the pipeline schema against the current target schema and derives the safe DDL plan. + * Differences that cannot be fixed by safe DDL are collected in {@link + * ExpansionPlan#incompatibilities}. + */ + private ExpansionPlan analyze(CreateTableEvent createTableEvent, Schema currentTargetSchema) { + ExpansionPlan plan = new ExpansionPlan(); + Schema pipelineSchema = createTableEvent.getSchema(); + boolean columnNameCaseSensitive = expansionSupport.isColumnNameCaseSensitive(); + ColumnIndex targetColumns = indexColumns(currentTargetSchema, columnNameCaseSensitive); + plan.targetColumns = targetColumns; + Set<String> ambiguousColumnNames = + new HashSet<>( + indexColumns(pipelineSchema, columnNameCaseSensitive) + .getAmbiguousColumnNames()); + ambiguousColumnNames.addAll(targetColumns.getAmbiguousColumnNames()); + Set<String> keyColumns = + getKeyColumns(pipelineSchema, currentTargetSchema, columnNameCaseSensitive); + Review Comment: Thanks for the thorough fix. On the two points you raised: 1. The tightening is fine. A pipeline whose keys differ from the target table is exactly the situation this option exists to surface, and `DISABLED` remains available for setups that want the previous behaviour. 2. Agree with keeping `TRY_EXPAND` non-fatal on key mismatch - a best-effort mode should not turn a capability gap into a hard failure, and delegating leaves the decision with the connector. Documenting `CHECK`/`EXPAND` as the strict options is the right resolution. One related gap filed inline: the primary key comparison does not yet account for Paimon storing partition columns inside the primary key, which makes a pipeline-created partitioned table look key-incompatible on the next run. -- 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]
