[flink] Add operator-uid cover options to name every table sink and source 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,92 @@ 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 maps checkpoint state to operators by id. An operator without a UID gets an id from its
+place in the job graph, so adding an operator upstream, or a parallelism change that alters
+chaining, gives it a new id and orphans its state.
+
+A Paimon source or sink is several operators. The suffix options name some of them; the cover
+options name the rest:
+
+| Option | Operators named |
+| --- | --- |
+| `sink.operator-uid.suffix` | Writer, Global Committer, dynamic-bucket-assigner |
+| `sink.operator-uid.cover-table-sink-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-table-source-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 leaves out the database, so two
+tables with the same name and suffix in one job collide, and Flink rejects the job with
+`Hash collision on user-specified ID`. Give each table its own suffix.
+
+The cover options do nothing without the matching suffix. Set the sink pair before the first
+streaming job writes the table, and the source pair before the first streaming job reads it:
+
+```sql
+ALTER TABLE my_table SET (
+    'sink.operator-uid.suffix' = 'my_table_v1',
+    'sink.operator-uid.cover-table-sink-operators' = 'true',
+    'source.operator-uid.suffix' = 'my_table_v1',
+    'source.operator-uid.cover-table-source-operators' = 'true'
+);
+```
+
+Not covered:
+
+- The CDC sync actions: the operators they add around the table sink, such as Side Output,
+  Schema Evolution and Case-insensitive Convert, and the shared operators of a database sync.
+  Such jobs cannot run with `pipeline.auto-generate-uids=false`.
+- Operators only a `BATCH` job builds, such as sort and clustering. Flink does not checkpoint in
+  `BATCH` mode. A bounded job in `STREAMING` mode is covered.
+
+### Migrate a running job
+
+Turning on a cover option gives the newly covered operators new ids, so their entries in the
+job's checkpoints become unclaimed. A restore from a savepoint or from
+`execution.state-recovery.path` skips an unclaimed entry that holds no state and rejects one
+that does. A restore from the high-availability checkpoint store, as in a JobManager failover or
+a Kubernetes Operator `last-state` upgrade, rejects every unclaimed entry, and neither
+`allowNonRestoredState` nor `execution.state-recovery.ignore-unclaimed-state` helps:
+
+```
+JobInitializationException: Could not start the JobMaster.
+Caused by: IllegalStateException: There is no operator for the state <operator id>
+```
+
+So migrate through a savepoint:
+
+1. **Stop the job** with
+   [stop-with-savepoint](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#stopping-a-job-with-savepoint)
+   (`flink stop`) and wait for it to finish.
+
+2. **Set the options** with `ALTER TABLE ... SET` or SQL hints. Keep the existing suffix:
+   changing it renames the Writer and Global Committer too, and loses their state.
+3. **Restart from the savepoint** with the same topology. The operators the sink options newly
+   cover hold no state, so Flink skips their entries without `allowNonRestoredState` and logs
+   `Skipping empty savepoint state for operator` once per operator. Check that the snapshot ids
+   in `<table>$snapshots` continue past the one the job stopped at, under the same `commit_user`.
+
+   Two cases drop an entry that holds state. Without `allowNonRestoredState` the restore fails
+   with `Cannot map checkpoint/savepoint state for operator <id>`, so submit this one restart
+   with it. Flink then logs `Skipping savepoint state for operator <id>` for each entry it skips,
+   empty ones included. The Writer and Global Committer keep their UIDs, so Flink still restores
+   their state.
+
+   - **`'partition.sink-strategy' = 'PARTITION_DYNAMIC'`**: the `Collect Statistics` coordinator's
+     entry is dropped. It rebuilds its statistics from the next records.
+   - **A read with dedicated split generation or an exactly-once consumer**: the split monitor's
+     entry is dropped. An exactly-once read needs a [consumer ID](./consumer-id) with the default
+     `consumer.mode`. After `flink stop` the consumer holds the same next snapshot as the
+     monitor, and the reader resumes from it, so no snapshot is skipped or read twice. This holds
+     only if you stop with `flink stop` (`flink savepoint` then cancel can re-read snapshots),
+     restart before `consumer.expiration-time`, and keep `consumer.ignore-progress` false. Check
+     that `next_snapshot_id` in `<table>$consumers` holds, then advances.
+
+4. **Do not restore from savepoints or checkpoints taken before the migration.** They still hold
+   the old ids.
+
+After the migration, topology changes around the table, such as a map in front of the sink or a
+new parallelism, keep every Paimon operator's state on the routes above.

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,11 +302,17 @@
             <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-table-sink-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 the table sink adds in a streaming job: the row conversions, the local merge, the compaction operators, the partition statistics operators, the index bootstrap and the final sink. Operators the CDC sync actions add before the table sink are not covered. Without it those operators get ids generated 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. Enabling it on an existing job changes those ids at the next restore; follow the <a href="../flink/savepoint#operator-uids">migration steps</a> first.</td>
+        </tr>
         <tr>
             <td><h5>sink.operator-uid.suffix</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is ${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.</td>
+            <td>Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink generates these ids from the shape of the stream graph, so a topology change can orphan their state. The suffix alone does not prevent that: the other operators the table sink adds keep generated ids, and a recovery from the HA checkpoint store fails when one of them moves. For stable ids also set 'sink.operator-uid.cover-table-sink-operators'; existing jobs must follow the <a href="../flink/savepoint#operator-uids">migration steps</a> first.</td>
         </tr>
         <tr>
             <td><h5>sink.parallelism</h5></td>
@@ -398,11 +404,17 @@
             <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-table-source-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 the table source adds in a streaming job: the split monitor, the split reader, the watermark assigner and the DataStream row conversion. Without it those operators get ids generated 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. Enabling it on an existing job changes those ids at the next restore; follow the <a href="../flink/savepoint#operator-uids">migration steps</a> first.</td>
+        </tr>
         <tr>
             <td><h5>source.operator-uid.suffix</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>Set the uid suffix for the source operators. After setting, the uid format is ${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.</td>
+            <td>Set the uid suffix for the source operators. The uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink generates these ids from the shape of the stream graph, so a topology change can orphan their state. The suffix alone does not prevent that: the other operators the table source adds keep generated ids, and a recovery from the HA checkpoint store fails when one of them moves. For stable ids also set 'source.operator-uid.cover-table-source-operators'; existing jobs must follow the <a href="../flink/savepoint#operator-uids">migration steps</a> first.</td>
         </tr>
         <tr>
             <td><h5>unaware-bucket.compaction.parallelism</h5></td>

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

@@ -25,6 +25,7 @@
 import org.apache.paimon.options.ConfigOptions;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.description.DescribedEnum;
+import org.apache.paimon.options.description.Description;
 import org.apache.paimon.options.description.InlineElement;
 
 import java.lang.reflect.Field;
@@ -33,11 +34,15 @@
 import java.util.List;
 
 import static org.apache.paimon.options.ConfigOptions.key;
+import static org.apache.paimon.options.description.LinkElement.link;
 import static org.apache.paimon.options.description.TextElement.text;
 
 /** Options for flink connector. */
 public class FlinkConnectorOptions {
 
+    /** Relative to the configurations page that embeds the generated option tables. */
+    private static final String OPERATOR_UID_GUIDE = "../flink/savepoint#operator-uids";
+
     public static final String NONE = "none";
 
     public static final String TABLE_DYNAMIC_OPTION_PREFIX = "paimon.";
@@ -481,18 +486,64 @@ public class FlinkConnectorOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Set the uid suffix for the source operators. After setting, the uid format is "
-                                    + "${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.");
+                            Description.builder()
+                                    .text(
+                                            "Set the uid suffix for the source operators. The uid format is "
+                                                    + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink "
+                                                    + "generates these ids from the shape of the stream graph, so a topology "
+                                                    + "change can orphan their state. The suffix alone does not prevent that: "
+                                                    + "the other operators the table source adds keep generated ids, and a "
+                                                    + "recovery from the HA checkpoint store fails when one of them moves. "
+                                                    + "For stable ids also set 'source.operator-uid.cover-table-source-operators'; existing jobs must follow the %s first.",
+                                            link(OPERATOR_UID_GUIDE, "migration steps"))
+                                    .build());
+
+    public static final ConfigOption<Boolean> SOURCE_OPERATOR_UID_COVER_TABLE_SOURCE_OPERATORS =
+            key("source.operator-uid.cover-table-source-operators")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            Description.builder()
+                                    .text(
+                                            "If true, 'source.operator-uid.suffix' also names every other operator the table source adds in a streaming job: the split monitor, the split reader, the watermark assigner and the DataStream row conversion. "
+                                                    + "Without it those operators get ids generated 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. Enabling it on an existing "
+                                                    + "job changes those ids at the next restore; follow the %s first.",
+                                            link(OPERATOR_UID_GUIDE, "migration steps"))
+                                    .build());
 
     public static final ConfigOption<String> SINK_OPERATOR_UID_SUFFIX =
             key("sink.operator-uid.suffix")
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is "
-                                    + "${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.");
+                            Description.builder()
+                                    .text(
+                                            "Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is "
+                                                    + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink "
+                                                    + "generates these ids from the shape of the stream graph, so a topology "
+                                                    + "change can orphan their state. The suffix alone does not prevent that: "
+                                                    + "the other operators the table sink adds keep generated ids, and a "
+                                                    + "recovery from the HA checkpoint store fails when one of them moves. "
+                                                    + "For stable ids also set 'sink.operator-uid.cover-table-sink-operators'; existing jobs must follow the %s first.",
+                                            link(OPERATOR_UID_GUIDE, "migration steps"))
+                                    .build());
+
+    public static final ConfigOption<Boolean> SINK_OPERATOR_UID_COVER_TABLE_SINK_OPERATORS =
+            key("sink.operator-uid.cover-table-sink-operators")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            Description.builder()
+                                    .text(
+                                            "If true, 'sink.operator-uid.suffix' also names every other operator the table sink adds in a streaming job: the row conversions, the local merge, the compaction operators, the partition statistics operators, the index bootstrap and the final sink. Operators the CDC sync actions add before the table sink are not covered. "
+                                                    + "Without it those operators get ids generated 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. Enabling it on an existing "
+                                                    + "job changes those ids at the next restore; follow the %s first.",
+                                            link(OPERATOR_UID_GUIDE, "migration steps"))
+                                    .build());
 
     public static final ConfigOption<Boolean> SCAN_BOUNDED =
             key("scan.bounded")