You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "haruki-830 (via GitHub)" <gi...@apache.org> on 2026/09/14 07:20:02 UTC
[PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
haruki-830 opened a new pull request, #4540:
URL: https://github.com/apache/flink-cdc/pull/4540
## What is the purpose of this pull request?
This PR introduces an opt-in best-effort schema expansion capability for existing sink tables during the initial `CreateTableEvent`.
When the target table already exists and its schema is narrower than the incoming schema, some sinks may ignore input columns that do not exist in the target table, potentially causing silent data loss.
When enabled, the framework attempts conservative schema expansion, including adding missing nullable non-key columns and safely widening non-key column types. Unsupported, unsafe, or failed operations are delegated to the sink's existing handling without introducing new framework-level fail-fast behavior.
## Brief change log
- Add the sink option `existing-table.schema-expansion.enabled`, disabled by default.
- Add an optional `MetadataApplier` extension for querying and normalizing the existing target schema.
- Perform best-effort schema expansion during initial table creation.
- Log derived DDL events and verify the target schema after successful operations.
- Integrate the extension with both Paimon and Fluss sinks.
- Add related tests and documentation.
---
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs and JavaDocs
---
## JIRA issue
[https://issues.apache.org/jira/browse/FLINK-40647](https://issues.apache.org/jira/browse/FLINK-40647)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#issuecomment-5708295606
Could we add an end-to-end test that exercises the full configuration pipeline? The current tests are thorough at the algorithm level (ExistingTableSchemaExpanderTest) and partially cover real connector behavior (FlussMetadataApplierTest, PaimonMetadataApplierTest), but they all bypass the wiring path:
```
YAML option → Composer → SchemaOperatorFactory → SchemaRegistry / BatchSchemaOperator → ExistingTableSchemaExpander → MetadataApplier
```
No test currently verifies that the existing-table.schema-expansion.enabled flag actually reaches the expander through the regular streaming, distributed streaming, and batch execution paths. A regression in any of these wiring hops would not be caught.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#issuecomment-5810055010
## Review Summary
Second-round review on `4ae9e39c5`. All three findings from the previous round are addressed, and the fixes hold up against the code and the new tests:
- `verifyExpansion` now renders all three kinds of unresolved difference through the shared `describeDifferences(plan)` helper, so a no-op DDL no longer reports an empty `[]`.
- `analyze()` now calls `validateTableKeys()` and records key differences as incompatibilities, so `CHECK`/`EXPAND` fail instead of silently passing a table that identifies rows differently.
- Missing columns now go through `normalizeType(...)` as a phase-1 gate, so an unrepresentable type is reported rather than surfacing as a DDL failure in phase 2.
The behaviour tightening you flagged is fine, and keeping `TRY_EXPAND` non-fatal on key mismatch is consistent with its best-effort contract.
### Remaining issue
**`validateTableKeys()` does not account for sinks that store partition columns inside the primary key.** `PaimonMetadataApplier#applyCreateTable` merges every partition key into `primaryKeys` before `catalog.createTable(...)` - `PaimonMetadataApplierTest` asserts that a pipeline schema with `primaryKey("col1")` + `partitionKey("dt")` produces a table whose `primaryKeys()` is `[col1, dt]` - and `getExistingTableSchema` reads that augmented list back. The strict set comparison therefore reports a pipeline-created partitioned table as key-incompatible on the next run. `CHECK`/`EXPAND` fail a valid job, and `TRY_EXPAND` takes the new skip path and issues no DDL at all, so the option quietly does nothing for partitioned tables. Filed inline with a suggested normalisation.
### Checklist
- [ ] Normalise partition columns out of the primary key comparison
- [ ] Add a partitioned-target unit case, and ideally a partitioned table in the Paimon e2e
## Leonard Verdict
**Ready to merge:** With fixes
**Reason:** The three previous gaps are properly closed with matching tests. The remaining item is a false-positive risk in the new key check on the most common Paimon layout (partitioned primary-key tables), which can either fail valid jobs or silently disable expansion; it should be normalised before merge.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4024473652
##########
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:
Thanks for pointing it out! Since production callers do not need to branch on the expansion result and always continue with the sink’s original handling, I’ve changed expand() to return void.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4089884310
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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.columnsToAdd.add(pipelineColumn.copy(pipelineColumn.getType().nullable()));
+ } else {
+ plan.incompatibilities.add(
+ String.format(
+ "target table is missing the non-addable column \"%s\"",
+ columnName));
+ }
+ 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(), remaining.incompatibilities);
+ if (strict) {
+ throw new SchemaEvolveException(createTableEvent, message);
Review Comment:
Would it make sense to also include `remaining.columnsToAdd` and `remaining.columnsToWiden` in the formatted message here? Currently, when `remaining.incompatibilities` is empty but some columns failed to be added or widened by DDL, the error/warning message reads `Target table ... still has unresolved differences after expansion: []`. Formatting the remaining columns to add/widen (similar to how `incompatibleException` constructs its error message) would make it clearer to users which columns remain unresolved.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
Review Comment:
Would it make sense to validate the consistency between `pipelineSchema.primaryKeys()` / `partitionKeys()` and `currentTargetSchema.primaryKeys()` / `partitionKeys()` here? Currently, `getKeyColumns` collects key column names to prevent adding or widening keys, but `analyze()` does not verify whether the upstream and existing target tables have matching primary keys and partition keys. If a target table already exists with a different primary key (e.g., target has `[name]` or no PK while upstream has `[id]`), `analyze()` could report 0 differences. In `CHECK` mode, this would falsely pass validation and return `false`, bypassing sink-side checks (such as `FlussMetaDataApplier.sanityCheck`) and potentially leading to unexpected upsert or partition behavior during runtime writes. Could we record any primary key or partition key mismatch in `plan.incompatibilities`?
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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;
Review Comment:
Would it make sense to invoke `normalizeType(...)` on missing columns before adding them to `plan.columnsToAdd` here? Currently, when `targetColumn == null`, `pipelineColumn` is directly added to `plan.columnsToAdd` without checking whether its type can be normalized to the target system. In `TRY_EXPAND` mode, if an upstream column type is unsupported by the target sink, it will not be detected during Phase 1 probe and will instead trigger an exception in Phase 2 `applyPlan` when executing DDL, which propagates out to fail over the job rather than gracefully delegating to the sink. Could we check `normalizeType(...)` here and record it as an incompatibility if normalization fails or returns empty?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4024815090
##########
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:
Addressed in the latest commit.
When schema expansion is explicitly enabled but the MetadataApplier does not provide the required support, the framework now logs a warning with the applier implementation and continues with the sink’s original handling.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4080146385
##########
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:
Consolidated in fe288eec0: `getExistingTableSchema`, `isColumnNameCaseSensitive` and `applySchemaChange` all go through a single private `getCatalog()`, and `close()` now also nulls the cached instance so the applier stays reusable after being closed. Added `testCloseIsIdempotentAndRecreatesCatalogOnReuse`, which triggers lazy creation through `applySchemaChange`, closes twice, applies again and asserts the recreated catalog sees the table.
##########
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:
Added (fe288eec0): the suggestion block is now introduced by `Suggested repair SQL templates (review and adjust to the target connector's DDL dialect before execution):` inside the aggregated `SchemaEvolveException` message, so it reaches operators in the failure text rather than only in the docs; the docs echo the same caveat. Differences that cannot be repaired safely still get no suggested statement.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4024844209
##########
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:
Thanks for the suggestion! I’ve added the existing target schema as a parameter to normalizeToTargetDataType.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4090752272
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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.columnsToAdd.add(pipelineColumn.copy(pipelineColumn.getType().nullable()));
+ } else {
+ plan.incompatibilities.add(
+ String.format(
+ "target table is missing the non-addable column \"%s\"",
+ columnName));
+ }
+ 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(), remaining.incompatibilities);
+ if (strict) {
+ throw new SchemaEvolveException(createTableEvent, message);
Review Comment:
Fixed — the guard in `verifyExpansion` already considered all three kinds of remaining differences, but only `incompatibilities` was rendered, hence the empty list. Difference rendering is now extracted into a shared `describeDifferences(ExpansionPlan)` helper used by both `verifyExpansion` and `incompatibleException` (which also removes the duplicated formatting there), so columns that are still missing or still too narrow are listed individually after an expansion that silently had no effect.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
Review Comment:
Agreed, this was a real correctness gap. `getKeyColumns()` only unions both sides' keys to keep expansion away from them, so nothing ever compared them: with identical columns and types but a different or absent primary key, `analyze()` reported zero differences, `CHECK` passed and returned `false`, and the original `CreateTableEvent` was therefore never handed to the sink — which also bypassed `FlussMetaDataApplier.sanityCheck`, the only component that validates primary keys, bucket keys and partition keys.
`analyze()` now calls `validateTableKeys()` before the column loop and records any mismatch in `plan.incompatibilities`. Primary keys are always compared, as case-normalized sets, matching how the connectors' own checks compare them, so an ordering-only difference is not reported. Partition keys are compared only when the pipeline declares them (through a `PARTITION BY` transform or a source that reports partitioning), since most sources do not and a two-sided comparison would flag externally partitioned target tables as incompatible. `CHECK` and `EXPAND` consequently fail on a mismatch, and `TRY_EXPAND` now skips expansion entirely — no DDL is issued against a table whose keys can never match; it logs a warning and delegates to the sink, consistent with the other phase-1 bail-outs. The option documentation states this rule.
Two points for your explicit sign-off:
1. This is a behavior tightening: a job whose existing target table has a different primary key passed `CHECK` before and now fails, same under `EXPAND`. I think that is correct, since it is precisely what the connector-side check already rejects, but flagging it in case you want some existing setups to keep working.
2. Residual gap: under `TRY_EXPAND`, Paimon performs no key validation of its own either (`createTable(ignoreIfExists = true)`), so such a job still starts and writes under the target's key semantics with only a warning. Closing it would mean making `TRY_EXPAND` fatal, which contradicts its best-effort contract, so I documented `CHECK`/`EXPAND` as the strict options instead. Happy to change that if you prefer `TRY_EXPAND` to fail on key mismatch.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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;
Review Comment:
Done — a brand new column is never compared against an existing target column, so an unsupported type only surfaced in phase 2 when the derived DDL ran, which now fails the job rather than degrading, even though a type the target cannot express is a capability gap and therefore phase-1 territory. `analyze()` now normalizes the new column's type too and records an incompatibility when normalization fails, so no doomed DDL is issued. The check is deliberately only a gate: `columnsToAdd` still carries the pipeline type, so the event handed to the sink is unchanged. The wording is kept distinct from the existing-column case so failures stay attributable.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4091165266
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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.columnsToAdd.add(pipelineColumn.copy(pipelineColumn.getType().nullable()));
+ } else {
+ plan.incompatibilities.add(
+ String.format(
+ "target table is missing the non-addable column \"%s\"",
+ columnName));
+ }
+ 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(), remaining.incompatibilities);
+ if (strict) {
+ throw new SchemaEvolveException(createTableEvent, message);
Review Comment:
Confirmed resolved - `describeDifferences(plan)` renders all three kinds now, and the `doesNotContain("after expansion: []")` assertion in `testUnresolvedDifferencesAfterExpansionAreListed` pins the behaviour. Thanks.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
+ 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;
Review Comment:
Confirmed resolved. Gating on `normalizeType(...)` while still handing the pipeline type to the sink is the behaviour I expected: the unsupported type is caught during the probe, so `TRY_EXPAND` delegates with no DDL issued and `EXPAND` reports it as an incompatibility. Thanks.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java:
##########
@@ -0,0 +1,808 @@
+/*
+ * 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);
+ }
+
+ // 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);
+ }
+
+ 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 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);
+
Review Comment:
Thanks for the thorough fix. On the two points you raised:
1. The tightening is fine. A pipeline whose keys differ from the target table is exactly the situation this option exists to surface, and `DISABLED` remains available for setups that want the previous behaviour.
2. Agree with keeping `TRY_EXPAND` non-fatal on key mismatch - a best-effort mode should not turn a capability gap into a hard failure, and delegating leaves the decision with the connector. Documenting `CHECK`/`EXPAND` as the strict options is the right resolution.
One related gap filed inline: the primary key comparison does not yet account for Paimon storing partition columns inside the primary key, which makes a pipeline-created partitioned table look key-incompatible on the next run.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4080144712
##########
docs/content.zh/docs/core-concept/schema-evolution.md:
##########
@@ -75,6 +75,29 @@ pipeline:
在此模式下，所有架构更改事件都将被 `SchemaOperator` 默默接收，并且永远不会尝试将它们应用于下游接收器。
当您的下游接收器尚未准备好进行任何架构更改，但想要继续从未更改的列中接收数据时，这很有用。
+## 已有目标表的安全 Schema 扩展
+
+设置 Sink 选项 `existing-table.schema-expansion.mode` 来控制框架在初始 `CreateTableEvent` 遇到已有目标表时的处理方式，默认值为 `OFF`。对于实现了该能力的 Sink，框架可能将缺失的普通非键物理列按 nullable 补充，并安全拓宽普通非键列类型；派生的 DDL 事件会记录在日志中。
+
+| 模式 | 已有目标表 | 目标表不存在 | 失败处理 |
+|---|---|---|---|
+| `OFF` | 不检查、不扩展，保持 Sink 原行为 | Sink 原生建表 | 不适用 |
Review Comment:
Good catch — renamed the default to `DISABLED` (fe288eec0). The rename also removes the YAML footgun itself: a bare `OFF` scalar is parsed as the boolean `false` by YAML 1.1, so `parseExpansionMode` no longer needs its `false`-accepting workaround and now rejects any non-textual value outright. Error messages and both docs pages list `DISABLED, CHECK, TRY_EXPAND, EXPAND`, and `YamlPipelineDefinitionParserTest` gained a case asserting that a boolean `false` is rejected.
##########
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:
Makes sense, and it is now split exactly along those lines (fe288eec0). The probe phase — `supportsAnyExpansionDdl()`, reading the target schema, and the per-event-type capability gaps — still delegates to the sink, since no derived DDL has been issued at that point. The apply/verify phase (`applyPlan` + `verifyExpansion`) no longer swallows anything: the failure propagates so the job fails over and the idempotent expander retries, instead of letting `CreateTableEvent` hit an already-existing table and silently drop the missing columns forever. Covered by `testTryExpandFailsFastOnUnsupportedDdlFailure`, `testTryExpandFailsFastOnTransientDdlFailure` and `testTryExpandFailsFastOnReadBackQueryFailure`, with the replay-safety premise coming from `testAddsMissingColumnsAsNullableIdempotently` / `testWidensNarrowTargetTypeIdempotently`. The docs now also state the boundary explicitly: `TRY_EXPAND` only swallows mechanism failures once the connector is known to support the capability
, and it never guarantees that all upstream columns land in the target table after a failed expansion.
##########
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:
Done (fe288eec0). The section now states that this is a framework-level option consumed by the `sink` block: it is stripped from the sink configuration before the connector is created, so it never reaches `MetadataApplier` as a configuration property. A short option / scope / "passed to connector" table contrasts it with `catalog.properties.*`, including that nesting it under `catalog.properties.existing-table.schema-expansion.mode` has no effect. It also names Paimon and Fluss as the connectors currently implementing `ExistingTableSchemaExpansionSupport`, and says any other non-`DISABLED` mode fails fast. Mirrored in the zh page.
##########
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:
Agreed, and it is not just louder logging now: `handleMissingExpansionSupport()` throws unconditionally for every explicitly configured mode, so the log-and-continue path under `TRY_EXPAND` is gone (fe288eec0). Covered by `SchemaCoordinatorTest.tryExpandModeFailsWhenConnectorLacksExpansionSupport`, and the `TRY_EXPAND` description now says a connector without the support is a configuration error rather than a silent degradation — same framing as the rest of the PR.
##########
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:
Adopted `handleExistingTableCreation` (your first suggestion) in fe288eec0, and the Javadoc now spells the two-phase contract out: `true` means the caller should still apply the original `CreateTableEvent` to the sink; `false` means the framework has fully handled the event and it must not be applied — which is `CHECK`'s success path.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#issuecomment-5713777617
From my side, I have tow advice in design side, @haruki-830 @leonardBang , @lvyanquan , WDYT?
---
### [Suggestion] Add a check-only mode for pipelines where the target schema is managed externally
>
> Not every user wants CDC to manage the target table's schema. When the table structure is owned by an external process (DBA change windows, a separate DDL orchestration system, data-platform governance), CDC should issue **no DDL at all** — but it must also not silently drop data when the upstream schema turns out to be wider than the existing target table. Currently this PR has no story for that user group: with the new option off, we are back to the pre-existing silent-drop behavior.
>
> I'd like to request a validation-only mode alongside the new expansion option, e.g.:
>
> ```yaml
> sink:
> existing-table.schema-expansion.mode: OFF | CHECK | EXPAND
> ```
>
> In `CHECK` mode, when the initial `CreateTableEvent` encounters an existing target table, the framework computes the same diff it already computes in `ExistingTableSchemaExpander`, but instead of deriving and applying DDL it:
>
> - **passes** only if every upstream column maps to a target column that can contain it (reusing the existing `canContain` rules), and
> - **fails the job otherwise**, with a single aggregated `SchemaEvolveException` that lists every difference (table, column, upstream type vs. target type) together with the suggested `ALTER TABLE` statements the external process can review and apply.
>
> Why I think this is worth it:
>
> - The diff and type-compatibility logic already exists in this PR; `CHECK` is essentially "compute the plan, then throw instead of apply", so the incremental cost is small.
> - It serves a use case `EXPAND` intentionally does not: users whose contract is "CDC never touches my DDL, but must fail loudly instead of dropping columns". For them, a precise failure message is strictly better than both silent loss and auto-evolution.
> - We run a similar validation-first mode in an internal deployment and have found the aggregated per-column error message (with suggested repair SQL) very effective for on-call triage — users fix the target table on their side and restart, without CDC ever having mutated their schema.
> - One semantic point worth deciding explicitly: `CHECK` should be independent of `schema.change.behavior`, since it guards the *initial* table state rather than runtime schema evolution. The docs should state this.
---
### [Concern] A transient failure of a *supported* expansion silently degrades into permanent column loss
>
> For differences the expander has already classified as supportable and safe, a transient network/database error during the derived DDL (or the post-expansion verification) currently results in:
>
> 1. a `WARN` + `DELEGATE_TO_SINK` from `ExistingTableSchemaExpander`;
> 2. the call site (`expandExistingTableSchemaIfNeeded`) **discarding the result** and proceeding to apply the original `CreateTableEvent`, which succeeds once the network recovers;
> 3. for Paimon, an existing table + `CreateTableEvent` is treated as redundant and skipped — so the missing columns are **never added**, and since the initial `CreateTableEvent` is one-shot, nothing retriggers the expansion later;
> 4. the sink then silently drops those columns' data forever.
>
> That is exactly the silent data loss this PR set out to fix — now reachable through a transient-error window that is actually *widened* by the expansion itself (≥3 extra round trips: diff query, DDL, read-back verification).
>
> It is also inconsistent with how the same class of failure is handled elsewhere in the framework: a runtime `AddColumnEvent` that fails in `EVOLVE` mode is rethrown, fails the job, and converges via failover + idempotent replay. The initial backfill of columns deserves no weaker a guarantee.
>
> Suggested fix (any of these would work, in order of preference):
>
> - **Propagate the exception** on derived-DDL / verification failures (narrow the catch-all in `ExistingTableSchemaExpander.applySchemaChange`) so it flows into the existing `applyAndUpdateEvolvedSchemaChange` error handling. This is safe: the expander is idempotent — the PR's own tests (`testAddsMissingColumnsAsNullableIdempotently`, `testWidensNarrowTargetTypeIdempotently`) prove replay converges to `NO_ACTION` — so failover retry has no destructive side effects. Unsupported/incompatible differences could still go through the `UnsupportedSchemaChangeEventException` path so `TRY_EVOLVE` keeps its tolerant semantics.
> - Or gate fail-fast behind a strict variant (e.g. `mode: EXPAND_STRICT`, or fail fast when `schema.change.behavior=EVOLVE`), keeping the best-effort default unchanged.
> - As a complement, a bounded retry with backoff for transient errors inside the expander would shrink the window without paying for a failover.
>
> Independently of which semantics is chosen: `ExpansionResult` is currently invisible — `EXPANDED`, `NO_ACTION`, and `DELEGATE_TO_SINK` are behaviorally identical after the call. At minimum please expose the outcome (a counter/metric, or include it in `SchemaChangeResponse`) so operators can detect "expansion did not take effect and columns are being dropped" from something other than WARN logs.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4044095706
##########
docs/content/docs/core-concept/schema-evolution.md:
##########
@@ -75,16 +75,27 @@ This is useful when your downstream sink is unready for any schema changes, but
## Existing Table Schema Expansion
-Set the sink option `existing-table.schema-expansion.enabled` to `true` to try safe schema expansion when the initial `CreateTableEvent` encounters an existing target table. 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.
+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.
-The default is `false`. Disabling this option does not disable the sink's own schema handling. The framework only derives DDL types that are enabled by `include.schema.changes` and supported by the sink; unsupported, unsafe, or failed expansions are delegated to the sink without introducing a framework fail-fast.
+| Mode | Behavior on an existing target table | Behavior when the target table is missing | Failure handling |
+|---|---|---|---|
+| `OFF` | No check or expansion; the sink's original behavior applies | Sink creates the table | N/A |
+| `CHECK` | Validate that every upstream column can be contained by the target table, without issuing any DDL | Fails the job; the table must be created externally | Any incompatibility, read failure, or missing capability fails the job with an aggregated error |
+| `TRY_EXPAND` | Check and best-effort apply safe DDL, then verify the result by reading the target schema back | Sink creates the table | Failures of this mechanism are logged and delegated to the sink's original behavior |
+| `EXPAND` | Check and apply safe DDL, then verify the result by reading the target schema back | Sink creates the table | Any incompatibility, unsupported DDL, execution or verification failure fails the job |
+
+`CHECK` never issues DDL, so it is independent of `include.schema.changes` and of the sink's DDL capabilities. It guards the initial table state and runs **regardless of `schema.change.behavior`** (including `IGNORE` and `EXCEPTION`); `TRY_EXPAND` and `EXPAND` skip the framework-side initial handling when `schema.change.behavior` is `IGNORE` or `EXCEPTION`. Note that `CHECK` only constrains the initial table handling: subsequent source schema changes are still controlled by `schema.change.behavior`, so it is not a job-wide "never issue DDL" switch. When the check fails, the aggregated error lists every difference (table, column, upstream type vs. target type) together with suggested `ALTER TABLE` repair statements that can be reviewed and adjusted to the target system's dialect.
+
+`TRY_EXPAND` swallows failures of this mechanism only; it neither hides errors from the sink's own schema handling nor guarantees that all upstream columns end up in the target table after a failed expansion.
```yaml
sink:
type: paimon
- existing-table.schema-expansion.enabled: true
+ existing-table.schema-expansion.mode: "EXPAND"
```
+> Note: `existing-table.schema-expansion.enabled` is no longer supported. Use `existing-table.schema-expansion.mode` with one of `OFF`, `CHECK`, `TRY_EXPAND`, `EXPAND` instead; the previous `enabled: true` maps to `TRY_EXPAND`. Quote the mode value to avoid the bare `OFF` scalar being parsed as a YAML boolean.
Review Comment:
This note is unnecessary as existing-table.schema-expansion.enabled was never exposed to user.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4540:
URL: https://github.com/apache/flink-cdc/pull/4540#discussion_r4060674008
##########
docs/content.zh/docs/core-concept/schema-evolution.md:
##########
@@ -75,6 +75,29 @@ pipeline:
在此模式下，所有架构更改事件都将被 `SchemaOperator` 默默接收，并且永远不会尝试将它们应用于下游接收器。
当您的下游接收器尚未准备好进行任何架构更改，但想要继续从未更改的列中接收数据时，这很有用。
+## 已有目标表的安全 Schema 扩展
+
+设置 Sink 选项 `existing-table.schema-expansion.mode` 来控制框架在初始 `CreateTableEvent` 遇到已有目标表时的处理方式，默认值为 `OFF`。对于实现了该能力的 Sink，框架可能将缺失的普通非键物理列按 nullable 补充，并安全拓宽普通非键列类型；派生的 DDL 事件会记录在日志中。
+
+| 模式 | 已有目标表 | 目标表不存在 | 失败处理 |
+|---|---|---|---|
+| `OFF` | 不检查、不扩展，保持 Sink 原行为 | Sink 原生建表 | 不适用 |
Review Comment:
`OFF` looks like a value of boolean config, could you introduce a new value ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org