diff --git a/docs/docs/flink/savepoint.md b/docs/docs/flink/savepoint.md index 825636e99c81..d3c082c7a0d3 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,92 @@ chosen recovery point before executing it. Coordinate this for each Paimon table [Resume the job from the matching savepoint](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#resuming-from-savepoints). Verify that checkpoints complete and new Paimon snapshots appear before resuming dependent work. + +## Operator UIDs + +Flink maps checkpoint state to operators by id. An operator without a UID gets an id from its +place in the job graph, so adding an operator upstream, or a parallelism change that alters +chaining, gives it a new id and orphans its state. + +A Paimon source or sink is several operators. The suffix options name some of them; the cover +options name the rest: + +| Option | Operators named | +| --- | --- | +| `sink.operator-uid.suffix` | Writer, Global Committer, dynamic-bucket-assigner | +| `sink.operator-uid.cover-table-sink-operators` | Also the row conversions, `local merge`, the compaction operators, `Collect Statistics` and `Strip Statistics`, `INDEX_BOOTSTRAP` and `cross-partition-bucket-assigner`, and the final `end` sink | +| `source.operator-uid.suffix` | The source | +| `source.operator-uid.cover-table-source-operators` | Also the split monitor and reader used by dedicated split generation and exactly-once consumers, the watermark assigner, and the DataStream row conversion | + +Each UID is `__`. The table name leaves out the database, so two +tables with the same name and suffix in one job collide, and Flink rejects the job with +`Hash collision on user-specified ID`. Give each table its own suffix. + +The cover options do nothing without the matching suffix. Set the sink pair before the first +streaming job writes the table, and the source pair before the first streaming job reads it: + +```sql +ALTER TABLE my_table SET ( + 'sink.operator-uid.suffix' = 'my_table_v1', + 'sink.operator-uid.cover-table-sink-operators' = 'true', + 'source.operator-uid.suffix' = 'my_table_v1', + 'source.operator-uid.cover-table-source-operators' = 'true' +); +``` + +Not covered: + +- The CDC sync actions: the operators they add around the table sink, such as Side Output, + Schema Evolution and Case-insensitive Convert, and the shared operators of a database sync. + Such jobs cannot run with `pipeline.auto-generate-uids=false`. +- Operators only a `BATCH` job builds, such as sort and clustering. Flink does not checkpoint in + `BATCH` mode. A bounded job in `STREAMING` mode is covered. + +### Migrate a running job + +Turning on a cover option gives the newly covered operators new ids, so their entries in the +job's checkpoints become unclaimed. A restore from a savepoint or from +`execution.state-recovery.path` skips an unclaimed entry that holds no state and rejects one +that does. A restore from the high-availability checkpoint store, as in a JobManager failover or +a Kubernetes Operator `last-state` upgrade, rejects every unclaimed entry, and neither +`allowNonRestoredState` nor `execution.state-recovery.ignore-unclaimed-state` helps: + +``` +JobInitializationException: Could not start the JobMaster. +Caused by: IllegalStateException: There is no operator for the state +``` + +So migrate through a savepoint: + +1. **Stop the job** with + [stop-with-savepoint](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#stopping-a-job-with-savepoint) + (`flink stop`) and wait for it to finish. + +2. **Set the options** with `ALTER TABLE ... SET` or SQL hints. Keep the existing suffix: + changing it renames the Writer and Global Committer too, and loses their state. +3. **Restart from the savepoint** with the same topology. The operators the sink options newly + cover hold no state, so Flink skips their entries without `allowNonRestoredState` and logs + `Skipping empty savepoint state for operator` once per operator. Check that the snapshot ids + in `
$snapshots` continue past the one the job stopped at, under the same `commit_user`. + + Two cases drop an entry that holds state. Without `allowNonRestoredState` the restore fails + with `Cannot map checkpoint/savepoint state for operator `, so submit this one restart + with it. Flink then logs `Skipping savepoint state for operator ` for each entry it skips, + empty ones included. The Writer and Global Committer keep their UIDs, so Flink still restores + their state. + + - **`'partition.sink-strategy' = 'PARTITION_DYNAMIC'`**: the `Collect Statistics` coordinator's + entry is dropped. It rebuilds its statistics from the next records. + - **A read with dedicated split generation or an exactly-once consumer**: the split monitor's + entry is dropped. An exactly-once read needs a [consumer ID](./consumer-id) with the default + `consumer.mode`. After `flink stop` the consumer holds the same next snapshot as the + monitor, and the reader resumes from it, so no snapshot is skipped or read twice. This holds + only if you stop with `flink stop` (`flink savepoint` then cancel can re-read snapshots), + restart before `consumer.expiration-time`, and keep `consumer.ignore-progress` false. Check + that `next_snapshot_id` in `
$consumers` holds, then advances. + +4. **Do not restore from savepoints or checkpoints taken before the migration.** They still hold + the old ids. + +After the migration, topology changes around the table, such as a map in front of the sink or a +new parallelism, keep every Paimon operator's state on the routes above. 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 b71319523303..54a936bc62c8 100644 --- a/docs/generated/flink_connector_configuration.html +++ b/docs/generated/flink_connector_configuration.html @@ -302,11 +302,17 @@ + + + + + + - + @@ -398,11 +404,17 @@ + + + + + + - + 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 0deded6d31f4..bc0b23693bfc 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 @@ -25,6 +25,7 @@ import org.apache.paimon.options.ConfigOptions; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.description.DescribedEnum; +import org.apache.paimon.options.description.Description; import org.apache.paimon.options.description.InlineElement; import java.lang.reflect.Field; @@ -33,11 +34,15 @@ import java.util.List; import static org.apache.paimon.options.ConfigOptions.key; +import static org.apache.paimon.options.description.LinkElement.link; import static org.apache.paimon.options.description.TextElement.text; /** Options for flink connector. */ public class FlinkConnectorOptions { + /** Relative to the configurations page that embeds the generated option tables. */ + private static final String OPERATOR_UID_GUIDE = "../flink/savepoint#operator-uids"; + public static final String NONE = "none"; public static final String TABLE_DYNAMIC_OPTION_PREFIX = "paimon."; @@ -481,18 +486,64 @@ public class FlinkConnectorOptions { .stringType() .noDefaultValue() .withDescription( - "Set the uid suffix for the source operators. After setting, the uid format is " - + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will " - + "automatically generate the operator uid, which may be incompatible when the topology changes."); + Description.builder() + .text( + "Set the uid suffix for the source operators. The uid format is " + + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink " + + "generates these ids from the shape of the stream graph, so a topology " + + "change can orphan their state. The suffix alone does not prevent that: " + + "the other operators the table source adds keep generated ids, and a " + + "recovery from the HA checkpoint store fails when one of them moves. " + + "For stable ids also set 'source.operator-uid.cover-table-source-operators'; existing jobs must follow the %s first.", + link(OPERATOR_UID_GUIDE, "migration steps")) + .build()); + + public static final ConfigOption SOURCE_OPERATOR_UID_COVER_TABLE_SOURCE_OPERATORS = + key("source.operator-uid.cover-table-source-operators") + .booleanType() + .defaultValue(false) + .withDescription( + Description.builder() + .text( + "If true, 'source.operator-uid.suffix' also names every other operator the table source adds in a streaming job: the split monitor, the split reader, the watermark assigner and the DataStream row conversion. " + + "Without it those operators get ids generated from the shape of the " + + "stream graph, so a change elsewhere in the job orphans their checkpoint " + + "state. Has no effect unless 'source.operator-uid.suffix' is set. Enabling it on an existing " + + "job changes those ids at the next restore; follow the %s first.", + link(OPERATOR_UID_GUIDE, "migration steps")) + .build()); public static final ConfigOption SINK_OPERATOR_UID_SUFFIX = key("sink.operator-uid.suffix") .stringType() .noDefaultValue() .withDescription( - "Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is " - + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will " - + "automatically generate the operator uid, which may be incompatible when the topology changes."); + Description.builder() + .text( + "Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is " + + "${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink " + + "generates these ids from the shape of the stream graph, so a topology " + + "change can orphan their state. The suffix alone does not prevent that: " + + "the other operators the table sink adds keep generated ids, and a " + + "recovery from the HA checkpoint store fails when one of them moves. " + + "For stable ids also set 'sink.operator-uid.cover-table-sink-operators'; existing jobs must follow the %s first.", + link(OPERATOR_UID_GUIDE, "migration steps")) + .build()); + + public static final ConfigOption SINK_OPERATOR_UID_COVER_TABLE_SINK_OPERATORS = + key("sink.operator-uid.cover-table-sink-operators") + .booleanType() + .defaultValue(false) + .withDescription( + Description.builder() + .text( + "If true, 'sink.operator-uid.suffix' also names every other operator the table sink adds in a streaming job: the row conversions, the local merge, the compaction operators, the partition statistics operators, the index bootstrap and the final sink. Operators the CDC sync actions add before the table sink are not covered. " + + "Without it those operators get ids generated from the shape of the " + + "stream graph, so a change elsewhere in the job orphans their checkpoint " + + "state. Has no effect unless 'sink.operator-uid.suffix' is set. Enabling it on an existing " + + "job changes those ids at the next restore; follow the %s first.", + link(OPERATOR_UID_GUIDE, "migration steps")) + .build()); public static final ConfigOption SCAN_BOUNDED = key("scan.bounded") 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 a1fd0cd7aa25..9b57e972b960 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; @@ -81,6 +82,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; @@ -174,26 +181,34 @@ public DataStream doWrite( options.get(SINK_WRITER_MANAGED_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; @@ -237,9 +252,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( @@ -291,7 +309,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 4857d40c13a0..a6dbc1be479f 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; @@ -453,7 +457,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); } /** @@ -585,7 +589,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..ea693d6bdecb --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/OperatorUidAssigner.java @@ -0,0 +1,103 @@ +/* + * 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_TABLE_SINK_OPERATORS; +import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX; +import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_COVER_TABLE_SOURCE_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 a cover option 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_TABLE_SINK_OPERATORS, SINK_OPERATOR_UID_SUFFIX); + } + + public static OperatorUidAssigner forSource(@Nullable Table table) { + return create( + table, + SOURCE_OPERATOR_UID_COVER_TABLE_SOURCE_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..d20fc1d18319 --- /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_TABLE_SINK_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-table-sink-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_TABLE_SINK_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..578d5d86d36b --- /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_TABLE_SINK_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_TABLE_SINK_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_TABLE_SINK_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..236c1ca6f0f6 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java @@ -0,0 +1,486 @@ +/* + * 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.api.bridge.java.StreamStatementSet; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.api.config.ExecutionConfigOptions.UidGeneration; +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.EnumSource; +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_TABLE_SINK_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-table-sink-operators} every operator the table + * sink 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); + } + + /** The planner neither drops nor replaces the uids Paimon sets, whatever its uid mode. */ + @ParameterizedTest(name = "{0}") + @EnumSource( + value = UidGeneration.class, + names = {"PLAN_ONLY", "ALWAYS"}) + void testSqlInsertKeepsTheUids(UidGeneration uidGeneration) { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_UID_GENERATION, uidGeneration); + tEnv.executeSql( + String.format( + "CREATE CATALOG paimon WITH ('type' = 'paimon', 'warehouse' = '%s')", + warehouse.resolve("sql-" + uidGeneration))); + tEnv.executeSql( + String.format( + "CREATE TABLE paimon.`default`.%s (k INT, v BIGINT, pt STRING) WITH " + + "('%s' = '%s', '%s' = 'true')", + TABLE_NAME, + SINK_OPERATOR_UID_SUFFIX.key(), + UID_SUFFIX, + SINK_OPERATOR_UID_COVER_TABLE_SINK_OPERATORS.key())); + StreamStatementSet statements = tEnv.createStatementSet(); + statements.addInsertSql( + "INSERT INTO paimon.`default`." + TABLE_NAME + " VALUES (1, 2, 'a')"); + statements.attachAsDataStream(); + + assertThat(uidsByOperatorName(env.getStreamGraph())) + .as("table.exec.uid.generation=%s", uidGeneration) + .containsEntry("Writer : tbl", "Writer_tbl_test-uid") + .containsEntry("Global Committer : tbl", "Global Committer_tbl_test-uid") + .containsEntry("end: Writer", "end_tbl_test-uid"); + } + + // ------------------------------------------------------------------------ + // 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_TABLE_SINK_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/MonitorSourceUidMigrationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MonitorSourceUidMigrationITCase.java new file mode 100644 index 000000000000..a8232de9e14c --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/MonitorSourceUidMigrationITCase.java @@ -0,0 +1,100 @@ +/* + * 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.flink.CatalogITCaseBase; + +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.core.execution.SavepointFormatType; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.time.Duration; +import java.util.Arrays; +import java.util.List; + +import static org.apache.paimon.utils.CommonTestUtils.waitUtil; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Turning {@code source.operator-uid.cover-table-source-operators} on renames the Monitor, whose + * state is then dropped; on the exactly-once consumer route the consumer carries the position. + */ +class MonitorSourceUidMigrationITCase extends CatalogITCaseBase { + + @Override + protected List ddl() { + return Arrays.asList( + "CREATE TABLE src (k INT) WITH ('source.operator-uid.suffix' = 'test-uid'," + + " 'consumer-id' = 'c', 'consumer.expiration-time' = '1 d')", + "CREATE TABLE snk (k INT) WITH ('sink.operator-uid.suffix' = 'test-uid')"); + } + + @Test + @Timeout(300) + void testConsumerKeepsPositionWhenMonitorStateIsDropped() throws Exception { + insert(0, 1, 2); + JobClient client = copy("false"); + awaitRows(3); + String savepoint = + client.stopWithSavepoint( + false, getTempDirPath("savepoint"), SavepointFormatType.CANONICAL) + .get(); + + insert(3, 4, 5); + sEnv.getConfig().getConfiguration().setString("execution.state-recovery.path", savepoint); + sEnv.getConfig() + .getConfiguration() + .setString("execution.state-recovery.ignore-unclaimed-state", "true"); + client = copy("true"); + awaitRows(6); + client.cancel().get(); + + assertThat(batchSql("SELECT k FROM snk")) + .as("every snapshot read exactly once across the restore") + .containsExactlyInAnyOrder( + Row.of(0), Row.of(1), Row.of(2), Row.of(3), Row.of(4), Row.of(5)); + } + + /** One snapshot per key. */ + private void insert(int... keys) { + for (int k : keys) { + batchSql("INSERT INTO src VALUES (%d)", k); + } + } + + private JobClient copy(String cover) throws Exception { + return sEnv.executeSql( + "INSERT INTO snk SELECT k FROM src /*+ OPTIONS(" + + "'source.operator-uid.cover-table-source-operators' = '" + + cover + + "') */") + .getJobClient() + .get(); + } + + private void awaitRows(int count) throws Exception { + waitUtil( + () -> batchSql("SELECT k FROM snk").size() >= count, + Duration.ofMinutes(2), + Duration.ofMillis(500), + "the sink never reached " + count + " rows"); + } +} 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..03a818f98b3f --- /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_TABLE_SOURCE_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-table-source-operators} every operator the + * table source 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_TABLE_SOURCE_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-table-sink-operators
falseBooleanIf true, 'sink.operator-uid.suffix' also names every other operator the table sink adds in a streaming job: the row conversions, the local merge, the compaction operators, the partition statistics operators, the index bootstrap and the final sink. Operators the CDC sync actions add before the table sink are not covered. Without it those operators get ids generated from the shape of the stream graph, so a change elsewhere in the job orphans their checkpoint state. Has no effect unless 'sink.operator-uid.suffix' is set. Enabling it on an existing job changes those ids at the next restore; follow the migration steps first.
sink.operator-uid.suffix
 (none) StringSet the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will automatically generate the operator uid, which may be incompatible when the topology changes.Set the uid suffix for the writer, dynamic bucket assigner and committer operators. The uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink generates these ids from the shape of the stream graph, so a topology change can orphan their state. The suffix alone does not prevent that: the other operators the table sink adds keep generated ids, and a recovery from the HA checkpoint store fails when one of them moves. For stable ids also set 'sink.operator-uid.cover-table-sink-operators'; existing jobs must follow the migration steps first.
sink.parallelism
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-table-source-operators
falseBooleanIf true, 'source.operator-uid.suffix' also names every other operator the table source adds in a streaming job: the split monitor, the split reader, the watermark assigner and the DataStream row conversion. Without it those operators get ids generated from the shape of the stream graph, so a change elsewhere in the job orphans their checkpoint state. Has no effect unless 'source.operator-uid.suffix' is set. Enabling it on an existing job changes those ids at the next restore; follow the migration steps first.
source.operator-uid.suffix
 (none) StringSet the uid suffix for the source operators. After setting, the uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. If the uid suffix is not set, flink will automatically generate the operator uid, which may be incompatible when the topology changes.Set the uid suffix for the source operators. The uid format is ${UID_PREFIX}_${TABLE_NAME}_${USER_UID_SUFFIX}. Without a suffix Flink generates these ids from the shape of the stream graph, so a topology change can orphan their state. The suffix alone does not prevent that: the other operators the table source adds keep generated ids, and a recovery from the HA checkpoint store fails when one of them moves. For stable ids also set 'source.operator-uid.cover-table-source-operators'; existing jobs must follow the migration steps first.
unaware-bucket.compaction.parallelism