lvyanquan commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4090752272
########## 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: Fixed — the guard in `verifyExpansion` already considered all three kinds of remaining differences, but only `incompatibilities` was rendered, hence the empty list. Difference rendering is now extracted into a shared `describeDifferences(ExpansionPlan)` helper used by both `verifyExpansion` and `incompatibleException` (which also removes the duplicated formatting there), so columns that are still missing or still too narrow are listed individually after an expansion that silently had no effect. ########## 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: Agreed, this was a real correctness gap. `getKeyColumns()` only unions both sides' keys to keep expansion away from them, so nothing ever compared them: with identical columns and types but a different or absent primary key, `analyze()` reported zero differences, `CHECK` passed and returned `false`, and the original `CreateTableEvent` was therefore never handed to the sink — which also bypassed `FlussMetaDataApplier.sanityCheck`, the only component that validates primary keys, bucket keys and partition keys. `analyze()` now calls `validateTableKeys()` before the column loop and records any mismatch in `plan.incompatibilities`. Primary keys are always compared, as case-normalized sets, matching how the connectors' own checks compare them, so an ordering-only difference is not reported. Partition keys are compared only when the pipeline declares them (through a `PARTITION BY` transform or a source that reports partitioning), since most sources do not and a two-sided comparison would flag externally partitioned target tables as incompatible. `CHECK` and `EXPAND` consequently fail on a mismatch, and `TRY_EXPAND` now skips expansion entirely — no DDL is issued against a table whose keys can never match; it logs a warning and delegates to the sink, consistent with the other phase-1 bail-outs. The option documentation states this rule. Two points for your explicit sign-off: 1. This is a behavior tightening: a job whose existing target table has a different primary key passed `CHECK` before and now fails, same under `EXPAND`. I think that is correct, since it is precisely what the connector-side check already rejects, but flagging it in case you want some existing setups to keep working. 2. Residual gap: under `TRY_EXPAND`, Paimon performs no key validation of its own either (`createTable(ignoreIfExists = true)`), so such a job still starts and writes under the target's key semantics with only a warning. Closing it would mean making `TRY_EXPAND` fatal, which contradicts its best-effort contract, so I documented `CHECK`/`EXPAND` as the strict options instead. Happy to change that if you prefer `TRY_EXPAND` to fail on key mismatch. ########## 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: Done — a brand new column is never compared against an existing target column, so an unsupported type only surfaced in phase 2 when the derived DDL ran, which now fails the job rather than degrading, even though a type the target cannot express is a capability gap and therefore phase-1 territory. `analyze()` now normalizes the new column's type too and records an incompatibility when normalization fails, so no doomed DDL is issued. The check is deliberately only a gate: `columnsToAdd` still carries the pipeline type, so the event handed to the sink is unchanged. The wording is kept distinct from the existing-column case so failures stay attributable. -- 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]
