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

[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 a streaming Paimon read or write 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-all-operators and source.operator-uid.cover-all-operators, boolean, default false. With the option and the matching suffix set, every streaming operator gets <prefix>_<table>_<suffix>. Existing uids and the default graph are unchanged. Batch-only operators are out of scope: they never restore from a checkpoint.

Things a reviewer should know:

• 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 Monitor source on the dedicated-split-generation and exactly-once-consumer routes. 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 source, writer and committer state stays mapped.
• 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. Open for discussion, as is the option name.
• 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 (17) 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.
• 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