[bube]

Purpose

Fixes #10107.

sink.operator-uid.suffix and source.operator-uid.suffix reach only Writer, Global Committer, dynamic-bucket-assigner and Source. Every other operator the Paimon table sink or source adds takes its id from the stream graph shape, so any topology change orphans its checkpoint entry, and a restore from the HA checkpoint store fails with There is no operator for the state <id>. Discussed on dev@: [DISCUSS] The operator-uid.suffix options do not cover every operator.

This adds sink.operator-uid.cover-table-sink-operators and source.operator-uid.cover-table-source-operators, boolean, default false. With the option and the matching suffix set, every streaming operator the table sink or source adds gets <prefix>_<table>_<suffix>. Existing uids and the default graph are unchanged. The suffix options' descriptions now point at the new keys. Batch-only operators are out of scope: they never restore from a checkpoint.

Things a reviewer should know:

• The CDC sync actions are not covered. Their Side Output, Schema Evolution and Case-insensitive Convert operators and the shared operators of a database sync keep generated ids, so such jobs still cannot run with pipeline.auto-generate-uids=false. The option names and the guide say so. Covering them would need a uid shape for table-less operators and, in combined mode, a migration that drops Writer and Committer state; any later CDC coverage would go under a new key, so nobody who enabled these options sees ids move on upgrade. The second commit is this scoping and the rename (review round 1).
• Enabling the option orphans the newly named operators' old entries once. All are empty except Collect Statistics (a CoordinatedOperatorFactory, present under partition.sink-strategy = PARTITION_DYNAMIC) and the split monitor on the exactly-once consumer route, whose read position the consumer keeps across flink stop. On the explicit-path route those two need execution.state-recovery.ignore-unclaimed-state once; a recovery from the HA checkpoint store cannot skip entries at all, so the migration goes through a savepoint or an explicit checkpoint path. The guide in docs/docs/flink/savepoint.md ("Operator UIDs") spells this out and shows that writer and committer state stays mapped and the read position is kept.
• The final end sink is covered. The HA route rejects any unclaimed entry regardless of content, end acknowledges checkpoints and so has one, and SQL users have no way to uid it themselves. It only changes with the option on, and a DataStream user's own .uid() on the returned sink still wins. A SQL-level test shows the planner keeps this uid in every table.exec.uid.generation mode.
• Same-table-name collision. db1.orders and db2.orders written with one suffix collide on Writer_orders_s. This is pre-existing for the four old uids and now applies to about twenty; the guide tells users to use a distinct suffix per table.
• Uid prefixes equal the operator display names (spaces included) and live in *_NAME constants, following WRITER_NAME, GLOBAL_COMMITTER_NAME and SOURCE_NAME. They are pinned literally in tests.

Tests

• OperatorUidSuffixTest (19) and SourceOperatorUidSuffixTest (12): 12 sink and 7 source topology shapes each built under pipeline.auto-generate-uids=false, so Flink itself rejects any uid-less operator; exact node-name guards per shape; literal pins for the four pre-existing uids and for every new one; option off and option-without-suffix leave today's uids only; operator ids survive an upstream topology change.
• OperatorUidSuffixRestoreITCase (2): restore from a retained checkpoint store (HA route, EmbeddedHaServicesWithLeadershipControl + PerJobCheckpointRecoveryFactory) after adding an upstream operator succeeds with the option on and fails with There is no operator for the state with it off.
• OperatorUidMigrationITCase (2): checkpoint taken without the option, restored with it through execution.state-recovery.path. Unaware-bucket succeeds. PARTITION_DYNAMIC fails with Cannot map checkpoint/savepoint state for operator, then succeeds with execution.state-recovery.ignore-unclaimed-state; the restored job commits under the same commitUser (committer state mapped) and the dropped entries, read from the checkpoint metadata, are exactly the previously uid-less operators.
• MonitorSourceUidMigrationITCase (1): a streaming read with an exactly-once consumer, stopped with a savepoint and restored with the source cover option on, reads every row exactly once (Flink 1.20 and 2.2.0).
• OperatorUidSuffixTest.testSqlInsertKeepsTheUids (2): a SQL INSERT keeps the Writer, Global Committer and end uids under table.exec.uid.generation PLAN_ONLY and ALWAYS.
• The guide's migration steps, run by hand on a Flink 1.20.5 cluster: append table, PARTITION_DYNAMIC, recovery from the HA checkpoint store (ZooKeeper, before and after the migration) and an exactly-once consumer read. No lost or duplicated rows; the log lines are the ones the guide quotes.
• After rebasing onto ed8ecdc231 with the rename: the five uid classes (36) on Flink 1.20, the two new tests and OperatorUidSuffixTest on Flink 2.2.0, the docs site build, ConfigOptionsDocsCompletenessITCase, -DskipTests verify and the paimon-flink-cdc install. Before the rebase: full paimon-flink-common unit phase on Flink 1.20.1 (1094 tests), the four classes above on Flink 2.2.0 (-Pflink2), paimon-flink-cdc compile, ConfigOptionsDocsCompletenessITCase, and -DskipTests verify (checkstyle, spotless, rat, enforcer). Not built: the paimon-flink-1.16..2.1 version modules and paimon-flink-action.

