lvyanquan commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4091762930
########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,898 @@ +/* + * 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); + } + + // Keys are not repairable by expansion, so issuing the remaining column-level DDL would + // mutate a target table that can never serve the pipeline. Delegate the whole event to the + // sink instead, which also lets the connector's own key validation have the last word. + if (plan.keysIncompatible) { + LOG.warn( + "Target table {} has keys differing from the pipeline schema, which expansion cannot realign. Skipping expansion and delegating schema handling to the sink.", + createTableEvent.tableId()); + return; + } + + // 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); + } + + /** Renders every unresolved difference of a plan as a bullet list. */ + private static String describeDifferences(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()); + } + return differences.toString(); + } + + private SchemaEvolveException incompatibleException( + CreateTableEvent createTableEvent, ExpansionPlan plan) { + String message = + String.format( + "Existing target table %s cannot contain the pipeline schema:%s", + createTableEvent.tableId(), describeDifferences(plan)); + 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); + // Table keys are never realignable by ADD_COLUMN or ALTER_COLUMN_TYPE, and a target table + // that identifies rows differently silently merges or splits upstream records, so a + // mismatch has to be reported instead of looking like "nothing to expand". + validateTableKeys( + createTableEvent.tableId(), + pipelineSchema, + currentTargetSchema, + columnNameCaseSensitive, + plan); + + 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.incompatibilities.add( + String.format( + "target table is missing the non-addable column \"%s\"", + columnName)); + } else { + // A brand new column is never compared against an existing target column, so + // its type still has to be expressible by the target system. Without this + // check the incompatibility would only surface once the derived DDL runs in + // the apply phase, failing the job instead of being handled up front. + Optional<DataType> normalizedNewType = + normalizeType( + createTableEvent.tableId(), + columnName, + pipelineColumn.getType(), + currentTargetSchema); + if (!normalizedNewType.isPresent()) { + plan.incompatibilities.add( + String.format( + "pipeline type %s of missing column \"%s\" cannot be normalized to the target type system", + pipelineColumn.getType(), columnName)); + } else { + plan.columnsToAdd.add( + pipelineColumn.copy(pipelineColumn.getType().nullable())); + } + } + 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(), describeDifferences(remaining)); + if (strict) { + throw new SchemaEvolveException(createTableEvent, message); + } + LOG.warn("{}. Sink data may lose those columns.", message); + } + } + + private Optional<Schema> queryTargetSchema(TableId tableId) { + try { + return expansionSupport.getExistingTableSchema(tableId); + } catch (Exception e) { + // Propagate so the job fails over and retries, instead of proceeding to apply the + // original CreateTableEvent and silently dropping columns. + throw new FlinkRuntimeException( + "Failed to query schema of existing target table " + tableId, e); + } + } + + private Optional<DataType> normalizeType( + TableId tableId, + String columnName, + DataType pipelineType, + Schema existingTargetSchema) { + try { + DataType normalizedType = + expansionSupport.normalizeToTargetDataType( + tableId, columnName, pipelineType, existingTargetSchema); + if (normalizedType == null) { + LOG.warn( + "Target schema expansion support returned a null normalized type for {}.{}. Delegating this column to the sink.", + tableId, + columnName); + return Optional.empty(); + } + return Optional.of(normalizedType); + } catch (Exception e) { + LOG.warn( + "Failed to normalize type {} for {}.{}. Delegating this column to the sink.", + pipelineType, + tableId, + columnName, + e); + return Optional.empty(); + } + } + + private void applySchemaChange(SchemaChangeEvent event, Object changes) { + // Failures propagate to the caller: TRY_EXPAND catches and delegates to the sink, while + // CHECK/EXPAND fail the job. + LOG.info( + "Attempting to apply schema change event derived for existing table expansion: {} ({})", + event, + changes); + metadataApplier.applySchemaChange(event); + LOG.info( + "The schema change call for existing table expansion completed without an exception: {}", + event); + } + + private boolean supportsSchemaEvolutionType(SchemaChangeEventType eventType) { + try { + return metadataApplier.acceptsSchemaEvolutionType(eventType) + && metadataApplier.getSupportedSchemaEvolutionTypes().contains(eventType); + } catch (Exception e) { + LOG.warn( + "Failed to determine whether {} is enabled and supported. Delegating schema handling to the sink.", + eventType, + e); + return false; + } + } + + private static ColumnIndex indexColumns(Schema schema, boolean caseSensitive) { + Map<String, Column> columns = new HashMap<>(); + Set<String> ambiguousColumnNames = new HashSet<>(); + for (Column column : schema.getColumns()) { + String comparisonName = normalizeColumnName(column.getName(), caseSensitive); + if (ambiguousColumnNames.contains(comparisonName)) { + continue; + } + if (columns.putIfAbsent(comparisonName, column) != null) { + columns.remove(comparisonName); + ambiguousColumnNames.add(comparisonName); + } + } + return new ColumnIndex(columns, ambiguousColumnNames, caseSensitive); + } + + private static Set<String> getKeyColumns( + Schema pipelineSchema, Schema targetSchema, boolean caseSensitive) { + Set<String> keyColumns = new HashSet<>(); + pipelineSchema.primaryKeys().stream() + .map(columnName -> normalizeColumnName(columnName, caseSensitive)) + .forEach(keyColumns::add); + pipelineSchema.partitionKeys().stream() + .map(columnName -> normalizeColumnName(columnName, caseSensitive)) + .forEach(keyColumns::add); + targetSchema.primaryKeys().stream() + .map(columnName -> normalizeColumnName(columnName, caseSensitive)) + .forEach(keyColumns::add); + targetSchema.partitionKeys().stream() + .map(columnName -> normalizeColumnName(columnName, caseSensitive)) + .forEach(keyColumns::add); + return keyColumns; + } + + /** + * Records primary key and partition key differences between the pipeline schema and the + * existing target table, and flags the plan as unexpandable when any is found. + * + * <p>Primary keys are always compared because they define how records are merged. Partition + * keys are only compared when the pipeline declares them: most sources do not report + * partitioning at all, so a two-sided comparison would make externally partitioned target + * tables look incompatible. Key names are compared as case-normalized sets, matching what the + * connectors do in their own checks, so that an ordering-only difference is not reported. + */ + private void validateTableKeys( + TableId tableId, + Schema pipelineSchema, + Schema targetSchema, + boolean caseSensitive, + ExpansionPlan plan) { + Set<String> pipelinePrimaryKeys = + normalizeKeyNames(pipelineSchema.primaryKeys(), caseSensitive); + Set<String> targetPrimaryKeys = + normalizeKeyNames(targetSchema.primaryKeys(), caseSensitive); + if (!pipelinePrimaryKeys.equals(targetPrimaryKeys)) { Review Comment: Reproduced and fixed. Against a real Paimon catalog, a table created from `primaryKey("col1")` + `partitionKey("dt")` reads back as `[col1, dt]`, so the strict comparison flagged the pipeline's own table as key-incompatible. The comparison now uses the identity part of the key - primary key minus partition columns on both sides - so a merged key is not a difference, while a genuinely different key still is (an explicit case guards that). Your second variant is covered too: with partitioning coming from the sink's configuration, the target side still reduces to `[col1]`. The message and the option docs say partition columns are excluded. Both the framework and a real-catalog `PaimonMetadataApplierTest` case fail without the normalization. I did not extend the Paimon e2e - there the job also writes data, so a partitioned table has to line up with the sink's pre-partitioning rather than just being declared; glad to take that here or as a follow-up if you prefer. -- 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]
