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

[lvyanquan]

This note is unnecessary as existing-table.schema-expansion.enabled was never exposed to user.