leonardBang commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4091160154
########## 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: Thanks for the quick follow-up - the key comparison closes the gap I raised. One thing that now looks like a false positive: Paimon merges partition columns into the stored primary key, so a table that this pipeline created itself can never match the pipeline's declared PK here. `PaimonMetadataApplier#applyCreateTable` copies every partition key into `primaryKeys` before `catalog.createTable(...)` (asserted by `PaimonMetadataApplierTest`: a pipeline schema with `primaryKey("col1")` + `partitionKey("dt")` yields `table.primaryKeys() == [col1, dt]`), and `getExistingTableSchema` reads that list back verbatim. On the next run the comparison above sees `pipelinePrimaryKeys = {col1}` vs `targetPrimaryKeys = {col1, dt}` and reports a mismatch even though nothing is wrong. The same holds when partitioning comes from the sink's `partition.key` option, where the pipeline schema does not carry partition keys at all. The consequences per mode are that `CHECK`/`EXPAND` fail an otherwise valid job, and `TRY_EXPAND` takes the new skip path and issues no DDL, so the option silently does nothing for partitioned tables. I reproduced both with a target schema of `{col1, dt}` with PK `[col1, dt]` / partition `[dt]` against a pipeline schema that only adds one nullable column: `TRY_EXPAND` delegated with zero applied events, and `CHECK` threw with the primary-key message. Would it make sense to compare the identity key after removing partition columns from both sides (e.g. `targetPrimaryKeys - targetPartitionKeys` vs `pipelinePrimaryKeys - pipelinePartitionKeys`), or otherwise treat a target PK equal to `pipelinePK | pipelinePartitionKeys` as compatible? A case like this in `ExistingTableSchemaExpanderTest`, and ideally a partitioned table in the Paimon e2e, would keep the two rules from drifting apart again. -- 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]
