From 36c8ed4c718a878414ea654759104b24f1789d03 Mon Sep 17 00:00:00 2001 From: Andreas Bube Date: Tue, 22 Sep 2026 10:05:28 +0200 Subject: [PATCH] [flink] Add operator-uid.cover-all-operators to name every streaming operator sink.operator-uid.suffix reaches the writer, the global committer and the dynamic bucket assigner; source.operator-uid.suffix reaches the source. Every other operator Paimon adds takes its id from the shape of the stream graph, so a change elsewhere in the job orphans its checkpoint entry. Recovery from the HA checkpoint store then fails with "There is no operator for the state ". Add sink.operator-uid.cover-all-operators and source.operator-uid.cover-all-operators, off by default. With the option and the matching suffix set, every operator a streaming read or write adds gets ${prefix}_${table}_${suffix}. Existing uids are unchanged. Batch-only topologies (clustering, sort compaction, postpone merge-on-read) never restore from a checkpoint and are left alone. Tests: a sink and a source graph matrix built under pipeline.auto-generate-uids=false, so Flink itself rejects any uid-less operator per topology shape; literal pins for every pre-existing and new uid; an ITCase restoring from a retained checkpoint store after an upstream topology change; an ITCase migrating from the old uid layout through execution.state-recovery.path, including PARTITION_DYNAMIC. Docs: a migration guide under docs/docs/flink/savepoint.md. Co-Authored-By: Claude Fable 5.1 --- docs/docs/flink/savepoint.md | 109 +++++ docs/docs/flink/troubleshooting.md | 4 + .../flink_connector_configuration.html | 12 + .../paimon/flink/FlinkConnectorOptions.java | 27 ++ .../paimon/flink/sink/AppendTableSink.java | 54 ++- .../apache/paimon/flink/sink/FlinkSink.java | 58 ++- .../paimon/flink/sink/FlinkSinkBuilder.java | 71 +-- .../sink/index/GlobalDynamicBucketSink.java | 38 +- .../flink/source/FlinkSourceBuilder.java | 12 +- .../flink/source/operator/MonitorSource.java | 39 +- .../flink/utils/OperatorUidAssigner.java | 99 ++++ .../paimon/flink/OperatorUidGraphs.java | 96 ++++ .../sink/OperatorUidMigrationITCase.java | 396 ++++++++++++++++ .../sink/OperatorUidSuffixRestoreITCase.java | 310 ++++++++++++ .../flink/sink/OperatorUidSuffixTest.java | 448 ++++++++++++++++++ .../source/SourceOperatorUidSuffixTest.java | 328 +++++++++++++ 16 files changed, 2006 insertions(+), 95 deletions(-) create mode 100644 paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/OperatorUidAssigner.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/OperatorUidGraphs.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidMigrationITCase.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixRestoreITCase.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java create mode 100644 paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SourceOperatorUidSuffixTest.java diff --git a/docs/docs/flink/savepoint.md b/docs/docs/flink/savepoint.md index 825636e99c81..5491fa52bd4e 100644 --- a/docs/docs/flink/savepoint.md +++ b/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 `__`. 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 + ``` + + 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 ...`. For that one + restart, submit with `allowNonRestoredState`. The JobManager then logs `Skipping savepoint + state for operator ` 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. diff --git a/docs/docs/flink/troubleshooting.md b/docs/docs/flink/troubleshooting.md index fbd68fa5151b..6b0b68b42939 100644 --- a/docs/docs/flink/troubleshooting.md +++ b/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 ` 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). diff --git a/docs/generated/flink_connector_configuration.html b/docs/generated/flink_connector_configuration.html index 37638d7f99a6..e07899494354 100644 --- a/docs/generated/flink_connector_configuration.html +++ b/docs/generated/flink_connector_configuration.html @@ -302,6 +302,12 @@ + + + + + + @@ -392,6 +398,12 @@ + + + + + + diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java index 7bb4f5c468c2..e4d8b4859d00 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java +++ b/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 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 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 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 SCAN_BOUNDED = key("scan.bounded") .booleanType() diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java index 6f33a5e45f02..d78a2d81350e 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendTableSink.java +++ b/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 extends FlinkWriteSink { 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 doWrite( DataStream 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 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 doWrite( // if enable compaction, we need to add compaction topology to this job if (enableCompaction && isStreamingMode) { SingleOutputStreamOperator 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; diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java index c28d2801288b..a1b78daac242 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java @@ -24,6 +24,7 @@ import org.apache.paimon.flink.compact.changelog.ChangelogCompactSortOperator; import org.apache.paimon.flink.compact.changelog.ChangelogCompactWorkerOperator; import org.apache.paimon.flink.compact.changelog.ChangelogTaskTypeInfo; +import org.apache.paimon.flink.utils.OperatorUidAssigner; import org.apache.paimon.manifest.ManifestCommittable; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; @@ -80,6 +81,12 @@ public abstract class FlinkSink implements Serializable { private static final String WRITER_WRITE_ONLY_NAME = "Writer(write-only)"; private static final String GLOBAL_COMMITTER_NAME = "Global Committer"; + private static final String CHANGELOG_COMPACT_COORDINATOR_NAME = + "Changelog Compact Coordinator"; + private static final String CHANGELOG_COMPACT_WORKER_NAME = "Changelog Compact Worker"; + private static final String CHANGELOG_SORT_NAME = "Changelog Sort by Creation Time"; + private static final String END_NAME = "end"; + protected final FileStoreTable table; private final boolean ignorePreviousFiles; @@ -170,26 +177,34 @@ public DataStream doWrite( written, options.get(SINK_WRITER_CPU), options.get(SINK_WRITER_MEMORY)); if (!table.primaryKeys().isEmpty() && options.get(PRECOMMIT_COMPACT)) { + OperatorUidAssigner uids = OperatorUidAssigner.forSink(table); SingleOutputStreamOperator beforeSort = - written.transform( - "Changelog Compact Coordinator", - new EitherTypeInfo<>( - new CommittableTypeInfo(), new ChangelogTaskTypeInfo()), - new ChangelogCompactCoordinateOperator(table.coreOptions())) - .forceNonParallel() + uids.assign( + written.transform( + CHANGELOG_COMPACT_COORDINATOR_NAME, + new EitherTypeInfo<>( + new CommittableTypeInfo(), + new ChangelogTaskTypeInfo()), + new ChangelogCompactCoordinateOperator( + table.coreOptions())) + .forceNonParallel(), + CHANGELOG_COMPACT_COORDINATOR_NAME) .transform( - "Changelog Compact Worker", + CHANGELOG_COMPACT_WORKER_NAME, new CommittableTypeInfo(), new ChangelogCompactWorkerOperator(table)); + uids.assign(beforeSort, CHANGELOG_COMPACT_WORKER_NAME); forwardParallelism(beforeSort, written); written = - beforeSort - .transform( - "Changelog Sort by Creation Time", - new CommittableTypeInfo(), - new ChangelogCompactSortOperator()) - .forceNonParallel(); + uids.assign( + beforeSort + .transform( + CHANGELOG_SORT_NAME, + new CommittableTypeInfo(), + new ChangelogCompactSortOperator()) + .forceNonParallel(), + CHANGELOG_SORT_NAME); } return written; @@ -233,9 +248,12 @@ private DataStreamSink doCoordinatorCommit( // The commit runs inside the writer's OperatorCoordinator on the JobManager, so there // is no global committer operator. Committables are still forwarded by the writer for // observability and are discarded here. - return written.sinkTo(new PaimonDiscardingSink<>(table)) - .name("end") - .setParallelism(written.getParallelism()); + return OperatorUidAssigner.forSink(table) + .assign( + written.sinkTo(new PaimonDiscardingSink<>(table)) + .name(END_NAME) + .setParallelism(written.getParallelism()), + END_NAME); } private DataStreamSink doOperatorCommit( @@ -288,7 +306,13 @@ private DataStreamSink doOperatorCommit( } configureSlotSharingGroup( committed, options.get(SINK_COMMITTER_CPU), options.get(SINK_COMMITTER_MEMORY)); - return committed.sinkTo(new PaimonDiscardingSink<>(table)).name("end").setParallelism(1); + return OperatorUidAssigner.forSink(table) + .assign( + committed + .sinkTo(new PaimonDiscardingSink<>(table)) + .name(END_NAME) + .setParallelism(1), + END_NAME); } public static void configureSlotSharingGroup( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java index 8bea0b5acfbe..304dc6a6b9a3 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java @@ -33,6 +33,7 @@ import org.apache.paimon.flink.sink.partition.StatisticsOrRecordTypeInfo; import org.apache.paimon.flink.sorter.TableSortInfo; import org.apache.paimon.flink.sorter.TableSorter; +import org.apache.paimon.flink.utils.OperatorUidAssigner; import org.apache.paimon.table.BlobDescriptorReaderFactory; import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; @@ -86,6 +87,12 @@ public class FlinkSinkBuilder { private static final Logger LOG = LoggerFactory.getLogger(FlinkSinkBuilder.class); + private static final String ROW_CONVERSION_NAME = "Row Conversion"; + private static final String INTERNAL_ROW_CONVERSION_NAME = "Internal Row Conversion"; + private static final String LOCAL_MERGE_NAME = "local merge"; + private static final String COLLECT_STATISTICS_NAME = "Collect Statistics"; + private static final String STRIP_STATISTICS_NAME = "Strip Statistics"; + protected final FileStoreTable table; private DataStream input; @@ -119,7 +126,7 @@ public FlinkSinkBuilder forRow(DataStream input, DataType rowDataType) { input.map((MapFunction) converter::toInternal) .returns(InternalTypeInfo.of(rowType)); setParallelism(newInput, input.getParallelism(), false); - this.input = newInput; + this.input = OperatorUidAssigner.forSink(table).assign(newInput, ROW_CONVERSION_NAME); return this; } @@ -226,23 +233,26 @@ public DataStreamSink build() { ? materializedBlobFieldIndexes( table.rowType(), table.coreOptions().blobInlineField()) : Collections.emptySet(); + OperatorUidAssigner uids = OperatorUidAssigner.forSink(table); DataStream input = - mapToInternalRowWithUriReaderFactory( - this.input, - table.rowType(), - readerFactoryForDescriptor, - table.coreOptions().blobWriteNullOnMissingFile(), - table.coreOptions().blobWriteNullOnFetchFailure(), - materializedBlobFields); + uids.assign( + mapToInternalRowWithUriReaderFactory( + this.input, + table.rowType(), + readerFactoryForDescriptor, + table.coreOptions().blobWriteNullOnMissingFile(), + table.coreOptions().blobWriteNullOnFetchFailure(), + materializedBlobFields), + INTERNAL_ROW_CONVERSION_NAME); if (table.coreOptions().localMergeEnabled() && table.schema().primaryKeys().size() > 0) { SingleOutputStreamOperator newInput = input.forward() .transform( - "local merge", + LOCAL_MERGE_NAME, input.getType(), new LocalMergeOperator.Factory(table.schema())); forwardParallelism(newInput, input); - input = newInput; + input = uids.assign(newInput, LOCAL_MERGE_NAME); } BucketMode bucketMode = table.bucketMode(); @@ -292,7 +302,7 @@ public static DataStream mapToInternalRow( Collections.emptySet()); } - private static DataStream mapToInternalRowWithUriReaderFactory( + private static SingleOutputStreamOperator mapToInternalRowWithUriReaderFactory( DataStream input, org.apache.paimon.types.RowType rowType, UriReaderFactory uriReaderFactory, @@ -423,14 +433,17 @@ private > T configureBlobDescriptorReaderFactory(T sink) } private DataStream applyDynamicPartitionShuffle(DataStream input) { + OperatorUidAssigner uids = OperatorUidAssigner.forSink(table); StatisticsOrRecordTypeInfo typeInfo = new StatisticsOrRecordTypeInfo(table.schema().logicalRowType()); SingleOutputStreamOperator statsStream = - input.transform( - "Collect Statistics: " + table.name(), - typeInfo, - new DataStatisticsOperatorFactory(table.schema())) - .setParallelism(input.getParallelism()); + uids.assign( + input.transform( + "Collect Statistics: " + table.name(), + typeInfo, + new DataStatisticsOperatorFactory(table.schema())) + .setParallelism(input.getParallelism()), + COLLECT_STATISTICS_NAME); DataStream partitioned = partition( @@ -438,18 +451,20 @@ private DataStream applyDynamicPartitionShuffle(DataStream) - (statisticsOrRecord, out) -> { - if (statisticsOrRecord.isRecord()) { - out.collect(statisticsOrRecord.record()); - } - }) - .name("Strip Statistics") - .setParallelism(parallelism != null ? parallelism : input.getParallelism()) - .returns(input.getType()); + return uids.assign( + partitioned + .flatMap( + (org.apache.flink.api.common.functions.FlatMapFunction< + StatisticsOrRecord, InternalRow>) + (statisticsOrRecord, out) -> { + if (statisticsOrRecord.isRecord()) { + out.collect(statisticsOrRecord.record()); + } + }) + .name(STRIP_STATISTICS_NAME) + .setParallelism(parallelism != null ? parallelism : input.getParallelism()) + .returns(input.getType()), + STRIP_STATISTICS_NAME); } private DataStream trySortInput(DataStream input) { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java index aaf05e78e026..a2099ba3729c 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java @@ -29,6 +29,7 @@ import org.apache.paimon.flink.sink.StoreSinkWrite; import org.apache.paimon.flink.utils.InternalRowTypeSerializer; import org.apache.paimon.flink.utils.InternalTypeInfo; +import org.apache.paimon.flink.utils.OperatorUidAssigner; import org.apache.paimon.options.Options; import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; @@ -60,6 +61,10 @@ public class GlobalDynamicBucketSink extends FlinkWriteSink overwritePartition) { super(table, overwritePartition); @@ -91,15 +96,18 @@ public DataStreamSink build(DataStream input, @Nullable Integer // input -- bootstrap -- shuffle by key hash --> bucket-assigner -- shuffle by bucket --> // writer --> committer + OperatorUidAssigner uids = OperatorUidAssigner.forSink(table); SingleOutputStreamOperator> bootstraped = - input.transform( - "INDEX_BOOTSTRAP", - new InternalTypeInfo<>( - new KeyWithRowSerializer<>( - bootstrapSerializer, rowSerializer)), - new IndexBootstrapOperator.Factory<>( - new IndexBootstrap(table), r -> r)) - .setParallelism(input.getParallelism()); + uids.assign( + input.transform( + INDEX_BOOTSTRAP_NAME, + new InternalTypeInfo<>( + new KeyWithRowSerializer<>( + bootstrapSerializer, rowSerializer)), + new IndexBootstrapOperator.Factory<>( + new IndexBootstrap(table), r -> r)) + .setParallelism(input.getParallelism()), + INDEX_BOOTSTRAP_NAME); // 1. shuffle by key hash Integer assignerParallelism = @@ -119,12 +127,14 @@ public DataStreamSink build(DataStream input, @Nullable Integer TupleTypeInfo> rowWithBucketType = new TupleTypeInfo<>(input.getType(), BasicTypeInfo.INT_TYPE_INFO); SingleOutputStreamOperator> bucketAssigned = - partitionByKeyHash - .transform( - "cross-partition-bucket-assigner", - rowWithBucketType, - GlobalIndexAssignerOperator.forRowData(table)) - .setParallelism(partitionByKeyHash.getParallelism()); + uids.assign( + partitionByKeyHash + .transform( + CROSS_PARTITION_BUCKET_ASSIGNER_NAME, + rowWithBucketType, + GlobalIndexAssignerOperator.forRowData(table)) + .setParallelism(partitionByKeyHash.getParallelism()), + CROSS_PARTITION_BUCKET_ASSIGNER_NAME); // declare managed memory for the local key-value index declareManagedMemory( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java index 4f8a10fd9f19..70b7db8ca2bb 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java @@ -26,6 +26,7 @@ import org.apache.paimon.flink.sink.FlinkSink; import org.apache.paimon.flink.source.align.AlignedContinuousFileStoreSource; import org.apache.paimon.flink.source.operator.MonitorSource; +import org.apache.paimon.flink.utils.OperatorUidAssigner; import org.apache.paimon.flink.utils.TableScanUtils; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.options.Options; @@ -84,6 +85,9 @@ public class FlinkSourceBuilder { private static final String SOURCE_NAME = "Source"; + private static final String ROW_CONVERSION_NAME = "Source Row Conversion"; + private static final String WATERMARKS_NAME = "Timestamps/Watermarks"; + private final Table table; private final Options conf; private final boolean unordered; @@ -415,7 +419,7 @@ public DataStream buildForRow() { source.map((MapFunction) converter::toExternal) .returns(ExternalTypeInfo.of(rowType)); forwardParallelism(result, source); - return result; + return OperatorUidAssigner.forSource(table).assign(result, ROW_CONVERSION_NAME); } /** Build source {@link DataStream} with {@link RowData}. */ @@ -494,7 +498,11 @@ private DataStream buildDedicatedSplitGenSource(boolean isBounded) { dataStream.getTransformation().setParallelism(parallelism); } if (watermarkStrategy != null) { - dataStream = dataStream.assignTimestampsAndWatermarks(watermarkStrategy); + dataStream = + OperatorUidAssigner.forSource(table) + .assign( + dataStream.assignTimestampsAndWatermarks(watermarkStrategy), + WATERMARKS_NAME); } return dataStream; } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java index 1b1ad2392fe1..0fd917dd1dcb 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java @@ -27,6 +27,7 @@ import org.apache.paimon.flink.source.SimpleSourceSplit; import org.apache.paimon.flink.source.SplitListState; import org.apache.paimon.flink.utils.JavaTypeInfo; +import org.apache.paimon.flink.utils.OperatorUidAssigner; import org.apache.paimon.options.Options; import org.apache.paimon.table.Table; import org.apache.paimon.table.sink.ChannelComputer; @@ -98,6 +99,9 @@ public class MonitorSource extends AbstractNonCoordinatedSource { private static final Logger LOG = LoggerFactory.getLogger(MonitorSource.class); + private static final String MONITOR_NAME = "Monitor"; + private static final String READER_NAME = "Reader"; + private final ReadBuilder readBuilder; private final long monitorInterval; private final boolean emitSnapshotWatermark; @@ -379,28 +383,33 @@ public static DataStream buildSource( if (table != null) { source = new PaimonDataStreamSource<>(monitorSource, table); } + OperatorUidAssigner uids = OperatorUidAssigner.forSource(table); SingleOutputStreamOperator operator = - env.fromSource( - source, - WatermarkStrategy.noWatermarks(), - name + "-Monitor", - new JavaTypeInfo<>(Split.class)) - .forceNonParallel(); + uids.assign( + env.fromSource( + source, + WatermarkStrategy.noWatermarks(), + name + "-Monitor", + new JavaTypeInfo<>(Split.class)) + .forceNonParallel(), + MONITOR_NAME); DataStream sourceDataStream = unordered ? shuffleUnordered(operator) : shuffleOrdered(operator, shuffleBucketWithPartition); - return sourceDataStream.transform( - name + "-Reader", - typeInfo, - new ReadOperator( - readBuilder::newRead, - nestedProjectedRowData, - limit, - readType, - blobAsDescriptor)); + return uids.assign( + sourceDataStream.transform( + name + "-Reader", + typeInfo, + new ReadOperator( + readBuilder::newRead, + nestedProjectedRowData, + limit, + readType, + blobAsDescriptor)), + READER_NAME); } private static DataStream shuffleUnordered( diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/OperatorUidAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/OperatorUidAssigner.java new file mode 100644 index 000000000000..eb66f6249a3e --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/OperatorUidAssigner.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.utils; + +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.Table; + +import org.apache.flink.streaming.api.datastream.DataStreamSink; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; + +import javax.annotation.Nullable; + +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.FlinkConnectorOptions.generateCustomUid; + +/** + * Assigns uids to the operators Paimon adds beyond the writer, the committer, the bucket assigner + * and the source, which carry one of their own. + * + *

An operator without a uid takes its id from the shape of the stream graph, so a change + * elsewhere in the job moves the id and orphans the checkpoint entry behind it. Assigning one is + * opt-in: the table has to set both {@code cover-all-operators} and the matching uid suffix. A job + * that sets neither builds the same graph as before. + * + *

A {@code *_NAME} constant passed here is a uid prefix, frozen once released. + */ +public class OperatorUidAssigner { + + private static final OperatorUidAssigner DISABLED = new OperatorUidAssigner(null, null); + + @Nullable private final String tableName; + + @Nullable private final String uidSuffix; + + private OperatorUidAssigner(@Nullable String tableName, @Nullable String uidSuffix) { + this.tableName = tableName; + this.uidSuffix = uidSuffix; + } + + public static OperatorUidAssigner forSink(@Nullable Table table) { + return create(table, SINK_OPERATOR_UID_COVER_ALL_OPERATORS, SINK_OPERATOR_UID_SUFFIX); + } + + public static OperatorUidAssigner forSource(@Nullable Table table) { + return create(table, SOURCE_OPERATOR_UID_COVER_ALL_OPERATORS, SOURCE_OPERATOR_UID_SUFFIX); + } + + private static OperatorUidAssigner create( + @Nullable Table table, + ConfigOption coverAllOperators, + ConfigOption suffix) { + if (table == null) { + return DISABLED; + } + Options options = Options.fromMap(table.options()); + String uidSuffix = options.get(suffix); + if (!options.get(coverAllOperators) || uidSuffix == null) { + return DISABLED; + } + return new OperatorUidAssigner(table.name(), uidSuffix); + } + + /** Assigns {@code ${uidPrefix}_${tableName}_${uidSuffix}} and returns the operator. */ + public SingleOutputStreamOperator assign( + SingleOutputStreamOperator operator, String uidPrefix) { + if (uidSuffix != null) { + operator.uid(generateCustomUid(uidPrefix, tableName, uidSuffix)); + } + return operator; + } + + /** The same for a terminal sink. */ + public DataStreamSink assign(DataStreamSink sink, String uidPrefix) { + if (uidSuffix != null) { + sink.uid(generateCustomUid(uidPrefix, tableName, uidSuffix)); + } + return sink; + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/OperatorUidGraphs.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/OperatorUidGraphs.java new file mode 100644 index 000000000000..6e315a454ffa --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/OperatorUidGraphs.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink; + +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamGraphHasherV2; +import org.apache.flink.streaming.api.graph.StreamNode; + +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.stream.Collectors; + +/** + * Reads uids and operator ids off a built {@link StreamGraph}, for the operator-uid coverage tests. + * + *

The tests uid everything they build themselves with {@link #TEST_UID_PREFIX}, so an operator + * without a uid, or with a uid lacking the prefix, can only have come from Paimon. + */ +public final class OperatorUidGraphs { + + public static final String TEST_UID_PREFIX = "test-"; + + private OperatorUidGraphs() {} + + /** Names of the operators without a uid. */ + public static List operatorsWithoutUid(StreamGraph graph) { + return graph.getStreamNodes().stream() + .filter(node -> node.getTransformationUID() == null) + .map(StreamNode::getOperatorName) + .sorted() + .collect(Collectors.toList()); + } + + /** Operator name to uid for every node, the uid being null where none was assigned. */ + public static Map uidsByOperatorName(StreamGraph graph) { + Map uids = new TreeMap<>(); + for (StreamNode node : graph.getStreamNodes()) { + uids.put(node.getOperatorName(), node.getTransformationUID()); + } + return uids; + } + + /** Operator name to uid for the nodes Paimon built and gave a uid. */ + public static Map paimonUidsByOperatorName(StreamGraph graph) { + Map uids = new TreeMap<>(); + for (StreamNode node : graph.getStreamNodes()) { + String uid = node.getTransformationUID(); + if (uid != null && !uid.startsWith(TEST_UID_PREFIX)) { + uids.put(node.getOperatorName(), uid); + } + } + return uids; + } + + /** + * Uid to operator id for the nodes Paimon built. A checkpoint entry is keyed by the operator + * id, so this map is what has to stay put as the rest of the job changes. + */ + public static Map paimonOperatorIdsByUid(StreamGraph graph) { + Map hashes = + new StreamGraphHasherV2().traverseStreamGraphAndGenerateHashes(graph); + Map ids = new TreeMap<>(); + for (StreamNode node : graph.getStreamNodes()) { + String uid = node.getTransformationUID(); + if (uid != null && !uid.startsWith(TEST_UID_PREFIX)) { + ids.put(uid, hex(hashes.get(node.getId()))); + } + } + return ids; + } + + private static String hex(byte[] bytes) { + StringBuilder builder = new StringBuilder(); + for (byte b : bytes) { + builder.append(String.format("%02x", b)); + } + return builder.toString(); + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidMigrationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidMigrationITCase.java new file mode 100644 index 000000000000..536caac20406 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidMigrationITCase.java @@ -0,0 +1,396 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.sink; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; + +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ExternalizedCheckpointRetention; +import org.apache.flink.configuration.StateRecoveryOptions; +import org.apache.flink.connector.file.src.FileSource; +import org.apache.flink.connector.file.src.reader.TextLineInputFormat; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.fs.FSDataInputStream; +import org.apache.flink.core.fs.Path; +import org.apache.flink.runtime.checkpoint.Checkpoints; +import org.apache.flink.runtime.checkpoint.OperatorState; +import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamGraphHasherV2; +import org.apache.flink.streaming.api.graph.StreamNode; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.io.DataInputStream; +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.apache.paimon.CoreOptions.PATH; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.LogicalTypeConversion.toDataType; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.fail; + +/** + * Steps 3 and 4 of the migration {@code docs/docs/flink/savepoint.md} prescribes for turning {@code + * sink.operator-uid.cover-all-operators} on under a running job. + * + *

Restores go through {@code execution.state-recovery.path}, which {@code + * Checkpoints.loadAndValidateCheckpoint} guards: unlike the high-availability store it skips an + * orphaned entry holding nothing and rejects one holding state. {@link + * OperatorUidSuffixRestoreITCase} covers the store route. + */ +class OperatorUidMigrationITCase { + + private static final String UID_SUFFIX = "test-uid"; + private static final String TABLE_NAME = "append_table"; + private static final long DEADLINE_MILLIS = 120_000; + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = new MiniClusterExtension(); + + private static final RowType TABLE_TYPE = + new RowType( + Arrays.asList( + new RowType.RowField("k", new IntType()), + new RowType.RowField("pt", new VarCharType(10)))); + + private static final DataType INPUT_TYPE = + DataTypes.ROW( + DataTypes.FIELD("k", DataTypes.INT()), + DataTypes.FIELD("pt", DataTypes.STRING())); + + @TempDir private java.nio.file.Path warehouse; + @TempDir private java.nio.file.Path checkpoints; + @TempDir private java.nio.file.Path input; + + /** Guide step 3: the operators an unaware-bucket append table covers all hold nothing. */ + @Test + @Timeout(300) + void testStatelessOperatorsMigrateWithoutIgnoringUnclaimedState() throws Exception { + FileStoreTable before = table("stateless", false, false); + String path = runAndCheckpoint(before); + + assertRestored(table("stateless", true, false), path, false, snapshot(before)); + } + + /** + * Guide step 4. {@code Collect Statistics} has an operator coordinator, whose entry is never + * empty, so the same restore is rejected until it is told to drop unclaimed state. Which + * entries it then drops is asserted by name, because the log prints only opaque ids. + */ + @Test + @Timeout(300) + void testPartitionDynamicNeedsIgnoreUnclaimedStateOnce() throws Exception { + FileStoreTable before = table("partition-dynamic", false, true); + String path = runAndCheckpoint(before); + Snapshot committed = snapshot(before); + FileStoreTable after = table("partition-dynamic", true, true); + + assertThatThrownBy(() -> assertRestored(after, path, false, committed)) + .as("the statistics coordinator's entry holds state, so it cannot be skipped") + .hasStackTraceContaining("Cannot map checkpoint/savepoint state for operator"); + + assertRestored(after, path, true, committed); + + Map claimed = operatorsById(job(before, path, false)); + Set dropped = new HashSet<>(operatorIdsIn(path)); + dropped.retainAll(claimed.keySet()); + dropped.removeAll(operatorsById(job(after, path, false)).keySet()); + List names = new ArrayList<>(); + List uids = new ArrayList<>(); + for (String id : dropped) { + names.add(claimed.get(id).name); + uids.add(claimed.get(id).uid); + } + + assertThat(names) + .as("the statistics coordinator's is the dropped entry that held state") + .contains("Collect Statistics: " + TABLE_NAME) + .doesNotContain("Writer : " + TABLE_NAME, "Global Committer : " + TABLE_NAME); + assertThat(uids) + .as("nothing that already carried a uid loses its entry") + .containsOnlyNulls(); + } + + // ------------------------------------------------------------------------ + // running jobs + // ------------------------------------------------------------------------ + + /** + * The commit user is a fresh UUID per job, kept in the Global Committer's {@code + * commit_user_state}, so a committer whose entry had been dropped would commit under a new one. + * Reaching {@code RUNNING} would prove nothing; a later snapshot under the same user can only + * come from restored state. + */ + private void assertRestored( + FileStoreTable table, String path, boolean ignoreUnclaimedState, Snapshot committed) + throws Exception { + JobClient client = job(table, path, ignoreUnclaimedState).executeAsync(); + awaitCheckpoint(client); + // a second file, because the restored source resumes past the first one + Files.write(input.resolve("more.txt"), Arrays.asList("3", "4")); + Snapshot restored = awaitSnapshotAfter(client, table, committed.id()); + client.cancel().get(); + + assertThat(restored.commitUser()) + .as("the committer kept its uid, so its commit user came back from state") + .isEqualTo(committed.commitUser()); + } + + /** Writes three rows, checkpoints once they are committed, and returns the path. */ + private String runAndCheckpoint(FileStoreTable table) throws Exception { + Files.write(input.resolve("rows.txt"), Arrays.asList("0", "1", "2")); + + JobClient client = job(table, null, false).executeAsync(); + awaitCheckpoint(client); + awaitSnapshotAfter(client, table, 0); + // one more, so the checkpoint is no older than the snapshot the restore has to continue + String path = miniCluster(client).triggerCheckpoint(client.getJobID()).get(); + client.cancel().get(); + return path; + } + + /** Checkpoints until the table has a snapshot later than {@code id}, and returns it. */ + private Snapshot awaitSnapshotAfter(JobClient client, FileStoreTable table, long id) + throws Exception { + long deadline = System.currentTimeMillis() + DEADLINE_MILLIS; + while (true) { + Snapshot snapshot = snapshot(table); + if (snapshot != null && snapshot.id() > id) { + return snapshot; + } + if (System.currentTimeMillis() > deadline) { + fail("no snapshot after %d was committed in time", id); + } + miniCluster(client).triggerCheckpoint(client.getJobID()).get(); + } + } + + /** + * A restore Flink rejects surfaces here, as the job going terminal rather than from the submit, + * so this waits on the status itself: {@code CommonTestUtils.waitForAllTaskRunning} would block + * forever on a job whose tasks never start. + */ + private static void awaitCheckpoint(JobClient client) throws Exception { + long deadline = System.currentTimeMillis() + DEADLINE_MILLIS; + while (true) { + JobStatus status = client.getJobStatus().get(); + if (status.isGloballyTerminalState()) { + client.getJobExecutionResult().get(20, TimeUnit.SECONDS); + fail("the job reached %s without checkpointing", status); + } + if (status == JobStatus.RUNNING && allTasksRunning(client)) { + miniCluster(client).triggerCheckpoint(client.getJobID()).get(); + return; + } + if (System.currentTimeMillis() > deadline) { + fail("the job never checkpointed, last status was %s", status); + } + Thread.sleep(500); + } + } + + private static boolean allTasksRunning(JobClient client) throws Exception { + AtomicBoolean running = new AtomicBoolean(true); + miniCluster(client) + .getExecutionGraph(client.getJobID()) + .thenAccept( + graph -> + graph.getAllExecutionVertices() + .forEach( + vertex -> { + if (vertex.getExecutionState() + != ExecutionState.RUNNING) { + running.set(false); + } + })) + .get(); + return running.get(); + } + + /** As {@code org.apache.paimon.flink.FlinkJobRecoveryITCase} does, for want of a public API. */ + private static MiniCluster miniCluster(JobClient client) throws Exception { + Field field = client.getClass().getDeclaredField("miniCluster"); + field.setAccessible(true); + return (MiniCluster) field.get(client); + } + + /** + * One topology across the migration, as the guide requires, with everything outside the sink + * uid'd, so only the sink's operators can move. + */ + private StreamExecutionEnvironment job( + FileStoreTable table, String recoverFrom, boolean ignoreUnclaimedState) { + Configuration conf = new Configuration(); + conf.set( + CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION, + ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION); + conf.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpoints.toUri().toString()); + if (recoverFrom != null) { + conf.set(StateRecoveryOptions.SAVEPOINT_PATH, recoverFrom); + conf.set(StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE, ignoreUnclaimedState); + } + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf); + env.setParallelism(1); + env.enableCheckpointing(Duration.ofDays(1).toMillis()); + + // the directory is monitored, so the job idles between files instead of finishing, and a + // file dropped in after a restore reaches the sink + FileSource source = + FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path(input.toUri())) + .monitorContinuously(Duration.ofMillis(200)) + .build(); + + DataStream rows = + env.fromSource(source, WatermarkStrategy.noWatermarks(), "rows") + .uid("rows") + .map(line -> Row.of(Integer.parseInt(line), "pt-" + line)) + .returns(Types.ROW(Types.INT, Types.STRING)) + .uid("to-row"); + + new FlinkSinkBuilder(table).forRow(rows, INPUT_TYPE).build(); + return env; + } + + // ------------------------------------------------------------------------ + // operator ids, on both sides of the restore + // ------------------------------------------------------------------------ + + /** The operator ids a checkpoint holds an entry for. */ + private static List operatorIdsIn(String path) throws Exception { + Path metadata = new Path(path, AbstractFsCheckpointStorageAccess.METADATA_FILE_NAME); + try (FSDataInputStream in = metadata.getFileSystem().open(metadata)) { + CheckpointMetadata checkpoint = + Checkpoints.loadCheckpointMetadata( + new DataInputStream(in), + OperatorUidMigrationITCase.class.getClassLoader(), + path); + List ids = new ArrayList<>(); + for (OperatorState state : checkpoint.getOperatorStates()) { + ids.add(state.getOperatorID().toHexString()); + } + return ids; + } + } + + /** + * Keyed by the id Flink compares a checkpoint entry against, so a dropped entry can be named. + */ + private static Map operatorsById(StreamExecutionEnvironment env) { + StreamGraph graph = env.getStreamGraph(); + Map hashes = + new StreamGraphHasherV2().traverseStreamGraphAndGenerateHashes(graph); + Map operators = new HashMap<>(); + for (StreamNode node : graph.getStreamNodes()) { + operators.put( + new OperatorID(hashes.get(node.getId())).toHexString(), + new Operator(node.getOperatorName(), node.getTransformationUID())); + } + return operators; + } + + private static final class Operator { + + private final String name; + private final String uid; + + private Operator(String name, String uid) { + this.name = name; + this.uid = uid; + } + } + + // ------------------------------------------------------------------------ + // the table + // ------------------------------------------------------------------------ + + private static Snapshot snapshot(FileStoreTable table) { + return table.snapshotManager().latestSnapshotFromFileSystem(); + } + + /** One directory per uid layout, so both jobs write a table of the same name. */ + private FileStoreTable table(String directory, boolean coverAll, boolean partitionDynamic) + throws Exception { + Options options = new Options(); + options.set(PATH, warehouse.resolve(directory).resolve(TABLE_NAME).toString()); + options.set(SINK_OPERATOR_UID_SUFFIX, UID_SUFFIX); + options.set(SINK_OPERATOR_UID_COVER_ALL_OPERATORS, coverAll); + if (partitionDynamic) { + options.set( + CoreOptions.PARTITION_SINK_STRATEGY, + CoreOptions.PartitionSinkStrategy.PARTITION_DYNAMIC); + } + + java.nio.file.Path path = warehouse.resolve(directory).resolve(TABLE_NAME); + if (!Files.exists(path)) { + new FileSystemSchemaManager( + LocalFileIO.create(), new CoreOptions(options.toMap()).path()) + .createTable( + new Schema( + toDataType(TABLE_TYPE).getFields(), + partitionDynamic + ? Collections.singletonList("pt") + : Collections.emptyList(), + Collections.emptyList(), + options.toMap(), + "")); + } + return FileStoreTableFactory.create(LocalFileIO.create(), options); + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixRestoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixRestoreITCase.java new file mode 100644 index 000000000000..559ad279faf1 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixRestoreITCase.java @@ -0,0 +1,310 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.sink; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; + +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HighAvailabilityOptions; +import org.apache.flink.connector.file.src.FileSource; +import org.apache.flink.connector.file.src.reader.TextLineInputFormat; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.runtime.checkpoint.CheckpointsCleaner; +import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore; +import org.apache.flink.runtime.checkpoint.PerJobCheckpointRecoveryFactory; +import org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore; +import org.apache.flink.runtime.highavailability.HighAvailabilityServices; +import org.apache.flink.runtime.highavailability.HighAvailabilityServicesFactory; +import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedHaServicesWithLeadershipControl; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.types.Row; +import org.apache.flink.util.ExceptionUtils; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; + +import static org.apache.paimon.CoreOptions.PATH; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.LogicalTypeConversion.toDataType; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowable; +import static org.assertj.core.api.Assertions.fail; + +/** + * A job must still restore from a retained checkpoint after an operator is added upstream of a + * Paimon sink, which is what {@code sink.operator-uid.suffix} is for. On its own the suffix does + * not deliver that: some sink operators carry no uid and Flink then derives their ids from the + * shape of the stream graph. + * + *

The table asks for full coverage through {@link + * org.apache.paimon.flink.FlinkConnectorOptions#SINK_OPERATOR_UID_COVER_ALL_OPERATORS}, without + * which the suffix reaches only the writer and the committer. + * + *

Only a checkpoint recovered from the {@link CompletedCheckpointStore} shows this, which is + * what a JobManager restores from when it is highly available. The savepoint route hides it. {@code + * org.apache.paimon.flink.FlinkJobRecoveryITCase#testRestoreFromSavepointWithJobGraphChange} takes + * a savepoint and restores through a configured {@code execution.state-recovery.path}, which runs + * {@code Checkpoints.loadAndValidateCheckpoint} first. That silently skips an unmatched operator + * holding no state, and the row conversions and the compaction operators hold none, so the orphaned + * entries are gone before the JobMaster ever compares ids. Recovering from the store compares every + * id and fails on the first it cannot place, which is what production does. + */ +class OperatorUidSuffixRestoreITCase { + + private static final String UID_SUFFIX = "test-uid"; + private static final String TABLE_NAME = "append_table"; + + /** + * Two of these run in sequence per test, plus the 20 second net below, so the worst case stays + * inside the {@code @Timeout} and the informative message is the one that gets printed. + */ + private static final long DEADLINE_MILLIS = 60_000; + + /** Read on every recovery, so each test gets a fresh store from a class scoped cluster. */ + private static volatile RetainingCheckpointStore retainedStore; + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setConfiguration( + new Configuration() + .set( + HighAvailabilityOptions.HA_MODE, + RetainingHaServicesFactory.class.getName())) + .build()); + + private static final RowType TABLE_TYPE = + new RowType( + Arrays.asList( + new RowType.RowField("k", new IntType()), + new RowType.RowField("v", new VarCharType(10)))); + + private static final DataType INPUT_TYPE = + DataTypes.ROW( + DataTypes.FIELD("k", DataTypes.INT()), + DataTypes.FIELD("v", DataTypes.STRING())); + + @TempDir private Path warehouse; + @TempDir private Path emptySource; + + @BeforeEach + void setUp() { + retainedStore = new RetainingCheckpointStore(); + } + + /** What the suffix promises. Fails as soon as one sink operator loses its uid. */ + @Test + @Timeout(180) + void testRestoreAfterUpstreamTopologyChange() throws Exception { + FileStoreTable table = unawareBucketTable("unchanged-suffix", UID_SUFFIX); + + runUntilCheckpointed(buildSinkJob(table, 0)); + + Throwable failure = restoreFailure(buildSinkJob(table, 1).executeAsync()); + assertThat(failure) + .as( + "the job did not restore from the retained checkpoint after one operator " + + "was added upstream of the sink:%n%s", + failure == null ? "" : ExceptionUtils.stringifyException(failure)) + .isNull(); + } + + /** + * The permanent negative. Changing the suffix renames every uid Paimon assigns, so the retained + * checkpoint holds entries no operator in the new graph claims and the restore has to fail. The + * exception is the only proof that a restore is being attempted at all, and it is the same one + * production reported: + * + * 

+     * IllegalStateException: There is no operator for the state <hash>
+     *
+ * + *

The two jobs are otherwise identical, down to the table name the uids are built from, so + * nothing but the suffix can account for the failure. + */ + @Test + @Timeout(180) + void testRestoreRejectsAChangedUidSuffix() throws Exception { + runUntilCheckpointed(buildSinkJob(unawareBucketTable("one-suffix", UID_SUFFIX), 0)); + + Throwable failure = + restoreFailure( + buildSinkJob(unawareBucketTable("another-suffix", "other-uid"), 0) + .executeAsync()); + + assertThat(failure) + .as("a renamed uid orphans a checkpoint entry, which has to be rejected") + .isNotNull(); + assertThat(ExceptionUtils.stringifyException(failure)) + .contains("There is no operator for the state"); + } + + private static void runUntilCheckpointed(StreamExecutionEnvironment env) throws Exception { + JobClient client = env.executeAsync(); + long deadline = System.currentTimeMillis() + DEADLINE_MILLIS; + while (retainedStore.getAllCheckpoints().isEmpty()) { + if (System.currentTimeMillis() > deadline) { + fail("no checkpoint completed in time"); + } + Thread.sleep(200); + } + client.cancel().get(); + assertThat(retainedStore.getAllCheckpoints()).isNotEmpty(); + } + + /** + * Why the job stopped, or null if it reached {@code RUNNING}. Null means only that nothing + * rejected the checkpoint, never that a restore took place. + */ + private static Throwable restoreFailure(JobClient client) throws Exception { + long deadline = System.currentTimeMillis() + DEADLINE_MILLIS; + while (true) { + JobStatus status = client.getJobStatus().get(); + if (status == JobStatus.RUNNING) { + client.cancel().get(); + return null; + } + if (status.isGloballyTerminalState()) { + // the job is already terminal, so the future is complete; this is only a net + return catchThrowable( + () -> client.getJobExecutionResult().get(20, TimeUnit.SECONDS)); + } + if (System.currentTimeMillis() > deadline) { + fail("job never reached %s, last status was %s", JobStatus.RUNNING, status); + } + Thread.sleep(200); + } + } + + /** + * Without a primary key the table is BUCKET_UNAWARE, the mode {@link AppendTableSink} serves. + * One parent directory per table, so two tables can differ only in their uid suffix and still + * be called the same thing, which is the other half of what a uid is built from. + */ + private FileStoreTable unawareBucketTable(String directory, String uidSuffix) throws Exception { + Options options = new Options(); + options.set(PATH, warehouse.resolve(directory).resolve(TABLE_NAME).toString()); + options.set(SINK_OPERATOR_UID_SUFFIX, uidSuffix); + options.set(SINK_OPERATOR_UID_COVER_ALL_OPERATORS, true); + + Schema schema = + new Schema( + toDataType(TABLE_TYPE).getFields(), + Collections.emptyList(), + Collections.emptyList(), + options.toMap(), + ""); + new FileSystemSchemaManager(LocalFileIO.create(), new CoreOptions(options.toMap()).path()) + .createTable(schema); + return FileStoreTableFactory.create(LocalFileIO.create(), options); + } + + /** Everything outside the sink is uid'd, so only the sink can lose its operator ids. */ + private StreamExecutionEnvironment buildSinkJob(FileStoreTable table, int extraOperators) { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.enableCheckpointing(500); + + // an empty directory monitored forever: no records, and the job never finishes + FileSource idle = + FileSource.forRecordStreamFormat( + new TextLineInputFormat(), + new org.apache.flink.core.fs.Path(emptySource.toUri())) + .monitorContinuously(Duration.ofDays(1)) + .build(); + + DataStream source = + env.fromSource(idle, WatermarkStrategy.noWatermarks(), "idle") + .uid("idle") + .map(line -> Row.of(0, line)) + .returns(Types.ROW(Types.INT, Types.STRING)) + .uid("to-row"); + + for (int i = 0; i < extraOperators; i++) { + source = + source.map(row -> row) + .returns(Types.ROW(Types.INT, Types.STRING)) + .uid("extra-" + i); + } + + // deliberately not uid'd here: a uid on the returned sink would override Paimon's own + // uid on the final 'end' operator, leaving its coverage untested + new FlinkSinkBuilder(table).forRow(source, INPUT_TYPE).build(); + return env; + } + + /** Keeps its checkpoints when the job goes away, as an external store would. */ + private static final class RetainingCheckpointStore extends StandaloneCompletedCheckpointStore { + + RetainingCheckpointStore() { + super(1); + } + + @Override + public void shutdown(JobStatus jobStatus, CheckpointsCleaner checkpointsCleaner) { + // Deliberately does not call super, which clears the checkpoint list in a finally + // whatever the job status, leaving the next job nothing to restore from. Retaining + // externalized checkpoints does not help either, that only spares the files. Outliving + // the JobManager is the one property of a real store being stood in for here. + } + } + + /** Instantiated by Flink from {@code high-availability.type}. */ + public static final class RetainingHaServicesFactory + implements HighAvailabilityServicesFactory { + + @Override + public HighAvailabilityServices createHAServices( + Configuration configuration, Executor executor) { + // ignoring the job id hands every job the one store, so the next job is offered it + return new EmbeddedHaServicesWithLeadershipControl( + executor, + new PerJobCheckpointRecoveryFactory( + (maxCheckpoints, previous, registry, ioExecutor, restoreMode) -> + retainedStore)); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java new file mode 100644 index 000000000000..72dc5ae68387 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java @@ -0,0 +1,448 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.sink; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.flink.FlinkConnectorOptions; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.PipelineOptions; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamNode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType; +import static org.apache.paimon.flink.OperatorUidGraphs.TEST_UID_PREFIX; +import static org.apache.paimon.flink.OperatorUidGraphs.operatorsWithoutUid; +import static org.apache.paimon.flink.OperatorUidGraphs.paimonOperatorIdsByUid; +import static org.apache.paimon.flink.OperatorUidGraphs.paimonUidsByOperatorName; +import static org.apache.paimon.flink.OperatorUidGraphs.uidsByOperatorName; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +/** + * Asserts that with {@code sink.operator-uid.cover-all-operators} every operator a streaming write + * adds carries a uid, one shape per site that adds operators. + */ +class OperatorUidSuffixTest { + + private static final String UID_SUFFIX = "test-uid"; + + /** Every table is called this, so a uid does not vary with the shape that built it. */ + private static final String TABLE_NAME = "tbl"; + + private static final RowType TABLE_TYPE = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.BIGINT(), DataTypes.STRING()}, + new String[] {"k", "v", "pt"}); + + @TempDir static Path warehouse; + + static List shapes() { + return Arrays.asList( + // AppendTableSink.doWrite, the streaming compaction pair + shape("append").reaches("Compact Coordinator: tbl", "Compact Worker: tbl"), + // FlinkSinkBuilder.forRow, a Row to RowData conversion on the caller's stream + shape("append-for-row").forRow().reaches("Map"), + // AppendTableSink.doWrite, the pre-commit compaction pair + shape("append-precommit-compact") + .option(FlinkConnectorOptions.PRECOMMIT_COMPACT.key(), "true") + .reaches( + "New Files Compact Coordinator: tbl", + "New Files Compact Worker: tbl"), + // FlinkSinkBuilder.applyDynamicPartitionShuffle + shape("append-partition-dynamic") + .partitioned() + .option( + CoreOptions.PARTITION_SINK_STRATEGY.key(), + CoreOptions.PartitionSinkStrategy.PARTITION_DYNAMIC.name()) + .reaches("Collect Statistics: tbl", "Strip Statistics"), + // FlinkSink.doCoordinatorCommit: 'end' follows the writer, with no global + // committer. Both commit routes name that node the same, so the route is + // identified by the committer it does not build. + shape("append-coordinator-commit") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(FlinkConnectorOptions.SINK_COORDINATOR_COMMIT_ENABLED.key(), "true") + .reaches("end: Writer") + .avoids("Global Committer : tbl"), + // FixedBucketSink on an append table, no compaction operators at all + shape("append-fixed").bucket(2).reaches("Writer : tbl", "Global Committer : tbl"), + shape("pk-fixed") + .primaryKey() + .bucket(2) + .reaches("Writer : tbl", "Global Committer : tbl"), + // FlinkSinkBuilder.build, the local merge operator + shape("pk-fixed-local-merge") + .primaryKey() + .bucket(2) + .option(CoreOptions.LOCAL_MERGE_BUFFER_SIZE.key(), "64 mb") + .reaches("local merge"), + // FlinkSink.doWrite, the changelog compaction trio + shape("pk-fixed-precommit-compact") + .primaryKey() + .bucket(2) + .option(FlinkConnectorOptions.PRECOMMIT_COMPACT.key(), "true") + .reaches( + "Changelog Compact Coordinator", + "Changelog Compact Worker", + "Changelog Sort by Creation Time"), + shape("pk-dynamic").primaryKey().bucket(-1).reaches("dynamic-bucket-assigner"), + // GlobalDynamicBucketSink, when the primary key does not cover the partition keys + shape("pk-cross-partition") + .primaryKey() + .partitioned() + .bucket(-1) + .reaches("INDEX_BOOTSTRAP", "cross-partition-bucket-assigner"), + shape("pk-postpone").primaryKey().bucket(-2).reaches("Writer : tbl")); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("shapes") + void testEveryOperatorCarriesAUid(Shape shape) { + // auto-generate-uids off, so building the graph is itself Flink's uid assertion + StreamGraph graph = buildGraph(shape, 0, false); + + List names = + graph.getStreamNodes().stream() + .map(StreamNode::getOperatorName) + .collect(Collectors.toList()); + assertThat(names) + .as("shape %s exists to reach %s", shape, shape.reaches) + .containsAll(shape.reaches); + for (String avoided : shape.avoids) { + assertThat(names) + .as("shape %s is the route that does not build '%s'", shape, avoided) + .doesNotContain(avoided); + } + assertThat(paimonUidsByOperatorName(graph).values()) + .as("shape %s: every uid is built from the table name and the suffix", shape) + .allMatch(uid -> uid.endsWith("_" + TABLE_NAME + "_" + UID_SUFFIX)); + } + + /** + * What the uids are for. Flink hashes a uid'd node from the uid alone, so with every node + * covered the operator ids no longer depend on what sits upstream of the sink. + */ + @Test + void testOperatorIdsSurviveAnUpstreamChange() { + Shape shape = shape("append"); + assertThat(paimonOperatorIdsByUid(buildGraph(shape, 1, true))) + .isEqualTo(paimonOperatorIdsByUid(buildGraph(shape, 0, true))); + } + + /** The coverage is opt-in: with the option off the graph is what it was before the option. */ + @Test + void testTheOptionOffLeavesTodaysUidsOnly() { + StreamGraph graph = buildGraph(shape("option-off").coverAll(false), 0, true); + + assertThat(paimonUidsByOperatorName(graph)) + .containsOnly( + entry("Writer : tbl", "Writer_tbl_test-uid"), + entry("Global Committer : tbl", "Global Committer_tbl_test-uid")); + assertThat(operatorsWithoutUid(graph)).isNotEmpty(); + } + + /** With no suffix there is nothing to build a uid out of, option or not. */ + @Test + void testTheOptionWithoutASuffixAssignsNothing() { + assertThat(paimonUidsByOperatorName(buildGraph(shape("no-suffix").suffix(null), 0, true))) + .isEmpty(); + } + + /** + * The uids that exist today, pinned literally. A change to {@link + * FlinkConnectorOptions#generateCustomUid} orphans every checkpoint written by an older Paimon, + * and nothing else in the repo would notice. + */ + @Test + void testTheUidsThatExistTodayAreUnchanged() { + assertThat(uidsByOperatorName(buildGraph(shape("golden").primaryKey().bucket(-1), 0, true))) + .containsEntry("Writer : tbl", "Writer_tbl_test-uid") + .containsEntry("Global Committer : tbl", "Global Committer_tbl_test-uid") + .containsEntry("dynamic-bucket-assigner", "dynamic-bucket-assigner_tbl_test-uid"); + + // write-only renames the writer's operator but not its uid + assertThat( + uidsByOperatorName( + buildGraph( + shape("golden-write-only") + .option(CoreOptions.WRITE_ONLY.key(), "true"), + 0, + true))) + .containsEntry("Writer(write-only) : tbl", "Writer_tbl_test-uid"); + } + + /** + * Every uid of two rich shapes, pinned literally, so a change to any prefix fails a test + * instead of silently orphaning the checkpoints of every job that set the option. + */ + @Test + void testTheNewUidsArePinned() { + Map appendPrecommitCompact = new TreeMap<>(); + appendPrecommitCompact.put("Compact Coordinator: tbl", "Compact Coordinator_tbl_test-uid"); + appendPrecommitCompact.put("Compact Worker: tbl", "Compact Worker_tbl_test-uid"); + appendPrecommitCompact.put("Global Committer : tbl", "Global Committer_tbl_test-uid"); + appendPrecommitCompact.put("Map", "Internal Row Conversion_tbl_test-uid"); + appendPrecommitCompact.put( + "New Files Compact Coordinator: tbl", "New Files Compact Coordinator_tbl_test-uid"); + appendPrecommitCompact.put( + "New Files Compact Worker: tbl", "New Files Compact Worker_tbl_test-uid"); + appendPrecommitCompact.put("Writer : tbl", "Writer_tbl_test-uid"); + appendPrecommitCompact.put("end: Writer", "end_tbl_test-uid"); + assertThat( + paimonUidsByOperatorName( + buildGraph( + shape("pinned-append-precommit-compact") + .option( + FlinkConnectorOptions.PRECOMMIT_COMPACT + .key(), + "true"), + 0, + true))) + .isEqualTo(appendPrecommitCompact); + + Map localMerge = new TreeMap<>(); + localMerge.put("Global Committer : tbl", "Global Committer_tbl_test-uid"); + localMerge.put("Map", "Internal Row Conversion_tbl_test-uid"); + localMerge.put("Writer : tbl", "Writer_tbl_test-uid"); + localMerge.put("end: Writer", "end_tbl_test-uid"); + localMerge.put("local merge", "local merge_tbl_test-uid"); + assertThat( + paimonUidsByOperatorName( + buildGraph( + shape("pinned-pk-fixed-local-merge") + .primaryKey() + .bucket(2) + .option( + CoreOptions.LOCAL_MERGE_BUFFER_SIZE.key(), + "64 mb"), + 0, + true))) + .isEqualTo(localMerge); + + Map crossPartition = new TreeMap<>(); + crossPartition.put("Global Committer : tbl", "Global Committer_tbl_test-uid"); + crossPartition.put("INDEX_BOOTSTRAP", "INDEX_BOOTSTRAP_tbl_test-uid"); + crossPartition.put("Map", "Internal Row Conversion_tbl_test-uid"); + crossPartition.put("Writer : tbl", "Writer_tbl_test-uid"); + crossPartition.put( + "cross-partition-bucket-assigner", "cross-partition-bucket-assigner_tbl_test-uid"); + crossPartition.put("end: Writer", "end_tbl_test-uid"); + assertThat( + paimonUidsByOperatorName( + buildGraph( + shape("pinned-pk-cross-partition") + .primaryKey() + .partitioned() + .bucket(-1), + 0, + true))) + .isEqualTo(crossPartition); + } + + // ------------------------------------------------------------------------ + // graph building + // ------------------------------------------------------------------------ + + private static StreamGraph buildGraph( + Shape shape, int extraUpstreamOperators, boolean autoGenerateUids) { + Configuration conf = new Configuration(); + conf.set(PipelineOptions.AUTO_GENERATE_UIDS, autoGenerateUids); + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf); + env.setParallelism(2); + env.setRuntimeMode(RuntimeExecutionMode.STREAMING); + // coordinator commit insists on checkpointing with one concurrent checkpoint + env.enableCheckpointing(1000); + env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); + + FileStoreTable table = shape.createTable(); + org.apache.flink.table.types.logical.RowType logicalType = toLogicalType(table.rowType()); + FlinkSinkBuilder builder = new FlinkSinkBuilder(table); + if (shape.forRow) { + org.apache.flink.table.types.DataType dataType = fromLogicalToDataType(logicalType); + builder.forRow( + source(env, ExternalTypeInfo.of(dataType), extraUpstreamOperators), dataType); + } else { + TypeInformation typeInfo = InternalTypeInfo.of(logicalType); + builder.forRowData(source(env, typeInfo, extraUpstreamOperators)); + } + // deliberately not uid'd by the test: the terminal sink is Paimon's operator, and SQL + // users have no hook to uid it themselves + builder.build(); + return env.getStreamGraph(); + } + + /** Everything the test adds carries a uid, so an operator without one is Paimon's. */ + private static DataStream source( + StreamExecutionEnvironment env, TypeInformation typeInfo, int extraOperators) { + DataStream stream = + env.fromCollection(Collections.emptyList(), typeInfo) + .name("test-source") + .uid(TEST_UID_PREFIX + "source"); + for (int i = 0; i < extraOperators; i++) { + stream = + stream.map((MapFunction) value -> value) + .returns(typeInfo) + .name("test-extra-" + i) + .uid(TEST_UID_PREFIX + "extra-" + i); + } + return stream; + } + + private static Shape shape(String name) { + return new Shape(name); + } + + /** A table definition plus the entry point that feeds the sink. */ + private static final class Shape { + + private final String name; + private final Map options = new LinkedHashMap<>(); + private List reaches = Collections.emptyList(); + private List avoids = Collections.emptyList(); + private List partitionKeys = Collections.emptyList(); + private List primaryKeys = Collections.emptyList(); + private int bucket = -1; + private boolean forRow; + private boolean coverAll = true; + private String uidSuffix = UID_SUFFIX; + + private Shape(String name) { + this.name = name; + } + + private Shape reaches(String... operatorNames) { + this.reaches = Arrays.asList(operatorNames); + return this; + } + + /** Names no node may carry, for a route identified by what it does not build. */ + private Shape avoids(String... operatorNames) { + this.avoids = Arrays.asList(operatorNames); + return this; + } + + private Shape partitioned() { + this.partitionKeys = Collections.singletonList("pt"); + return this; + } + + private Shape primaryKey() { + this.primaryKeys = Collections.singletonList("k"); + return this; + } + + /** {@code -1} is unaware or dynamic, {@code -2} postpone, a positive value fixed. */ + private Shape bucket(int bucket) { + this.bucket = bucket; + return this; + } + + private Shape forRow() { + this.forRow = true; + return this; + } + + private Shape coverAll(boolean coverAll) { + this.coverAll = coverAll; + return this; + } + + private Shape suffix(String uidSuffix) { + this.uidSuffix = uidSuffix; + return this; + } + + private Shape option(String key, String value) { + this.options.put(key, value); + return this; + } + + /** A fresh table per graph, in its own directory so every table is called TABLE_NAME. */ + private FileStoreTable createTable() { + Options options = Options.fromMap(this.options); + org.apache.paimon.fs.Path path = + new org.apache.paimon.fs.Path( + warehouse + .resolve(name + "-" + UUID.randomUUID()) + .resolve(TABLE_NAME) + .toString()); + options.set(CoreOptions.PATH, path.toString()); + options.set(CoreOptions.BUCKET, bucket); + if (bucket > 0 && primaryKeys.isEmpty()) { + options.set(CoreOptions.BUCKET_KEY, "k"); + } + if (uidSuffix != null) { + options.set(SINK_OPERATOR_UID_SUFFIX, uidSuffix); + } + options.set(SINK_OPERATOR_UID_COVER_ALL_OPERATORS, coverAll); + try { + new FileSystemSchemaManager(LocalFileIO.create(), path) + .createTable( + new Schema( + TABLE_TYPE.getFields(), + partitionKeys, + primaryKeys, + options.toMap(), + "")); + } catch (Exception e) { + throw new AssertionError("shape " + name + " has an illegal table", e); + } + return FileStoreTableFactory.create(LocalFileIO.create(), options); + } + + @Override + public String toString() { + return name; + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SourceOperatorUidSuffixTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SourceOperatorUidSuffixTest.java new file mode 100644 index 000000000000..fec3a425930f --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/SourceOperatorUidSuffixTest.java @@ -0,0 +1,328 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.flink.source; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.flink.FlinkConnectorOptions; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.PipelineOptions; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.streaming.api.graph.StreamGraph; +import org.apache.flink.streaming.api.graph.StreamNode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_COVER_ALL_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.OperatorUidGraphs.TEST_UID_PREFIX; +import static org.apache.paimon.flink.OperatorUidGraphs.operatorsWithoutUid; +import static org.apache.paimon.flink.OperatorUidGraphs.paimonOperatorIdsByUid; +import static org.apache.paimon.flink.OperatorUidGraphs.paimonUidsByOperatorName; +import static org.apache.paimon.flink.OperatorUidGraphs.uidsByOperatorName; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +/** + * Asserts that with {@code source.operator-uid.cover-all-operators} every operator a streaming read + * adds carries a uid, one shape per route {@link FlinkSourceBuilder} can take. + */ +class SourceOperatorUidSuffixTest { + + private static final String UID_SUFFIX = "test-uid"; + + /** Every table is called this, so a uid does not vary with the shape that built it. */ + private static final String TABLE_NAME = "tbl"; + + private static final RowType TABLE_TYPE = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.BIGINT(), DataTypes.STRING()}, + new String[] {"k", "v", "pt"}); + + @TempDir static Path warehouse; + + static List shapes() { + return Arrays.asList( + // FlinkSourceBuilder.toDataStream, the one node the suffix reaches on its own + shape("bounded").bounded().reaches("Source: tbl"), + shape("unbounded").reaches("Source: tbl"), + // FlinkSourceBuilder.buildForRow, a RowData to Row conversion + shape("bounded-for-row").bounded().forRow().reaches("Map"), + // MonitorSource.buildSource: a monitor and a reader, neither built by toDataStream + shape("dedicated-split-generation") + .bounded() + .option(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION.key(), "true") + .reaches("Source: tbl-Monitor", "tbl-Reader"), + // the same route with a watermark strategy, assigned as a separate operator here + shape("dedicated-split-generation-watermarked") + .bounded() + .watermarked() + .option(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION.key(), "true") + .reaches("Timestamps/Watermarks"), + // the unbounded way into the same MonitorSource route + shape("consumer-exactly-once") + .option(CoreOptions.CONSUMER_ID.key(), "uid-coverage-consumer") + .option(CoreOptions.CONSUMER_EXPIRATION_TIME.key(), "1 d") + .option(CoreOptions.CONSUMER_CONSISTENCY_MODE.key(), "exactly-once") + .reaches("Source: tbl-Monitor", "tbl-Reader"), + // buildAlignedContinuousFileSource, still toDataStream + shape("checkpoint-align") + .option(FlinkConnectorOptions.SOURCE_CHECKPOINT_ALIGN_ENABLED.key(), "true") + .reaches("Source: tbl")); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("shapes") + void testEveryOperatorCarriesAUid(Shape shape) { + // auto-generate-uids off, so building the graph is itself Flink's uid assertion + StreamGraph graph = buildGraph(shape, false, false); + + List names = + graph.getStreamNodes().stream() + .map(StreamNode::getOperatorName) + .collect(Collectors.toList()); + assertThat(names) + .as("shape %s exists to reach %s", shape, shape.reaches) + .containsAll(shape.reaches); + assertThat(paimonUidsByOperatorName(graph).values()) + .as("shape %s: every uid is built from the table name and the suffix", shape) + .allMatch(uid -> uid.endsWith("_" + TABLE_NAME + "_" + UID_SUFFIX)); + } + + /** + * A source has nothing upstream, so the change that would move its ids is another branch added + * to the same job. Flink hashes a uid-less node partly from how many nodes it hashed before it. + */ + @Test + void testOperatorIdsSurviveAnotherSourceInTheJob() { + Shape shape = + shape("dedicated-split-generation") + .bounded() + .option( + FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION.key(), + "true"); + assertThat(paimonOperatorIdsByUid(buildGraph(shape, true, true))) + .isEqualTo(paimonOperatorIdsByUid(buildGraph(shape, false, true))); + } + + /** The coverage is opt-in: with the option off the graph is what it was before the option. */ + @Test + void testTheOptionOffLeavesTodaysUidOnly() { + StreamGraph graph = + buildGraph(shape("option-off").bounded().forRow().coverAll(false), false, true); + + assertThat(paimonUidsByOperatorName(graph)) + .containsOnly(entry("Source: tbl", "Source_tbl_test-uid")); + assertThat(operatorsWithoutUid(graph)).isNotEmpty(); + } + + /** With no suffix there is nothing to build a uid out of, option or not. */ + @Test + void testTheOptionWithoutASuffixAssignsNothing() { + assertThat( + paimonUidsByOperatorName( + buildGraph(shape("no-suffix").bounded().suffix(null), false, true))) + .isEmpty(); + } + + /** The one source uid that exists today, pinned literally. */ + @Test + void testTheUidThatExistsTodayIsUnchanged() { + assertThat(uidsByOperatorName(buildGraph(shape("golden").bounded(), false, true))) + .containsEntry("Source: tbl", "Source_tbl_test-uid"); + } + + /** + * Every uid of the richest read shape, pinned literally, so a change to any prefix fails a test + * instead of silently orphaning the checkpoints of every job that set the option. + */ + @Test + void testTheNewUidsArePinned() { + Map expected = new TreeMap<>(); + expected.put("Source: tbl-Monitor", "Monitor_tbl_test-uid"); + expected.put("Timestamps/Watermarks", "Timestamps/Watermarks_tbl_test-uid"); + expected.put("tbl-Reader", "Reader_tbl_test-uid"); + assertThat( + paimonUidsByOperatorName( + buildGraph( + shape("pinned-dedicated-split-generation-watermarked") + .bounded() + .watermarked() + .option( + FlinkConnectorOptions + .SCAN_DEDICATED_SPLIT_GENERATION + .key(), + "true"), + false, + true))) + .isEqualTo(expected); + } + + // ------------------------------------------------------------------------ + // graph building + // ------------------------------------------------------------------------ + + private static StreamGraph buildGraph( + Shape shape, boolean withAnotherSource, boolean autoGenerateUids) { + Configuration conf = new Configuration(); + conf.set(PipelineOptions.AUTO_GENERATE_UIDS, autoGenerateUids); + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf); + env.setParallelism(2); + env.setRuntimeMode( + shape.bounded ? RuntimeExecutionMode.BATCH : RuntimeExecutionMode.STREAMING); + // the checkpoint-align route insists on checkpointing with one concurrent checkpoint + env.enableCheckpointing(1000); + env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); + + if (withAnotherSource) { + // an unrelated branch of the same job, entirely the test's own and fully uid'd + env.fromCollection(Collections.emptyList(), Types.LONG) + .name("test-other-source") + .uid(TEST_UID_PREFIX + "other-source") + .sinkTo(new DiscardingSink<>()) + .name("test-other-sink") + .uid(TEST_UID_PREFIX + "other-sink"); + } + + FlinkSourceBuilder builder = + new FlinkSourceBuilder(shape.createTable()).env(env).sourceBounded(shape.bounded); + if (shape.watermarked) { + builder.watermarkStrategy(WatermarkStrategy.noWatermarks()); + } + DataStream stream = shape.forRow ? builder.buildForRow() : builder.build(); + stream.sinkTo(new DiscardingSink<>()).name("test-sink").uid(TEST_UID_PREFIX + "sink"); + return env.getStreamGraph(); + } + + private static Shape shape(String name) { + return new Shape(name); + } + + /** A table definition plus the entry point and boundedness of the read. */ + private static final class Shape { + + private final String name; + private final Map options = new LinkedHashMap<>(); + private List reaches = Collections.emptyList(); + private boolean bounded; + private boolean forRow; + private boolean watermarked; + private boolean coverAll = true; + private String uidSuffix = UID_SUFFIX; + + private Shape(String name) { + this.name = name; + } + + private Shape reaches(String... operatorNames) { + this.reaches = Arrays.asList(operatorNames); + return this; + } + + private Shape bounded() { + this.bounded = true; + return this; + } + + private Shape forRow() { + this.forRow = true; + return this; + } + + private Shape watermarked() { + this.watermarked = true; + return this; + } + + private Shape coverAll(boolean coverAll) { + this.coverAll = coverAll; + return this; + } + + private Shape suffix(String uidSuffix) { + this.uidSuffix = uidSuffix; + return this; + } + + private Shape option(String key, String value) { + this.options.put(key, value); + return this; + } + + /** A fresh table per graph, in its own directory so every table is called TABLE_NAME. */ + private FileStoreTable createTable() { + Options options = Options.fromMap(this.options); + org.apache.paimon.fs.Path path = + new org.apache.paimon.fs.Path( + warehouse + .resolve(name + "-" + UUID.randomUUID()) + .resolve(TABLE_NAME) + .toString()); + options.set(CoreOptions.PATH, path.toString()); + if (uidSuffix != null) { + options.set(SOURCE_OPERATOR_UID_SUFFIX, uidSuffix); + } + options.set(SOURCE_OPERATOR_UID_COVER_ALL_OPERATORS, coverAll); + try { + new FileSystemSchemaManager(LocalFileIO.create(), path) + .createTable( + new Schema( + TABLE_TYPE.getFields(), + Collections.emptyList(), + Collections.emptyList(), + options.toMap(), + "")); + } catch (Exception e) { + throw new AssertionError("shape " + name + " has an illegal table", e); + } + return FileStoreTableFactory.create(LocalFileIO.create(), options); + } + + @Override + public String toString() { + return name; + } + } +}

MemorySize 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.
sink.operator-uid.cover-all-operators
falseBooleanIf 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.
sink.operator-uid.suffix
 (none)Duration 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.
source.operator-uid.cover-all-operators
falseBooleanIf 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.
source.operator-uid.suffix
 (none)