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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