🤖 Generated with Claude Code

[JingsongLi]

Requirement fit: SUPPORTED. Stable Paimon operator IDs address the documented HA checkpoint restore failure after upstream topology changes. Implementation: FINDINGS: the new cover-all option still leaves the public streaming CDC route partially unnamed. The graph below was built through CdcSinkBuilder with the actual option enabled, not a synthetic helper. Existing IDs and default behavior were also checked. Validation: 29 graph cases and 2 HA restore cases pass in the formal Flink 1 Maven run. Both explicit-path migration cases pass through a freshly built production Flink 1.20 bundle (including actual subsequent commits under the restored commitUser); the first flat-classpath run failed before migration, so it is not counted as passing. The normal production package build and current CI pass.

[JingsongLi]

[P2] Include the streaming CDC operators in cover-all coverage

The new option promises to name every operator a streaming write adds, but the public CdcSinkBuilder route still creates Side Output and Schema Evolution without a UID. On a fixed-bucket PK table with sink.operator-uid.suffix=stable and cover-all=true, an actual CDC StreamGraph has UID=null for both. Adding one explicitly uid-named map upstream changes their operator hashes (Side Output 5f5dbcb6... -> f2512467..., Schema Evolution bf7a1542... -> 75147ea0...). Building the same graph under pipeline.auto-generate-uids=false fails with no UID assigned to operator Side Output. Thus users enabling the new option for streaming CDC remain exposed to the same orphaned HA checkpoint entries it is intended to prevent. Apply the new assigner to the CDC parser/schema-evolution/conversion route and add a CDC topology case; if CDC is intentionally excluded, explicitly scope the option/documentation instead of guaranteeing every streaming write operator.

[bube]

Thanks for the review. I went with your second option and scoped the options instead of covering CDC.

Covering only the single-table CdcSinkBuilder route would still leave the option's promise false. --table_conf copies the keys into every table of a database sync, so users can turn it on there too. Covering the database sync needs a uid shape for operators that belong to no single table: the shared Side Output, the keyed Schema Evolution, the multiplexed Convert and, in combined mode, the Writer and Global Committer of FlinkCdcMultiTableSink. Those last two hold state and have no uid today. Turning the option on would drop their state on the first restore, which is the kind of migration you asked us not to recommend. I don't think that belongs in this PR.

So:

• The options are now sink.operator-uid.cover-table-sink-operators and source.operator-uid.cover-table-source-operators (c815cc4). Their descriptions and the guide say the CDC sync actions and BATCH-only operators are not covered.
• The *.operator-uid.suffix descriptions now say the suffix alone does not keep ids stable, and point at the new options. All four link the migration guide, which is also shorter (781072b).

Two things I checked on the way (e2797de):

• Source state. The only source entry the migration drops is the split monitor's, on the exactly-once consumer route. That route always has a consumer ID, and after flink stop the consumer holds the same next snapshot as the monitor, so the restored reader neither skips nor re-reads anything. A new ITCase shows this on Flink 1.20 and 2.2. A bounded read with dedicated split generation would re-read its scan, so the guide says to let such a job finish first.
• The end sink for SQL users. For a DataStreamSinkProvider the planner leaves Paimon's uids alone in every table.exec.uid.generation mode. A SQL-level test pins this.

If CDC coverage is wanted later, it would go under a new key, so nobody who turned these options on sees ids move on upgrade.

I also rebased onto current master. Does the new name work for you?

[bube]

Correction to my last comment: the bounded dedicated-split route does not re-read because of the migration. It re-plans its scan on every restore, and the Paimon sink drops the repeated end-of-input commit. Its monitor entry does need allowNonRestoredState, like the exactly-once route. I removed the sentence from the guide (e7aee65).

I also ran the guide's steps on a real Flink 1.20.5 cluster for the append, PARTITION_DYNAMIC, HA-store recovery and exactly-once consumer routes: no lost or duplicated rows, and the log lines match the guide.