leonardBang commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4089884310
########## 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: Would it make sense to also include `remaining.columnsToAdd` and `remaining.columnsToWiden` in the formatted message here? Currently, when `remaining.incompatibilities` is empty but some columns failed to be added or widened by DDL, the error/warning message reads `Target table ... still has unresolved differences after expansion: []`. Formatting the remaining columns to add/widen (similar to how `incompatibleException` constructs its error message) would make it clearer to users which columns remain unresolved. ########## 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: Would it make sense to validate the consistency between `pipelineSchema.primaryKeys()` / `partitionKeys()` and `currentTargetSchema.primaryKeys()` / `partitionKeys()` here? Currently, `getKeyColumns` collects key column names to prevent adding or widening keys, but `analyze()` does not verify whether the upstream and existing target tables have matching primary keys and partition keys. If a target table already exists with a different primary key (e.g., target has `[name]` or no PK while upstream has `[id]`), `analyze()` could report 0 differences. In `CHECK` mode, this would falsely pass validation and return `false`, bypassing sink-side checks (such as `FlussMetaDataApplier.sanityCheck`) and potentially leading to unexpected upsert or partition behavior during runtime writes. Could we record any primary key or partition key mismatch in `plan.incompatibilities`? ########## 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: Would it make sense to invoke `normalizeType(...)` on missing columns before adding them to `plan.columnsToAdd` here? Currently, when `targetColumn == null`, `pipelineColumn` is directly added to `plan.columnsToAdd` without checking whether its type can be normalized to the target system. In `TRY_EXPAND` mode, if an upstream column type is unsupported by the target sink, it will not be detected during Phase 1 probe and will instead trigger an exception in Phase 2 `applyPlan` when executing DDL, which propagates out to fail over the job rather than gracefully delegating to the sink. Could we check `normalizeType(...)` here and record it as an incompatibility if normalization fails or returns empty? -- 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]
