leonardBang commented on code in PR #4540: URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4060797239
########## docs/content/docs/core-concept/schema-evolution.md: ########## @@ -73,6 +73,29 @@ This is the default schema evolution behavior. In this mode, all schema change events will be silently swallowed by `SchemaOperator` and never attempt to apply them to downstream sink. This is useful when your downstream sink is unready for any schema changes, but wants to keep receiving data from unchanged columns. +## Existing Table Schema Expansion + +Set the sink option `existing-table.schema-expansion.mode` to control how the framework handles the initial `CreateTableEvent` when the target table already exists. The default is `OFF`. For sinks that implement this capability, the framework may add missing non-key physical columns as nullable columns and safely widen non-key column types. Derived DDL events are logged. Review Comment: Would it make sense to clarify that `existing-table.schema-expansion.mode` is a framework-level sink option rather than a connector option? Currently, `YamlPipelineDefinitionParser` strips this key from the sink node before building `SinkDef.getConfig()`, so it never reaches the connector's `MetadataApplier` as a configuration property. A user reading this section might assume it behaves like `catalog.properties.*` and try to set it in connector-specific properties. It may also be helpful to list which connectors currently implement `ExistingTableSchemaExpansionSupport` (Paimon, Fluss) so users know the supported matrix upfront. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,767 @@ +/* + * 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. + * + * @return whether the caller should proceed to apply the original {@link CreateTableEvent} to + * the sink. {@code CHECK} mode returns {@code false} after a successful check so that no + * external DDL is issued by the pipeline. + */ + public boolean expand(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 OFF: + 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) { + try { + 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 = 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()); + for (String incompatibility : plan.incompatibilities) { + LOG.warn( + "Target table {} has an unsupported difference: {}. Delegating it to the sink.", + createTableEvent.tableId(), + incompatibility); + } + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); + } catch (Exception e) { + LOG.warn( + "Best-effort schema expansion failed for existing target table {}. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + } + } + + 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 statements (adjust to the target system's DDL dialect):%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) { Review Comment: Would it make sense to emphasize in the log message or docs that the suggested `ALTER TABLE` statements are dialect-agnostic templates? Currently the suggestions use a generic SQL syntax (`ALTER TABLE ... ADD COLUMN ...`, `ALTER TABLE ... ALTER COLUMN ... TYPE ...`) that may not be directly executable on all target systems (e.g. some sinks use `MODIFY COLUMN` or require additional clauses). The docs do mention "adjust to the target system's DDL dialect," but the log output itself does not carry this disclaimer. Adding a short prefix like "Suggested repair statements (review and adjust to the target dialect):" in the log would help on-call engineers avoid pasting them verbatim. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,767 @@ +/* + * 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. + * + * @return whether the caller should proceed to apply the original {@link CreateTableEvent} to + * the sink. {@code CHECK} mode returns {@code false} after a successful check so that no + * external DDL is issued by the pipeline. + */ + public boolean expand(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 OFF: + 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) { + try { + 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 = 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()); + for (String incompatibility : plan.incompatibilities) { + LOG.warn( + "Target table {} has an unsupported difference: {}. Delegating it to the sink.", + createTableEvent.tableId(), + incompatibility); + } + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); + } catch (Exception e) { + LOG.warn( + "Best-effort schema expansion failed for existing target table {}. Delegating schema handling to the sink.", + createTableEvent.tableId(), + e); + } Review Comment: Would it make sense to distinguish `TRY_EXPAND` failures by cause? Currently, when a transient error (e.g. network/catalog timeout) occurs after the expander has already identified supportable differences and started applying derived DDL, the failure is caught here, logged as a WARN, and the call site proceeds to apply the original `CreateTableEvent`. For sinks like Paimon, `CreateTableEvent` on an existing table is typically a no-op, so the missing columns are never added and subsequent data silently drops those columns. This is the same silent data loss the PR sets out to fix, just reachable through a transient-error window that is actually widened by the expansion itself (diff query + DDL + read-back verification). Could we split the catch into: (1) connector unsupported / target table missing — safe to delegate, and (2) identified differences but DDL/verification failed — propagate so the job fails over and the idempotent expander retries? The existing tests (`testAddsMissingColumnsAsNullableIdempotently`, `testWidensNarrowTargetTypeIdempotently`) already prove replay converges to no-op, so failover retry has no destructive side effects. It may also be helpful to document the real guarantee boundary of `TRY_EXPAND` in the docs. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java: ########## @@ -145,6 +154,66 @@ private void initializeBaseRuntimeState() { this.schemaManager = new SchemaManager(); } this.router = new TableIdRouter(routingRules, routeMode); + // TRY_EXPAND/EXPAND never run under IGNORE/EXCEPTION, so skip their initialization there. + // CHECK guards the initial table state and initializes regardless of the behavior. + if (existingTableSchemaExpansionMode != ExistingTableSchemaExpansionMode.OFF Review Comment: Would it make sense to fail-fast here for `EXPAND` when the connector does not provide `ExistingTableSchemaExpansionSupport`? Currently, only `EXPAND` throws `FlinkRuntimeException` in `handleMissingExpansionSupport()`, while `TRY_EXPAND` silently logs a WARN and continues. Since the user has explicitly configured a non-`OFF` mode, silently ignoring it for `TRY_EXPAND` may give users a false sense that expansion is active. Could we at least log at a higher severity (e.g. ERROR with a clear action item), or consider failing for any explicitly configured non-`OFF` mode when the connector lacks support? This would be more consistent with how the PR frames the problem of silent behavior. ########## flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java: ########## @@ -0,0 +1,767 @@ +/* + * 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. + * + * @return whether the caller should proceed to apply the original {@link CreateTableEvent} to + * the sink. {@code CHECK} mode returns {@code false} after a successful check so that no + * external DDL is issued by the pipeline. + */ + public boolean expand(CreateTableEvent createTableEvent) { Review Comment: Would it make sense to rename `expand()` to something that better exposes its contract? The return value controls whether the caller proceeds to apply the original `CreateTableEvent` to the sink, but the name `expand` reads as "perform the expansion," not "decide whether to also let the sink create the table." A name like `handleExistingTableCreation` or `prepareExistingTableCreation` might make the two-phase contract (try expansion, then signal whether the sink should still apply the original event) more obvious to future maintainers. It may also be helpful to strengthen the Javadoc to explicitly state that `return false` means "the framework has fully handled this event; do not apply the original CreateTableEvent." ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java: ########## @@ -114,6 +119,57 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() { SchemaChangeEventType.ALTER_COLUMN_TYPE); } + @Override + public Optional<ExistingTableSchemaExpansionSupport> getExistingTableSchemaExpansionSupport() { + return Optional.of(this); + } + + @Override + public Optional<Schema> getExistingTableSchema(TableId tableId) { Review Comment: Would it make sense to consolidate the lazy `catalog` initialization? Currently `getExistingTableSchema`, `isColumnNameCaseSensitive`, and `applySchemaChange` each independently trigger `FlinkCatalogFactory.createPaimonCatalog(catalogOptions)` when `catalog == null`. This means even a `CHECK`-only run that never issues DDL will pay for catalog initialization on both `getExistingTableSchema` and `isColumnNameCaseSensitive` calls. Could we centralize the lazy init into a single `getCatalog()` helper to make the lifecycle easier to reason about and avoid potential double-init races if these methods are ever called from different threads? It may also be helpful to add a test verifying `close()` is idempotent when multiple init paths have triggered. -- 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]
