[FLINK-40647][runtime] Support best-effort schema expansion for existing sink tables during table creation

[haruki-830]

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

[lvyanquan]

Thank you for your contribution. I’ve left a few comments.

[lvyanquan]

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.

[lvyanquan]

LGTM.

[loserwang1024]

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.:

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.

[loserwang1024]

LGTM now

[leonardBang]

Thanks for this PR — the overall design (especially the CHECK mode and the safe-widening type rules) is well thought out, and the E2E coverage across regular/batch/distributed topologies is solid. I have a few suggestions focused on configuration clarity and the TRY_EXPAND failure boundary. Leaving them as inline comments for discussion.

[leonardBang]

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.

[lvyanquan]

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.

[leonardBang]

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.

[leonardBang]

Thanks @haruki-830 and @lvyanquan for the contribution, +1 from my side.

I'll merge once the CI green.