[flink] Add operator-uid.cover-all-operators to name every streaming operator

docs/docs/flink/savepoint.md

@@ -33,6 +33,7 @@ or undo commits made after it.
 | Stop a writer and resume from its stopping point | [Stop with savepoint](#stop-with-savepoint). |
 | Retain a recovery point while the job continues writing | [Create a tag with the savepoint](#tag-with-savepoint), then roll the table back before restoring. |
 | Restart a streaming reader using stored table progress | See [Consumer ID](./consumer-id). |
+| Change the job around a Paimon table and keep its state | [Give every operator a stable UID](#operator-uids) before the change. |
 
 ![Savepoint and Paimon tag preserve matching job and table states; recovery stops writers, rolls back the table, then restores the job.](/img/flink-savepoint-recovery.svg)
 
@@ -84,3 +85,111 @@ chosen recovery point before executing it. Coordinate this for each Paimon table
 
 [Resume the job from the matching savepoint](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#resuming-from-savepoints).
 Verify that checkpoints complete and new Paimon snapshots appear before resuming dependent work.
+
+## Operator UIDs
+
+Flink stores state per operator and finds it again by operator id, which comes from the UID when
+the operator has one. An operator without one gets an id derived from its position in the job
+graph. Adding an operator upstream, or changing a parallelism that alters operator chaining, then
+gives it a new id and orphans its checkpoint entry.
+
+A Paimon source or sink is several operators. The suffix options name some of them; the
+cover-all options name the rest:
+
+| Option | Operators named |
+| --- | --- |
+| `sink.operator-uid.suffix` | Writer, Global Committer, dynamic-bucket-assigner |
+| `sink.operator-uid.cover-all-operators` | Also the row conversions, `local merge`, the compaction operators, `Collect Statistics` and `Strip Statistics`, `INDEX_BOOTSTRAP` and `cross-partition-bucket-assigner`, and the final `end` sink |
+| `source.operator-uid.suffix` | The source |
+| `source.operator-uid.cover-all-operators` | Also the split monitor and reader used by dedicated split generation and exactly-once consumers, the watermark assigner, and the DataStream row conversion |
+
+Each UID is `<operator>_<table name>_<suffix>`. The table name does not include the database, so
+two tables with the same name and the same suffix in one job collide. Flink then rejects the job
+with `Hash collision on user-specified ID`. Give each table its own suffix.
+
+The `cover-all-operators` options are off by default and do nothing without the matching suffix.
+Set the sink pair on a table before the first streaming job that writes it, and the source pair
+before the first streaming job that reads it. The example below sets all four, for a table used
+both ways:
+
+```sql
+ALTER TABLE my_table SET (
+    'sink.operator-uid.suffix' = 'my_table_v1',
+    'sink.operator-uid.cover-all-operators' = 'true',
+    'source.operator-uid.suffix' = 'my_table_v1',
+    'source.operator-uid.cover-all-operators' = 'true'
+);
+```
+
+Batch jobs are not covered. They do not restore from checkpoints today, and the options name only operators that a streaming job also builds.
+
+### How an orphaned entry fails
+
+Flink handles an entry that no operator claims in two different ways, and the way depends on how
+the job is restored:
+
+- **From a savepoint or an explicit checkpoint path** (`execution.state-recovery.path`), Flink
+  skips an unclaimed entry that holds no state and rejects one that does.
+- **From the high-availability checkpoint store**, which is what a JobManager failover and the
+  Flink Kubernetes Operator's `last-state` upgrade mode use, Flink rejects every unclaimed entry,
+  empty or not:
+
+  ```
+  JobInitializationException: Could not start the JobMaster.
+  Caused by: IllegalStateException: There is no operator for the state <operator id>
+  ```
+
+  The job then stays in a terminal state. The `allowNonRestoredState` setting of a job
+  submission does not apply on this route. Nor does `execution.state-recovery.ignore-unclaimed-state`;
+  Flink reads it only together with `execution.state-recovery.path`. A recovery from the HA
+  checkpoint store has no way to skip an entry. Restore from a savepoint or an explicit
+  checkpoint path instead.
+
+### Migrate a running job
+
+Turning `cover-all-operators` on changes the ids of the operators it newly covers, so their
+entries in the job's existing checkpoints become unclaimed. Migrate through a savepoint, never
+through a `last-state` upgrade or a JobManager failover.
+
+1. **Stop the job with a savepoint.** Use
+   [stop-with-savepoint](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#stopping-a-job-with-savepoint)
+   and wait for it to finish.
+
+2. **Set the options** on the table with `ALTER TABLE ... SET`, or in the job's SQL hints. Keep the
+   existing suffix. Changing it renames the Writer and Global Committer UIDs too, and their state
+   would be lost.
+
+3. **Restart from the savepoint.** With `'partition.sink-strategy' = 'PARTITION_DYNAMIC'`, or a
+   read with dedicated split generation or an exactly-once consumer, go to step 4 first. Otherwise
+   restart with the same topology as before. The newly covered sink operators keep nothing in a
+   checkpoint, so Flink skips their old entries and the restore succeeds without
+   `allowNonRestoredState`. The JobManager log shows one `Skipping empty savepoint state for
+   operator` line per operator that gained a UID. Confirm the job resumes committing from the
+   snapshot it stopped at.
+
+4. **Partition-dynamic append tables need one more step.** With
+   `'partition.sink-strategy' = 'PARTITION_DYNAMIC'` the `Collect Statistics` operator has an
+   operator coordinator, and a coordinator's entry is never empty. Restarting the normal way
+   then fails with `Cannot map checkpoint/savepoint state for operator <id>...`. For that one
+   restart, submit with `allowNonRestoredState`. The JobManager then logs `Skipping savepoint
+   state for operator <id>` once per dropped entry. The only dropped entry that held state is
+   the statistics coordinator, which rebuilds it from the records it sees next. The source,
+   Writer and Global Committer already carry UIDs and keep their ids, so Flink still finds
+   their state.
+
+5. **Reads with dedicated split generation or an exactly-once consumer hold their position in
+   the split monitor.** Only these read routes gain operators from
+   `source.operator-uid.cover-all-operators`. Every other read already carries its UID on the
+   source. A DataStream job that calls `buildForRow` gains the row conversion operator on every
+   route. The monitor's entry holds the next snapshot to read, so step 3 rejects it the same
+   way. Restart with `allowNonRestoredState`, and one of the dropped entries is the monitor's.
+   With a [consumer ID](./consumer-id) the reader resumes from the snapshot recorded in
+   the consumer and nothing is lost. Without one it starts again from the configured scan
+   mode, so plan for replayed or skipped snapshots, or set the source option on a fresh job
+   instead of migrating.
+
+6. **Take a new savepoint** once the job checkpoints again. Later restores, including
+   `last-state` upgrades, then start from the new UID layout.
+
+After the migration, changing the topology around the table, for example adding a map function
+in front of the sink or changing parallelism, keeps every Paimon operator's state.

docs/docs/flink/troubleshooting.md

@@ -124,3 +124,7 @@ not automatically undo later table commits. For a tagged recovery point, match t
 
 Also check whether source topology or consumer mode changed. Dedicated split generation and
 switching consumer modes can make the existing Flink state incompatible.
+
+If the JobManager fails with `There is no operator for the state <id>` after a change elsewhere in
+the job, a Paimon operator without a UID has taken a new id. Set the `operator-uid` options and
+follow the [migration steps](./savepoint#migrate-a-running-job).

docs/generated/flink_connector_configuration.html

@@ -302,6 +302,12 @@
             <td>MemorySize</td>
             <td>Weight of writer buffer in managed memory, Flink will compute the memory size for writer according to the weight, the actual memory used depends on the running environment.</td>
         </tr>
+        <tr>
+            <td><h5>sink.operator-uid.cover-all-operators</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, 'sink.operator-uid.suffix' also names every other operator a streaming write adds: the row conversions, the local merge, the compaction operators, the partition statistics operators, the index bootstrap and the final sink. Without it those operators take their uid from the shape of the stream graph, so a change elsewhere in the job orphans their checkpoint state. Has no effect unless 'sink.operator-uid.suffix' is set. Turning it on for a running job changes those uids, see the migration steps in the Flink savepoint documentation.</td>
+        </tr>
         <tr>
             <td><h5>sink.operator-uid.suffix</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
@@ -392,6 +398,12 @@
             <td>Duration</td>
             <td>If the new snapshot has not been generated when the checkpoint starts to trigger, the enumerator will block the checkpoint and wait for the new snapshot. Set the maximum waiting time to avoid infinite waiting, if timeout, the checkpoint will fail. Note that it should be set smaller than the checkpoint timeout.</td>
         </tr>
+        <tr>
+            <td><h5>source.operator-uid.cover-all-operators</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, 'source.operator-uid.suffix' also names every other operator a streaming read adds: the split monitor, the split reader, the watermark assigner and the DataStream row conversion. Without it those operators take their uid from the shape of the stream graph, so a change elsewhere in the job orphans their checkpoint state. Has no effect unless 'source.operator-uid.suffix' is set. Turning it on for a running job changes those uids, see the migration steps in the Flink savepoint documentation.</td>
+        </tr>
         <tr>
             <td><h5>source.operator-uid.suffix</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -474,6 +474,19 @@ public class FlinkConnectorOptions {
                                     + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will "
                                     + "automatically generate the operator uid, which may be incompatible when the topology changes.");
 
+    public static final ConfigOption<Boolean> SOURCE_OPERATOR_UID_COVER_ALL_OPERATORS =
+            key("source.operator-uid.cover-all-operators")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, 'source.operator-uid.suffix' also names every other operator a streaming "
+                                    + "read adds: the split monitor, the split reader, the watermark assigner and the "
+                                    + "DataStream row conversion. Without it those operators take their uid from the "
+                                    + "shape of the stream graph, so a change elsewhere in the job orphans their "
+                                    + "checkpoint state. Has no effect unless 'source.operator-uid.suffix' is set. "
+                                    + "Turning it on for a running job changes those uids, see the migration steps "
+                                    + "in the Flink savepoint documentation.");
+
     public static final ConfigOption<String> SINK_OPERATOR_UID_SUFFIX =
             key("sink.operator-uid.suffix")
                     .stringType()
@@ -483,6 +496,20 @@ public class FlinkConnectorOptions {
                                     + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will "
                                     + "automatically generate the operator uid, which may be incompatible when the topology changes.");
 
+    public static final ConfigOption<Boolean> SINK_OPERATOR_UID_COVER_ALL_OPERATORS =
+            key("sink.operator-uid.cover-all-operators")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, 'sink.operator-uid.suffix' also names every other operator a streaming "
+                                    + "write adds: the row conversions, the local merge, the compaction operators, "
+                                    + "the partition statistics operators, the index bootstrap and the final sink. "
+                                    + "Without it those operators take their uid from the shape of the stream graph, "
+                                    + "so a change elsewhere in the job orphans their checkpoint state. Has no effect "
+                                    + "unless 'sink.operator-uid.suffix' is set. Turning it on for a running job "
+                                    + "changes those uids, see the migration steps in the Flink savepoint "
+                                    + "documentation.");
+
     public static final ConfigOption<Boolean> SCAN_BOUNDED =
             key("scan.bounded")
                     .booleanType()

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.flink.compact.AppendPreCommitCompactCoordinatorOperator;
 import org.apache.paimon.flink.compact.AppendPreCommitCompactWorkerOperator;
 import org.apache.paimon.flink.source.AppendBypassCoordinateOperatorFactory;
+import org.apache.paimon.flink.utils.OperatorUidAssigner;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
@@ -54,6 +55,12 @@ public abstract class AppendTableSink<T> extends FlinkWriteSink<T> {
 
     private static final long serialVersionUID = 1L;
 
+    private static final String NEW_FILES_COMPACT_COORDINATOR_NAME =
+            "New Files Compact Coordinator";
+    private static final String NEW_FILES_COMPACT_WORKER_NAME = "New Files Compact Worker";
+    private static final String COMPACT_COORDINATOR_NAME = "Compact Coordinator";
+    private static final String COMPACT_WORKER_NAME = "Compact Worker";
+
     protected final FileStoreTable table;
 
     @Nullable protected final Integer parallelism;
@@ -73,24 +80,29 @@ public DataStream<Committable> doWrite(
         DataStream<Committable> written = super.doWrite(input, initialCommitUser, this.parallelism);
 
         Options options = new Options(table.options());
+        OperatorUidAssigner uids = OperatorUidAssigner.forSink(table);
         if (options.get(FlinkConnectorOptions.PRECOMMIT_COMPACT)) {
             SingleOutputStreamOperator<Committable> newWritten =
-                    written.transform(
-                                    "New Files Compact Coordinator: " + table.name(),
-                                    new EitherTypeInfo<>(
-                                            new CommittableTypeInfo(),
-                                            new TupleTypeInfo<>(
-                                                    BasicTypeInfo.LONG_TYPE_INFO,
-                                                    new CompactionTaskTypeInfo())),
-                                    new AppendPreCommitCompactCoordinatorOperator(
-                                            table.coreOptions()))
-                            .startNewChain()
-                            .forceNonParallel()
+                    uids.assign(
+                                    written.transform(
+                                                    "New Files Compact Coordinator: "
+                                                            + table.name(),
+                                                    new EitherTypeInfo<>(
+                                                            new CommittableTypeInfo(),
+                                                            new TupleTypeInfo<>(
+                                                                    BasicTypeInfo.LONG_TYPE_INFO,
+                                                                    new CompactionTaskTypeInfo())),
+                                                    new AppendPreCommitCompactCoordinatorOperator(
+                                                            table.coreOptions()))
+                                            .startNewChain()
+                                            .forceNonParallel(),
+                                    NEW_FILES_COMPACT_COORDINATOR_NAME)
                             .transform(
                                     "New Files Compact Worker: " + table.name(),
                                     new CommittableTypeInfo(),
                                     new AppendPreCommitCompactWorkerOperator(table))
                             .startNewChain();
+            uids.assign(newWritten, NEW_FILES_COMPACT_WORKER_NAME);
             forwardParallelism(newWritten, written);
             written = newWritten;
         }
@@ -108,20 +120,24 @@ public DataStream<Committable> doWrite(
         // if enable compaction, we need to add compaction topology to this job
         if (enableCompaction && isStreamingMode) {
             SingleOutputStreamOperator<Committable> newWritten =
-                    written.transform(
-                                    "Compact Coordinator: " + table.name(),
-                                    new EitherTypeInfo<>(
-                                            new CommittableTypeInfo(),
-                                            new CompactionTaskTypeInfo()),
-                                    new AppendBypassCoordinateOperatorFactory<>(table))
-                            .startNewChain()
-                            .forceNonParallel()
+                    uids.assign(
+                                    written.transform(
+                                                    "Compact Coordinator: " + table.name(),
+                                                    new EitherTypeInfo<>(
+                                                            new CommittableTypeInfo(),
+                                                            new CompactionTaskTypeInfo()),
+                                                    new AppendBypassCoordinateOperatorFactory<>(
+                                                            table))
+                                            .startNewChain()
+                                            .forceNonParallel(),
+                                    COMPACT_COORDINATOR_NAME)
                             .transform(
                                     "Compact Worker: " + table.name(),
                                     new CommittableTypeInfo(),
                                     new AppendBypassCompactWorkerOperator.Factory(
                                             table, initialCommitUser, true))
                             .startNewChain();
+            uids.assign(newWritten, COMPACT_WORKER_NAME);
             setParallelism(newWritten, written.getParallelism(), false);
             written = newWritten;