lvyanquan commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4022275946
########## flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/ExistingTableSchemaExpansionSupport.java: ########## @@ -0,0 +1,53 @@ +/* + * 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.common.sink; + +import org.apache.flink.cdc.common.annotation.Experimental; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.exceptions.SchemaEvolveException; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.types.DataType; + +import java.io.Serializable; +import java.util.Optional; + +/** Connector-specific capabilities required to safely expand an existing target table schema. */ +@Experimental +public interface ExistingTableSchemaExpansionSupport extends Serializable { + + /** + * Returns the current schema of the target table, or {@link Optional#empty()} if the table does + * not exist. Target physical types must be converted to the same canonical CDC representation + * produced by {@link #normalizeToTargetDataType(TableId, String, DataType)}. + */ + Optional<Schema> getExistingTableSchema(TableId tableId) throws SchemaEvolveException; + + /** + * Converts a pipeline type to the canonical CDC representation of the physical type that this + * applier will create in the target system. For example, if both {@code CHAR(n)} and {@code + * VARCHAR(n)} map to target {@code VARCHAR(n)}, this method should return the same {@code + * VARCHAR(n)} representation for both input types. + * + * <p>This method must not query or modify the target table. + */ + DataType normalizeToTargetDataType( Review Comment: normalizeToTargetDataType should receive the existing target table context. Type mapping may depend on table-level properties, while the current contract prohibits querying the target table inside this method. Since ExistingTableSchemaExpander has already loaded the target schema, consider passing either the existing Schema or at least its table options to avoid incorrect normalization for property-dependent sinks. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java: ########## @@ -132,6 +138,28 @@ public void start() throws Exception { this.schemaManager = new SchemaManager(); } this.router = new TableIdRouter(routingRules, routeMode); + if (existingTableSchemaExpansionEnabled) { + try { + metadataApplier + .getExistingTableSchemaExpansionSupport() + .ifPresent( + support -> + this.existingTableSchemaExpander = + new ExistingTableSchemaExpander( + metadataApplier, support, behavior)); + } catch (Exception e) { + LOG.warn( + "Failed to initialize existing target table schema expansion. The sink's original schema handling will be used.", + e); + } + } + } + + /** Tries optional target schema expansion without changing the sink's failure behavior. */ + protected void expandExistingTableSchemaIfNeeded(SchemaChangeEvent schemaChangeEvent) { + if (existingTableSchemaExpander != null && schemaChangeEvent instanceof CreateTableEvent) { + existingTableSchemaExpander.expand((CreateTableEvent) schemaChangeEvent); Review Comment: ExpansionResult is ignored by all production callers. Please either consume it—for example through metrics or explicit caller-side handling—or simplify expand() to return void. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,635 @@ +/* + * 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.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.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; + +/** Performs best-effort safe schema expansion for an existing target table. */ +@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; + + public ExistingTableSchemaExpander( + MetadataApplier metadataApplier, + ExistingTableSchemaExpansionSupport expansionSupport, + SchemaChangeBehavior schemaChangeBehavior) { + this.metadataApplier = metadataApplier; + this.expansionSupport = expansionSupport; + this.schemaChangeBehavior = schemaChangeBehavior; + } + + /** Tries safe expansions without imposing new compatibility failures. */ + public ExpansionResult expand(CreateTableEvent createTableEvent) { + try { + return expandInternal(createTableEvent); + } catch (Exception e) { + LOG.warn( + "Unexpected error while expanding target table {}. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + return ExpansionResult.DELEGATE_TO_SINK; + } + } + + private ExpansionResult expandInternal(CreateTableEvent createTableEvent) throws Exception { + Optional<Schema> targetSchema = queryTargetSchema(createTableEvent.tableId()); Review Comment: IGNORE and EXCEPTION are only checked later in supportsSchemaEvolutionType, after querying the target schema and normalizing its columns. Since these modes can never apply expansion DDL, could we short-circuit before the catalog query? The same optimization applies when neither ADD_COLUMN nor ALTER_COLUMN_TYPE is accepted/supported. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,635 @@ +/* + * 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.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.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; + +/** Performs best-effort safe schema expansion for an existing target table. */ +@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; + + public ExistingTableSchemaExpander( + MetadataApplier metadataApplier, + ExistingTableSchemaExpansionSupport expansionSupport, + SchemaChangeBehavior schemaChangeBehavior) { + this.metadataApplier = metadataApplier; + this.expansionSupport = expansionSupport; + this.schemaChangeBehavior = schemaChangeBehavior; + } + + /** Tries safe expansions without imposing new compatibility failures. */ + public ExpansionResult expand(CreateTableEvent createTableEvent) { + try { + return expandInternal(createTableEvent); + } catch (Exception e) { + LOG.warn( + "Unexpected error while expanding target table {}. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + return ExpansionResult.DELEGATE_TO_SINK; + } + } + + private ExpansionResult expandInternal(CreateTableEvent createTableEvent) throws Exception { + Optional<Schema> targetSchema = queryTargetSchema(createTableEvent.tableId()); + if (!targetSchema.isPresent()) { + return ExpansionResult.NO_ACTION; + } + + Schema pipelineSchema = createTableEvent.getSchema(); + Schema currentTargetSchema = targetSchema.get(); + boolean columnNameCaseSensitive = expansionSupport.isColumnNameCaseSensitive(); + ColumnIndex targetColumns = indexColumns(currentTargetSchema, columnNameCaseSensitive); + Set<String> ambiguousColumnNames = + new HashSet<>( + indexColumns(pipelineSchema, columnNameCaseSensitive) + .getAmbiguousColumnNames()); + ambiguousColumnNames.addAll(targetColumns.getAmbiguousColumnNames()); + Set<String> keyColumns = + getKeyColumns(pipelineSchema, currentTargetSchema, columnNameCaseSensitive); + + List<Column> columnsToAdd = new ArrayList<>(); + Map<String, DataType> columnsToWiden = new LinkedHashMap<>(); + Map<String, DataType> expectedPipelineTypes = new HashMap<>(); + boolean delegateToSink = false; + + for (Column pipelineColumn : pipelineSchema.getColumns()) { + String columnName = pipelineColumn.getName(); + String comparisonName = normalizeColumnName(columnName, columnNameCaseSensitive); + if (ambiguousColumnNames.contains(comparisonName)) { + delegateToSink = true; + LOG.info( + "Column name {} in target table {} is ambiguous under the target system's case-sensitivity rule. Delegating this difference to the sink.", + columnName, + createTableEvent.tableId()); + continue; + } + + Column targetColumn = targetColumns.get(columnName); + if (targetColumn == null) { + if (pipelineColumn.isPhysical() && !keyColumns.contains(comparisonName)) { + columnsToAdd.add(pipelineColumn.copy(pipelineColumn.getType().nullable())); + } else { + delegateToSink = true; + LOG.info( + "Target table {} is missing special column {}. Delegating this difference to the sink.", + createTableEvent.tableId(), + columnName); + } + continue; + } + + Optional<DataType> normalizedPipelineTypeOptional = + normalizeType(createTableEvent.tableId(), columnName, pipelineColumn.getType()); + if (!normalizedPipelineTypeOptional.isPresent()) { + delegateToSink = true; + continue; + } + DataType normalizedPipelineType = normalizedPipelineTypeOptional.get().nullable(); + DataType targetType = targetColumn.getType().nullable(); + + if (pipelineColumn.getType().isNullable() && !targetColumn.getType().isNullable()) { + delegateToSink = true; + LOG.info( + "Target column {}.{} is NOT NULL while the pipeline column is nullable. Delegating this difference to the sink.", + createTableEvent.tableId(), + columnName); + } + + if (canContain(targetType, normalizedPipelineType)) { + continue; + } + if (keyColumns.contains(comparisonName)) { + delegateToSink = true; + LOG.info( + "Target key column {}.{} cannot contain pipeline type {}. Delegating this difference to the sink.", + createTableEvent.tableId(), + columnName, + normalizedPipelineType); + continue; + } + + Optional<DataType> widenedType = getSafeWidenedType(targetType, normalizedPipelineType); + if (!widenedType.isPresent()) { + delegateToSink = true; + LOG.info( + "Target column {}.{} with type {} cannot safely contain pipeline type {}. Delegating this difference to the sink.", + createTableEvent.tableId(), + columnName, + targetType, + normalizedPipelineType); + continue; + } + + Optional<DataType> normalizedWidenedTypeOptional = + normalizeType(createTableEvent.tableId(), columnName, widenedType.get()); + if (!normalizedWidenedTypeOptional.isPresent()) { + delegateToSink = true; + continue; + } + DataType normalizedWidenedType = normalizedWidenedTypeOptional.get().nullable(); + if (!canContain(normalizedWidenedType, targetType) + || !canContain(normalizedWidenedType, normalizedPipelineType)) { + delegateToSink = true; + LOG.info( + "Target system normalizes proposed type {} for {}.{} to {}, which is not a safe widening. Delegating this difference to the sink.", + widenedType.get(), + createTableEvent.tableId(), + columnName, + normalizedWidenedType); + continue; + } + + String targetColumnName = targetColumn.getName(); + columnsToWiden.put( + targetColumnName, widenedType.get().copy(targetColumn.getType().isNullable())); + expectedPipelineTypes.put(targetColumnName, normalizedPipelineType); + } + + List<Column> addedColumns = new ArrayList<>(); + Map<String, DataType> widenedColumns = new LinkedHashMap<>(); + + if (!columnsToAdd.isEmpty()) { + if (supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + AddColumnEvent addColumnEvent = + new AddColumnEvent( + createTableEvent.tableId(), + columnsToAdd.stream() + .map(AddColumnEvent.ColumnWithPosition::new) + .collect(Collectors.toList())); + if (applySchemaChange(addColumnEvent, columnsToAdd)) { + addedColumns.addAll(columnsToAdd); + } else { + delegateToSink = true; + } + } else { + delegateToSink = true; + LOG.info( + "Target table {} is missing columns {}, but ADD_COLUMN is not enabled or supported. Delegating this difference to the sink.", + createTableEvent.tableId(), + getColumnNames(columnsToAdd)); + } + } + + if (!columnsToWiden.isEmpty()) { + if (supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + AlterColumnTypeEvent alterColumnTypeEvent = + new AlterColumnTypeEvent( + createTableEvent.tableId(), + columnsToWiden, + columnsToWiden.keySet().stream() + .collect( + Collectors.toMap( + columnName -> columnName, + columnName -> + targetColumns + .get(columnName) + .getType()))); + if (applySchemaChange(alterColumnTypeEvent, columnsToWiden)) { + widenedColumns.putAll(columnsToWiden); + } else { + delegateToSink = true; + } + } else { + delegateToSink = true; + LOG.info( + "Target table {} has narrow columns {}, but ALTER_COLUMN_TYPE is not enabled or supported. Delegating this difference to the sink.", + createTableEvent.tableId(), + columnsToWiden.keySet()); + } + } + + if (addedColumns.isEmpty() && widenedColumns.isEmpty()) { + return delegateToSink ? ExpansionResult.DELEGATE_TO_SINK : ExpansionResult.NO_ACTION; + } + + Optional<Schema> refreshedTargetSchema = queryTargetSchema(createTableEvent.tableId()); Review Comment: The post-DDL schema refresh should also be justified because it currently affects only diagnostics and this unused return value. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java: ########## @@ -132,6 +138,28 @@ public void start() throws Exception { this.schemaManager = new SchemaManager(); } this.router = new TableIdRouter(routingRules, routeMode); + if (existingTableSchemaExpansionEnabled) { + try { + metadataApplier + .getExistingTableSchemaExpansionSupport() + .ifPresent( Review Comment: When `existing-table.schema-expansion.enabled` is explicitly enabled but `getExistingTableSchemaExpansionSupport()` returns `Optional.empty()`, the option is silently ignored. This may lead users to believe that existing target tables will be expanded when the sink does not actually support it. Could we handle the empty case explicitly and log a warning identifying the MetadataApplier implementation? -- 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]
