haruki-830 commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4024486339
########## 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: Actually, the post-DDL schema refresh was only used for diagnostic logging, whereas it could introduce an additional catalog query and potentially trigger false warnings due to delays in metadata visibility. Given these trade-offs, I've removed it. -- 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]
