diff --git a/docs/docs/spark/structured-streaming.md b/docs/docs/spark/structured-streaming.md index cc28c1d652f5..f9929bc0b225 100644 --- a/docs/docs/spark/structured-streaming.md +++ b/docs/docs/spark/structured-streaming.md @@ -66,6 +66,56 @@ Use a durable checkpoint location accessible to the cluster for deployed jobs. G its own checkpoint directory. Stop this example with `writer.stop()`. Streaming writes also support [Schema Evolution on Write](./schema-evolution). +### Exactly-once + +Structured Streaming replays a micro-batch with its original batch id when a query is restarted +after failing between the sink writing the batch and Spark recording that batch as completed. +Paimon commits every micro-batch under a commit user that is stable across restarts, and skips a +batch that the same user already committed, so a replay does not write the data twice. Micro-batch +`n` is committed under commit identifier `n + 1`, the way Flink numbers its checkpoints, which is +what the `$snapshots` system table shows and what a `compacted-full` scan recognises a scheduled +full compaction by. + +What the commit user identifies is one incarnation of a checkpoint, since the batch ids it +numbers are only unique within one: reusing it across two checkpoints would make Paimon skip the +data of the second one, while changing it within one query would bring the duplicate back. It is +therefore derived from the query id that Spark persists in the checkpoint, which is new when a +checkpoint is recreated, unchanged when a query resumes from one, and independent of how the +location is spelled, and it cannot be set explicitly. `commit.user-prefix` is read from the table as +usual and prefixes the derived user, so a job keeps the name its table was configured with. + +:::note + +A skipped replay leaves the data files it wrote behind, uncommitted. They are removed by +[orphan file cleaning](../maintenance/manage-snapshots#remove-orphan-files), like any other +uncommitted file. + +A query that starts from a new checkpoint gets a new commit user, so a micro-batch the previous +run committed is not recognised and its data is written again. + +A replay is recognised by the snapshots its commit user left behind, which snapshot expiration +eventually removes. Before committing a micro-batch, the sink records under the checkpoint location +of the query which snapshot was the latest before its first attempt. Retries preserve this original +marker rather than advancing it. If filtering cannot recognise the replay and expiration has removed +the snapshot immediately after that boundary, the query fails instead of possibly writing it twice. +An incomplete or corrupt marker also fails the query, since it may belong to an already committed +batch. The error names the marker file; restore it or delete it only after verifying that the batch +was not committed. Retain snapshots for longer than a query may be down. + +A streaming write to a postpone bucket table writes the postpone bucket, as a Flink streaming job +does, whatever `postpone.batch-write-fixed-bucket` says: that option is for a batch job which ends +with its commit. The rows become readable once a +[compaction](../primary-key-table/data-distribution#postpone-bucket) has sorted them into real +buckets. + +::: + +The committer of a query is created with its first micro-batch and lives until the query +terminates, again as a Flink committer lives across checkpoints. Only the first micro-batch of a +run looks up what a previous run committed; the maintenance a commit schedules, such as tag +creation and snapshot or partition expiration, runs while the next micro-batch is written when +`snapshot.expire.execution-mode` is `async`. + ## Streaming Query :::info diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index c8719aae32de..97bf697f56c7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -442,7 +442,7 @@ private List createCommitCallbacks(String commitUser, FileStoreT } if (options.isChainTable()) { - callbacks.add(new ChainTableOverwriteCommitCallback(table)); + callbacks.add(new ChainTableOverwriteCommitCallback(table, commitUser)); } if (options.visibilityCallbackEnabled() && shouldWaitForVisibility(table)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java index 67255bd7f402..07ee85117e68 100644 --- a/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java @@ -19,22 +19,43 @@ package org.apache.paimon.metastore; import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.manifest.FileEntry; +import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.ManifestCommittable; import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.operation.FileStoreCommit; +import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.CommitCallback; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.source.ScanMode; import org.apache.paimon.utils.ChainTableUtils; import org.apache.paimon.utils.InternalRowPartitionComputer; +import org.apache.paimon.utils.SnapshotManager; +import org.apache.paimon.utils.Triple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +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.stream.Collectors; +import static org.apache.paimon.CoreOptions.createCommitUser; + /** * A {@link CommitCallback} implementation to maintain chain table snapshot branch for overwrite * commits. @@ -52,52 +73,341 @@ */ public class ChainTableOverwriteCommitCallback implements CommitCallback { + private static final Logger LOG = + LoggerFactory.getLogger(ChainTableOverwriteCommitCallback.class); + + /** + * The latest snapshot of the snapshot branch when an overwrite of the delta branch was + * published, 0 if it had none, kept in the properties of the overwrite's snapshot. + */ + public static final String SNAPSHOT_BRANCH_POSITION = "chain-table.snapshot-branch.snapshot-id"; + private transient FileStoreTable table; private transient CoreOptions coreOptions; + private final String commitUser; - public ChainTableOverwriteCommitCallback(FileStoreTable table) { + public ChainTableOverwriteCommitCallback(FileStoreTable table, String commitUser) { this.table = table; this.coreOptions = table.coreOptions(); + this.commitUser = commitUser; + } + + /** + * Records where the snapshot branch stands as the overwrite is published: what the overwrite + * supersedes there are the files that snapshot holds, and a {@link #retry} of the overwrite has + * no other exact way to tell them from what the snapshot branch receives afterwards, since the + * two branches commit independently and their clocks cannot order one against the other. + * + *

A snapshot branch that cannot be read fails the overwrite before it is published. + */ + @Override + public void beforeOverwrite(ManifestCommittable committable) { + if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) { + return; + } + Long latest = snapshotTable().snapshotManager().latestSnapshotId(); + committable.addProperty( + SNAPSHOT_BRANCH_POSITION, String.valueOf(latest == null ? 0L : latest)); } @Override public void call(Context context) { + if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) { + return; + } + if (context.snapshot.commitKind() != CommitKind.OVERWRITE) { + return; + } + truncateSnapshotPartitions(context.deltaFiles); + } + /** + * The commit of this committable was published by an earlier attempt whose callback may not + * have completed, for example because the snapshot branch was unreachable right after the delta + * snapshot was written. Resolve that snapshot and redo the cleanup, which is idempotent. The + * partitions are taken from the manifest changes of the snapshot rather than from the + * committable, since an overwrite also clears partitions it wrote no new file to. + */ + @Override + public void retry(ManifestCommittable committable) { if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) { return; } + List snapshots = + table.snapshotManager() + .findSnapshotsForIdentifiers( + commitUser, Collections.singletonList(committable.identifier())); + if (snapshots.isEmpty()) { + LOG.warn( + "No snapshot of commit user {} with identifier {} in table {}, " + + "cannot redo the snapshot branch cleanup of its overwrite.", + commitUser, + committable.identifier(), + table.name()); + return; + } + for (Snapshot snapshot : snapshots) { + if (snapshot.commitKind() != CommitKind.OVERWRITE) { + continue; + } + List overwritePartitions = + overwritePartitions( + table.store() + .newScan() + .withKind(ScanMode.DELTA) + .withSnapshot(snapshot.id()) + .plan() + .files()); + clearSnapshotFilesAsOf(overwritePartitions, snapshot); + } + } - if (context.snapshot.commitKind() != CommitKind.OVERWRITE) { + /** + * Clear what the overwrite superseded in the given partitions of the snapshot branch, and + * nothing that landed there since. That is what {@link #call} cleared at the time; a replay + * that repeats it after later data arrived must not take that data with it, whether the + * original cleanup had completed or not. + * + *

What the overwrite superseded are the files the snapshot branch held in those partitions + * when the overwrite was published, and whatever compactions of the snapshot branch have since + * rewritten from them alone. If a compaction has merged them with data written after the + * overwrite, another commit such as a rescale has rewritten them, or the snapshot branch no + * longer retains the history to tell, the cleanup cannot be done exactly, and the retry fails + * rather than report a cleanup it did not do. + */ + private void clearSnapshotFilesAsOf(List partitions, Snapshot overwrite) { + if (partitions.isEmpty()) { return; } + String position = + overwrite.properties() == null + ? null + : overwrite.properties().get(SNAPSHOT_BRANCH_POSITION); + if (position == null) { + throw new IllegalStateException( + String.format( + "Cannot redo the snapshot branch cleanup of the overwrite of partitions " + + "%s in table %s (snapshot %s): the overwrite did not record " + + "where the snapshot branch stood when it was published, so " + + "what it superseded cannot be told from what was written " + + "after it. Clear the superseded rows of these partitions " + + "manually.", + partitions, table.name(), overwrite.id())); + } + long asOfId = Long.parseLong(position); + if (asOfId == 0) { + // The snapshot branch had no snapshot when the overwrite was published. + return; + } + FileStoreTable snapshotTable = snapshotTable(); + SnapshotManager snapshotManager = snapshotTable.snapshotManager(); + if (!snapshotManager.snapshotExists(asOfId)) { + throw new IllegalStateException( + String.format( + "Cannot redo the snapshot branch cleanup of the overwrite of partitions " + + "%s in table %s (snapshot %s): the snapshot branch no longer " + + "retains snapshot %s, where it stood when that overwrite was " + + "published, so what the overwrite superseded cannot be told " + + "from what was written after it. Retain snapshots of the " + + "snapshot branch for longer than a job may take to recover, " + + "and clear the superseded rows of these partitions manually.", + partitions, table.name(), overwrite.id(), asOfId)); + } + Snapshot asOf = snapshotManager.snapshot(asOfId); + Snapshot latest = snapshotManager.latestSnapshot(); + // Follow the superseded files through the rewrites of the snapshot branch since then. + Set superseded = new HashSet<>(); + for (ManifestEntry entry : filesOf(snapshotTable, asOf, partitions)) { + superseded.add(entry.identifier()); + } + Set mixed = new HashSet<>(); + for (long id = asOf.id() + 1; id <= latest.id(); id++) { + boolean compaction = snapshotManager.snapshot(id).commitKind() == CommitKind.COMPACT; + Map> changes = + snapshotTable.store().newScan().withKind(ScanMode.DELTA).withSnapshot(id) + .withPartitionFilter(partitions).plan().files().stream() + .collect(Collectors.groupingBy(ManifestEntry::partition)); + for (List partitionChanges : changes.values()) { + if (compaction) { + // A compaction rewrites the files of one bucket into that bucket: its + // outputs hold superseded rows only if all its inputs did. + partitionChanges.stream() + .collect(Collectors.groupingBy(ManifestEntry::bucket)) + .values() + .forEach(bucketChanges -> follow(bucketChanges, superseded, mixed)); + } else if (removesTracked(partitionChanges, superseded, mixed)) { + // Any other commit that removes superseded rows and adds files to the + // partition may have written them again, the way a rescale rewrites a + // partition into other buckets; nothing tells them from new data. + for (ManifestEntry entry : partitionChanges) { + if (entry.kind() == FileKind.ADD) { + mixed.add(entry.identifier()); + } + } + } + // Anything else a commit adds is data written after the overwrite. + } + } + + List current = filesOf(snapshotTable, latest, partitions); + List inseparable = + current.stream() + .filter(entry -> mixed.contains(entry.identifier())) + .map(entry -> entry.file().fileName()) + .collect(Collectors.toList()); + if (!inseparable.isEmpty()) { + throw new IllegalStateException( + String.format( + "Cannot redo the snapshot branch cleanup of the overwrite of partitions " + + "%s in table %s (snapshot %s): the snapshot branch has since " + + "rewritten the rows that overwrite superseded into files %s, " + + "either merged with rows written after it by a compaction or " + + "by an overwrite such as a rescale, and they can no longer " + + "be told apart. Clear the superseded rows of these " + + "partitions manually.", + partitions, table.name(), overwrite.id(), inseparable)); + } + Map, List> toClear = new HashMap<>(); + for (ManifestEntry entry : current) { + if (superseded.contains(entry.identifier())) { + toClear.computeIfAbsent( + Triple.of( + entry.partition().copy(), + entry.bucket(), + entry.totalBuckets()), + k -> new ArrayList<>()) + .add(entry.file()); + } + } + if (toClear.isEmpty()) { + return; + } + ManifestCommittable committable = + new ManifestCommittable(BatchWriteBuilder.COMMIT_IDENTIFIER); + toClear.forEach( + (key, files) -> + committable.addFileCommittable( + new CommitMessageImpl( + key.f0, + key.f1, + key.f2, + new DataIncrement( + Collections.emptyList(), + files, + Collections.emptyList()), + CompactIncrement.emptyIncrement()))); + try (FileStoreCommit commit = + snapshotTable + .store() + .newCommit( + createCommitUser(new Options(snapshotTable.options())), + snapshotTable)) { + // The cleared partitions are the ones the overwrite rewrote on the delta branch, as + // in the original cleanup, so the pre-callback must not reject dropping their + // baselines on a retry either. + Set previous = + ChainTableOverwriteScope.setFreshlyWrittenDeltaPartitions( + new HashSet<>(partitions)); + try { + // The files being removed must still be there; a concurrent change to them is a + // conflict to report, not to skip over. + commit.commit(committable, true); + } finally { + ChainTableOverwriteScope.restore(previous); + } + } catch (Exception e) { + throw new RuntimeException( + String.format( + "Failed to clear the files of partitions %s in the snapshot table.", + partitions), + e); + } + } + + /** Follows the files one bucket of a compaction removes into the files it adds. */ + private static void follow( + List bucketChanges, + Set superseded, + Set mixed) { + boolean fromSuperseded = false; + boolean fromOther = false; + for (ManifestEntry entry : bucketChanges) { + if (entry.kind() == FileKind.DELETE) { + FileEntry.Identifier identifier = entry.identifier(); + if (superseded.remove(identifier)) { + fromSuperseded = true; + } else if (mixed.remove(identifier)) { + fromSuperseded = true; + fromOther = true; + } else { + fromOther = true; + } + } + } + if (!fromSuperseded) { + return; + } + for (ManifestEntry entry : bucketChanges) { + if (entry.kind() == FileKind.ADD) { + (fromOther ? mixed : superseded).add(entry.identifier()); + } + } + } + + /** Stops tracking the files a commit removes, and returns whether any of them was tracked. */ + private static boolean removesTracked( + List changes, + Set superseded, + Set mixed) { + boolean removed = false; + for (ManifestEntry entry : changes) { + if (entry.kind() == FileKind.DELETE) { + FileEntry.Identifier identifier = entry.identifier(); + removed |= superseded.remove(identifier) | mixed.remove(identifier); + } + } + return removed; + } + + private static List filesOf( + FileStoreTable table, Snapshot snapshot, List partitions) { + return table.store() + .newScan() + .withSnapshot(snapshot) + .withPartitionFilter(partitions) + .plan() + .files(); + } + + private FileStoreTable snapshotTable() { FileStoreTable candidateTable = ChainTableUtils.resolveChainPrimaryTable(table); + return candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch()); + } - FileStoreTable snapshotTable = - candidateTable.switchToBranch(coreOptions.scanFallbackSnapshotBranch()); + private static List overwritePartitions(List deltaFiles) { + return deltaFiles.stream() + .map(ManifestEntry::partition) + .distinct() + .collect(Collectors.toList()); + } + private void truncateSnapshotPartitions(List deltaFiles) { + List overwritePartitions = overwritePartitions(deltaFiles); + if (overwritePartitions.isEmpty()) { + return; + } InternalRowPartitionComputer partitionComputer = new InternalRowPartitionComputer( coreOptions.partitionDefaultName(), table.schema().logicalPartitionType(), table.schema().partitionKeys().toArray(new String[0]), coreOptions.legacyPartitionName()); - - List overwritePartitions = - context.deltaFiles.stream() - .map(ManifestEntry::partition) - .distinct() - .collect(Collectors.toList()); - - if (overwritePartitions.isEmpty()) { - return; - } - List> candidatePartitions = overwritePartitions.stream() .map(partitionComputer::generatePartValues) .collect(Collectors.toList()); - + FileStoreTable snapshotTable = snapshotTable(); try (BatchTableCommit commit = snapshotTable.newBatchWriteBuilder().newCommit()) { // The truncated snapshot partitions are exactly the partitions this overwrite just // rewrote on the delta branch, so their surviving delta followers hold fresh data @@ -120,12 +430,6 @@ public void call(Context context) { } } - @Override - public void retry(ManifestCommittable committable) { - // No-op. Truncating the same partitions again is safe, but we prefer to only rely on the - // successful commit callback. - } - @Override public void close() throws Exception { // no resources to close diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index e518c8f0bc31..c31f2ce7f1b3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -494,6 +494,7 @@ private int overwritePartition( "Ready to overwrite to table {}, number of commit messages: {}", tableName, committable.fileCommittables().size()); + commitCallbacks.forEach(callback -> callback.beforeOverwrite(committable)); long started = System.nanoTime(); int generatedSnapshot = 0; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitCallback.java index 01ef58ef0926..0ce7e3ed23af 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitCallback.java @@ -44,6 +44,13 @@ public interface CommitCallback extends AutoCloseable { void retry(ManifestCommittable committable); + /** + * Called before an overwrite is committed, with the committable it is about to publish. A + * callback may add properties to it, which the snapshot of the overwrite then carries, to + * record what only the moment of the overwrite knows and a later {@link #retry} of it needs. + */ + default void beforeOverwrite(ManifestCommittable committable) {} + default void setTable(FileStoreTable table) {} /** Context for callback. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java index f5c51a74735b..6971e2642ee0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java @@ -29,6 +29,13 @@ /** Inner {@link TableCommit} contains overwrite setter. */ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit { + /** + * Filters committed identifiers, then validates the remaining commits before publishing any of + * them. The check is not run for an already committed replay; its retry callbacks still run. + */ + int filterAndCommit( + Map> commitIdentifiersAndMessages, Runnable beforeCommit); + /** Overwrite writing, same as the 'INSERT OVERWRITE T PARTITION (...)' semantics of SQL. */ InnerTableCommit withOverwrite(@Nullable Map spec); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java index 23f1132fd795..ed0d7665c143 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java @@ -255,10 +255,18 @@ public void commit(long identifier, List commitMessages) { @Override public int filterAndCommit(Map> commitIdentifiersAndMessages) { + return filterAndCommit(commitIdentifiersAndMessages, () -> {}); + } + + @Override + public int filterAndCommit( + Map> commitIdentifiersAndMessages, Runnable beforeCommit) { return filterAndCommitMultiple( commitIdentifiersAndMessages.entrySet().stream() .map(e -> createManifestCommittable(e.getKey(), e.getValue())) - .collect(Collectors.toList())); + .collect(Collectors.toList()), + true, + beforeCommit); } private ManifestCommittable createManifestCommittable( @@ -318,6 +326,13 @@ public int filterAndCommitMultiple(List committables) { public int filterAndCommitMultiple( List committables, boolean checkAppendFiles) { + return filterAndCommitMultiple(committables, checkAppendFiles, () -> {}); + } + + private int filterAndCommitMultiple( + List committables, + boolean checkAppendFiles, + Runnable beforeCommit) { List sortedCommittables = committables.stream() // identifier must be in increasing order @@ -326,6 +341,7 @@ public int filterAndCommitMultiple( List retryCommittables = commit.filterCommitted(sortedCommittables); if (!retryCommittables.isEmpty()) { + beforeCommit.run(); checkFilesExistence(retryCommittables); commitMultiple(retryCommittables, checkAppendFiles); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java index a651cfc2c9d1..eaf3f8a086f8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java @@ -19,9 +19,11 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.catalog.TableQueryAuthResult; +import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -30,6 +32,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.metastore.ChainTableOverwriteCommitCallback; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.CastTransform; @@ -44,9 +47,15 @@ import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.InnerTableCommit; import org.apache.paimon.table.sink.InnerTableWrite; import org.apache.paimon.table.sink.StreamTableCommit; +import org.apache.paimon.table.sink.StreamTableWrite; +import org.apache.paimon.table.sink.StreamWriteBuilder; import org.apache.paimon.table.source.ChainSplit; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.DataTableScan; @@ -63,6 +72,7 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap; +import org.assertj.core.api.ThrowableAssert; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -72,6 +82,8 @@ import org.junit.jupiter.params.provider.ValueSource; import org.mockito.Mockito; +import java.io.File; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1086,6 +1098,577 @@ private List readSplits(FileStoreTable table, List splits) th return result; } + @Test + public void testChainOverwriteClearsSnapshotPartition() throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20")); + + // An overwrite of the delta branch clears the same partition of the snapshot branch, so + // that reads fall through to the delta. + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayCompletesSnapshotCleanup() throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20")); + + // The overwrite publishes its snapshot, and only then clears the snapshot branch. Make + // that cleanup fail, the way a transient failure of the snapshot branch would: the delta + // snapshot is published, the commit throws, and the partition is not cleared. + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "value-2", "CN", "20250810", "20"))); + assertThat(deltaTable.snapshotManager().snapshotCount()) + .as("the delta snapshot was published before the cleanup failed") + .isEqualTo(1); + assertThat(getResult(loadTable(), partition)) + .as("the snapshot branch still hides the delta until its partition is cleared") + .containsExactly(row(1L, 1L, "value-1", "CN", "20250810", "20")); + + // A restarted job replays the batch under the same user and identifier. The commit is + // recognised as already published, so the callback is retried rather than called, and + // the retry has to complete the cleanup the first attempt did not. + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20")); + + assertThat(deltaTable.snapshotManager().snapshotCount()) + .as("the replay must not publish a second snapshot") + .isEqualTo(1); + assertThat(getResult(loadTable(), partition)) + .as("the replay must complete the cleanup of the snapshot branch") + .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayClearsPartitionWithOnlyRemovedFiles() throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + + writeWithCommit( + snapshotTable, + row(1L, 1L, "value-1", "CN", "20250810", "20"), + row(2L, 1L, "value-1", "CN", "20250810", "21")); + writeWithCommit(deltaTable, row(1L, 1L, "delta-old", "CN", "20250810", "20")); + + // A static overwrite of the whole delta branch that writes only to hour 21 removes the + // file of hour 20 without adding one there. The cleanup has to cover hour 20 as well, + // which the commit messages of the batch do not mention; only the manifest changes of the + // snapshot do. (A dynamic partition overwrite, the default, would leave hour 20 alone.) + Map staticOverwrite = new HashMap<>(); + staticOverwrite.put(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), "false"); + FileStoreTable staticDelta = deltaTable.copy(staticOverwrite); + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + staticDelta, + Collections.emptyMap(), + 0, + row(2L, 2L, "value-2", "CN", "20250810", "21"))); + commitOverwrite( + staticDelta, + Collections.emptyMap(), + 0, + row(2L, 2L, "value-2", "CN", "20250810", "21")); + + assertThat(deltaTable.snapshotManager().snapshotCount()).isEqualTo(2); + assertThat(getResult(loadTable(), partition("20"))) + .as("a partition the overwrite only removed files from must be cleared too") + .isEmpty(); + assertThat(getResult(loadTable(), partition("21"))) + .containsExactly(row(2L, 2L, "value-2", "CN", "20250810", "21")); + } + + @Test + public void testChainOverwriteReplayKeepsSnapshotDataWrittenAfterTheOverwrite() + throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20")); + // The overwrite and its cleanup both succeed. + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20")); + + // New data lands in the same partition of the snapshot branch afterwards and, being the + // snapshot, takes precedence over the delta. + writeWithCommit(snapshotTable, row(2L, 1L, "value-3", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20")); + + // The batch is replayed although its callback had completed, for example because the + // engine failed after the sink returned. The retry must not clear what landed later. + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20")); + + assertThat(getResult(snapshotTable, partition)) + .as( + "data written to the snapshot branch after the overwrite must survive its replay") + .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayClearsOnlyWhatTheOverwriteSuperseded() throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "value-1", "CN", "20250810", "20")); + // The overwrite is published, its cleanup fails. + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "value-2", "CN", "20250810", "20"))); + + // New data lands in the same partition of the snapshot branch before the replay. + writeWithCommit(snapshotTable, row(2L, 1L, "value-3", "CN", "20250810", "20")); + + // Had the cleanup run when the overwrite was published, it would have cleared value-1 + // and left value-3, which came later, in place. The retry has to do the same. + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "value-2", "CN", "20250810", "20")); + + assertThat(getResult(snapshotTable, partition)) + .as( + "the retry must clear what the overwrite superseded and nothing that came later") + .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20")); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(2L, 1L, "value-3", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayAfterSnapshotBranchCompaction() throws Exception { + // Reviewer P1 on #10105: the retry clears the files the snapshot branch held when the + // overwrite was published, identified by file. A compaction of the snapshot branch in + // between rewrites those files, so none of them is found any more. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + writeWithCommit(snapshotTable, row(3L, 1L, "old-2", "CN", "20250810", "20")); + + // The overwrite is published, its cleanup fails. + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + + // The snapshot branch compacts the partition before the replay. + List before = snapshotFileNames(snapshotTable); + compactSnapshotPartition(snapshotTable); + List after = snapshotFileNames(snapshotTable); + assertThat(after) + .as("the compaction rewrote every old file") + .doesNotContainAnyElementsOf(before); + + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "new", "CN", "20250810", "20")); + + assertThat(getResult(loadTable(), partition)) + .as("the replay must still clear what the overwrite superseded") + .containsExactly(row(1L, 2L, "new", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayAfterOverwriteTimeSnapshotExpired() throws Exception { + // Reviewer P1 on #10105: the retry resolves what the snapshot branch held when the + // overwrite was published from the snapshot of that time. Once that snapshot has + // expired, the retry gives up silently although the old rows are still there. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + + // The overwrite is published, its cleanup fails. + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + + // Later data lands in the snapshot branch, and the snapshot of the overwrite's time + // expires. + Thread.sleep(10); + writeWithCommit(snapshotTable, row(2L, 1L, "later", "CN", "20250810", "20")); + Map retainOne = new HashMap<>(); + retainOne.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); + retainOne.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1"); + FileStoreTable expiring = snapshotTable.copy(retainOne); + expiring.newExpireSnapshots().config(expiring.coreOptions().expireConfig()).expire(); + assertThat(snapshotTable.snapshotManager().snapshotCount()) + .as("only the snapshot with the later data is retained") + .isEqualTo(1); + + // What the overwrite superseded can no longer be told from what came after it, so the + // replay must fail rather than report a cleanup it did not do. + assertThatThrownBy( + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))) + .hasStackTraceContaining("no longer retains snapshot"); + assertThat(deltaTable.snapshotManager().snapshotCount()).isEqualTo(1); + assertThat(getResult(snapshotTable, partition)) + .containsExactlyInAnyOrder( + row(1L, 1L, "old", "CN", "20250810", "20"), + row(2L, 1L, "later", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayRefusesWhenCompactionMergedLaterData() throws Exception { + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + + // Later data lands in the partition, and a compaction merges it with the old rows into + // the same file: which rows the overwrite superseded can no longer be told. + writeWithCommit(snapshotTable, row(2L, 1L, "later", "CN", "20250810", "20")); + compactSnapshotPartition(snapshotTable); + + assertThatThrownBy( + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))) + .hasStackTraceContaining("can no longer be told apart"); + assertThat(getResult(snapshotTable, partition)) + .containsExactlyInAnyOrder( + row(1L, 1L, "old", "CN", "20250810", "20"), + row(2L, 1L, "later", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayRefusesAfterSnapshotBranchRewrite() throws Exception { + // An overwrite of the snapshot branch that writes the old rows again, the way a rescale + // rewrites a partition into new files, leaves the superseded rows in files that are not + // compaction outputs. Nothing tells them from data written after the overwrite, so the + // retry must not report the cleanup as done. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + + BatchWriteBuilder rewrite = snapshotTable.newBatchWriteBuilder().withOverwrite(partition); + try (BatchTableWrite write = + rewrite.newWrite().withIOManager(new IOManagerImpl(tempDir.toString())); + BatchTableCommit commit = rewrite.newCommit()) { + write.write(row(1L, 1L, "old", "CN", "20250810", "20")); + commit.commit(write.prepareCommit()); + } + assertThat(snapshotTable.snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(Snapshot.CommitKind.OVERWRITE); + + assertThatThrownBy( + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))) + .hasStackTraceContaining("Cannot redo the snapshot branch cleanup"); + assertThat(getResult(snapshotTable, partition)) + .containsExactly(row(1L, 1L, "old", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayWithSnapshotBranchCommitAtTheSameTime() throws Exception { + // Reviewer P1 on #10105: the retry finds the snapshot branch as of the overwrite by the + // overwrite's commit time. The two branches have their own clocks and histories, so a + // snapshot branch commit made after the overwrite can carry the same time and be taken + // for the state the overwrite superseded. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + long overwriteMillis = deltaTable.snapshotManager().latestSnapshot().timeMillis(); + + // New data lands in the snapshot branch after the overwrite, recorded at the same + // millisecond, as a commit on another host or within the same millisecond can be. + writeWithCommit(snapshotTable, row(2L, 1L, "later", "CN", "20250810", "20")); + Snapshot later = snapshotTable.snapshotManager().latestSnapshot(); + snapshotTable + .fileIO() + .overwriteFileUtf8( + snapshotTable.snapshotManager().snapshotPath(later.id()), + later.toJson() + .replaceFirst( + "\"timeMillis\"\\s*:\\s*\\d+", + "\"timeMillis\" : " + overwriteMillis)); + assertThat( + loadTable() + .switchToBranch(SNAPSHOT_BRANCH) + .snapshotManager() + .latestSnapshot() + .timeMillis()) + .isEqualTo(overwriteMillis); + + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "new", "CN", "20250810", "20")); + + assertThat(getResult(loadTable().switchToBranch(SNAPSHOT_BRANCH), partition)) + .as( + "the retry must clear what the overwrite superseded and nothing that came later") + .containsExactly(row(2L, 1L, "later", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayRefusesWithoutRecordedPosition() throws Exception { + // An overwrite published before its position on the snapshot branch was recorded leaves + // the retry nothing exact to go by; it must not fall back to guessing. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + writeWithCommit(snapshotTable, row(1L, 1L, "old", "CN", "20250810", "20")); + assertCleanupFails( + snapshotTable, + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))); + Snapshot overwrite = deltaTable.snapshotManager().latestSnapshot(); + assertThat(overwrite.properties()) + .containsKey(ChainTableOverwriteCommitCallback.SNAPSHOT_BRANCH_POSITION); + deltaTable + .fileIO() + .overwriteFileUtf8( + deltaTable.snapshotManager().snapshotPath(overwrite.id()), + overwrite + .toJson() + .replace( + ChainTableOverwriteCommitCallback.SNAPSHOT_BRANCH_POSITION, + "unrelated")); + + assertThatThrownBy( + () -> + commitOverwrite( + deltaTable, + partition, + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))) + .hasStackTraceContaining("did not record where the snapshot branch stood"); + assertThat(getResult(snapshotTable, partition)) + .containsExactly(row(1L, 1L, "old", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteReplayWithEmptySnapshotBranchKeepsLaterData() throws Exception { + // The snapshot branch had nothing when the overwrite was published, so the overwrite + // superseded nothing there, and whatever it holds by the replay came later. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable snapshotTable = chainTable.switchToBranch(SNAPSHOT_BRANCH); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "new", "CN", "20250810", "20")); + assertThat( + deltaTable + .snapshotManager() + .latestSnapshot() + .properties() + .get(ChainTableOverwriteCommitCallback.SNAPSHOT_BRANCH_POSITION)) + .isEqualTo("0"); + writeWithCommit(snapshotTable, row(2L, 1L, "later", "CN", "20250810", "20")); + + commitOverwrite(deltaTable, partition, 0, row(1L, 2L, "new", "CN", "20250810", "20")); + + assertThat(getResult(snapshotTable, partition)) + .containsExactly(row(2L, 1L, "later", "CN", "20250810", "20")); + } + + @Test + public void testChainOverwriteFailsBeforePublishingWhenSnapshotBranchIsUnreadable() + throws Exception { + // Where the snapshot branch stands is read before the overwrite is published; if it + // cannot be, nothing is published, and there is no cleanup left for a replay to redo. + createChainTable(options -> {}); + FileStoreTable chainTable = loadTable(); + FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH); + Map unreadable = new HashMap<>(); + unreadable.put(CoreOptions.SCAN_FALLBACK_SNAPSHOT_BRANCH.key(), "unreachable"); + // Snapshot expiration of a delta branch resolves the snapshot branch as the committer is + // created; keep it off so that the failure is the one of the overwrite itself. + unreadable.put(CoreOptions.WRITE_ONLY.key(), "true"); + + assertThatThrownBy( + () -> + commitOverwrite( + deltaTable.copy(unreadable), + partition("20"), + 0, + row(1L, 2L, "new", "CN", "20250810", "20"))) + .hasStackTraceContaining("does not exist"); + assertThat(deltaTable.snapshotManager().latestSnapshot()).isNull(); + } + + private void compactSnapshotPartition(FileStoreTable snapshotTable) throws Exception { + BinaryRow partitionRow = + snapshotTable.newSnapshotReader().read().dataSplits().get(0).partition(); + try (InnerTableWrite write = + snapshotTable + .newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString()))) { + write.compact(partitionRow, 0, true); + try (StreamTableCommit commit = snapshotTable.newCommit(commitUser)) { + commit.commit(1, write.prepareCommit(true, 1)); + } + } + } + + private static List snapshotFileNames(FileStoreTable table) { + return table.newSnapshotReader().read().dataSplits().stream() + .flatMap(split -> split.dataFiles().stream()) + .map(DataFileMeta::fileName) + .collect(Collectors.toList()); + } + + @Test + public void testChainOverwriteReplayWithoutSnapshotIsSkipped() throws Exception { + createChainTable(options -> {}); + FileStoreTable deltaTable = loadTable().switchToBranch(DELTA_BRANCH); + Map partition = partition("20"); + + commitOverwrite(deltaTable, partition, 0, row(1L, 1L, "value-1", "CN", "20250810", "20")); + commitOverwrite(deltaTable, partition, 2, row(1L, 2L, "value-2", "CN", "20250810", "20")); + long snapshots = deltaTable.snapshotManager().snapshotCount(); + + // Identifier 1 is below the latest committed one, so it is treated as a replay and + // retried, but no snapshot carries it. That is not an error: nothing is left to clean. + commitOverwrite(deltaTable, partition, 1, row(1L, 3L, "value-3", "CN", "20250810", "20")); + + assertThat(deltaTable.snapshotManager().snapshotCount()).isEqualTo(snapshots); + assertThat(getResult(loadTable(), partition)) + .containsExactly(row(1L, 2L, "value-2", "CN", "20250810", "20")); + } + + private static Map partition(String hour) { + return ImmutableMap.of("region", "CN", "dt", "20250810", "hour", hour); + } + + /** + * Runs an overwrite of the delta branch while the snapshot branch cannot be written, the way a + * transient failure of its storage would: the overwrite reads where the snapshot branch stands + * and publishes its delta snapshot, and only its cleanup of the snapshot branch fails. + */ + private static void assertCleanupFails( + FileStoreTable snapshotTable, ThrowableAssert.ThrowingCallable overwrite) { + File snapshotDirectory = + new File(snapshotTable.snapshotManager().snapshotDirectory().toUri().getPath()); + assertThat(snapshotDirectory.setWritable(false)).isTrue(); + try { + assertThatThrownBy(overwrite) + .hasStackTraceContaining("Failed to truncate partitions in snapshot table"); + } finally { + assertThat(snapshotDirectory.setWritable(true)).isTrue(); + } + } + + private void commitOverwrite( + FileStoreTable deltaTable, + Map overwrite, + long identifier, + GenericRow... rows) + throws Exception { + StreamWriteBuilder builder = deltaTable.newStreamWriteBuilder().withCommitUser("user"); + try (StreamTableWrite write = + (StreamTableWrite) + builder.newWrite() + .withIOManager(new IOManagerImpl(tempDir.toString())); + InnerTableCommit commit = (InnerTableCommit) builder.newCommit()) { + commit.withOverwrite(overwrite); + for (GenericRow r : rows) { + write.write(r); + } + commit.filterAndCommit( + Collections.singletonMap(identifier, write.prepareCommit(true, identifier))); + } + } + private FileStoreTable loadTable() { return loadTable(CatalogEnvironment.empty()); } @@ -1117,6 +1700,9 @@ private void createChainTable(Consumer optionCustomizer) throws Excepti options.set(PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd HH:mm:ss"); options.set(CHAIN_TABLE_CHAIN_PARTITION_KEYS, "dt,hour"); options.set(PATH, tablePath.toString()); + // A commit that cannot write retries with backoff; the tests that make the snapshot + // branch unwritable need it to give up quickly. + options.set(CoreOptions.COMMIT_MAX_RETRY_WAIT, Duration.ofMillis(10)); optionCustomizer.accept(options); Schema schema = @@ -1168,6 +1754,9 @@ private void createPartialUpdateChainTable(Consumer optionCustomizer) options.set(PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd"); options.set(CHAIN_TABLE_CHAIN_PARTITION_KEYS, "dt"); options.set(PATH, tablePath.toString()); + // A commit that cannot write retries with backoff; the tests that make the snapshot + // branch unwritable need it to give up quickly. + options.set(CoreOptions.COMMIT_MAX_RETRY_WAIT, Duration.ofMillis(10)); optionCustomizer.accept(options); Schema schema = diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java index f768753d7fe7..f011b8a37826 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/SimpleTableTestBase.java @@ -1594,6 +1594,44 @@ public void testBatchWriteAsyncExpireFallbackToSync() throws Exception { } } + @Test + public void testFilterAndCommitAcrossRunsOfOneCommitUser() throws Exception { + FileStoreTable table = createFileStoreTable(conf -> {}); + SnapshotManager sm = table.snapshotManager(); + + // A run of a streaming job, committing two identifiers through one committer, the way a + // Spark micro-batch sink and a Flink checkpoint do. + StreamWriteBuilder first = table.newStreamWriteBuilder().withCommitUser("user"); + try (StreamTableWrite write = first.newWrite(); + StreamTableCommit commit = first.newCommit()) { + for (long identifier = 1; identifier <= 2; identifier++) { + write.write(rowData((int) identifier, (int) identifier * 10, identifier * 100L)); + commit.filterAndCommit( + Collections.singletonMap( + identifier, write.prepareCommit(true, identifier))); + } + } + long committed = sm.latestSnapshotId(); + assertThat(committed).isEqualTo(2); + + // The job is restarted and replays the identifier it committed last, then commits the one + // after it. Only the second one produces a snapshot. + StreamWriteBuilder replay = table.newStreamWriteBuilder().withCommitUser("user"); + try (StreamTableWrite write = replay.newWrite(); + StreamTableCommit commit = replay.newCommit()) { + write.write(rowData(2, 20, 200L)); + commit.filterAndCommit(Collections.singletonMap(2L, write.prepareCommit(true, 2L))); + assertThat(sm.latestSnapshotId()) + .as("a replayed identifier must not be committed again") + .isEqualTo(committed); + + write.write(rowData(3, 30, 300L)); + commit.filterAndCommit(Collections.singletonMap(3L, write.prepareCommit(true, 3L))); + } + assertThat(sm.latestSnapshotId()).isEqualTo(committed + 1); + assertThat(sm.latestSnapshot().commitIdentifier()).isEqualTo(3); + } + @Test @Timeout(120) public void testExpireWithLimit() throws Exception { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java index 00eb30d2f9d6..5a7cf00b3ea7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java @@ -65,6 +65,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.LongStream; @@ -207,6 +208,57 @@ public void retry(ManifestCommittable committable) { public void close() throws Exception {} } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testValidationAfterFiltering(boolean overwrite) throws Exception { + Path path = new Path(tempDir.toString()); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.BIGINT()}, + new String[] {"k", "v"}); + TableSchema schema = + SchemaUtils.forceCommit( + new FileSystemSchemaManager(LocalFileIO.create(), path), + new Schema( + rowType.getFields(), + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyMap(), + "")); + FileStoreTable table = + FileStoreTableFactory.create( + LocalFileIO.create(), path, schema, CatalogEnvironment.empty()); + String user = UUID.randomUUID().toString(); + try (StreamTableWrite write = table.newWrite(user); + InnerTableCommit commit = table.newCommit(user)) { + if (overwrite) { + commit.withOverwrite(Collections.emptyMap()); + } + write.write(GenericRow.of(0, 0L)); + List messages = write.prepareCommit(true, 1); + Runnable refuseCommit = + () -> { + throw new IllegalStateException("Replay history expired"); + }; + assertThatThrownBy( + () -> commit.filterAndCommit(singletonMap(1L, messages), refuseCommit)) + .hasMessage("Replay history expired"); + assertThat(table.snapshotManager().latestSnapshotId()).isNull(); + + AtomicInteger validations = new AtomicInteger(); + assertThat( + commit.filterAndCommit( + singletonMap(1L, messages), validations::incrementAndGet)) + .isEqualTo(1); + assertThat(validations.get()).isEqualTo(1); + long snapshotId = table.snapshotManager().latestSnapshotId(); + + // Validation must not reject a replay the actual filter could still recognise. + assertThat(commit.filterAndCommit(singletonMap(1L, messages), refuseCommit)).isZero(); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId); + } + } + @Test public void testRecoverDeletedFiles() throws Exception { String path = tempDir.toString(); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala index 132b8bfe470f..202367333462 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala @@ -37,7 +37,7 @@ import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_C import org.apache.paimon.spark.sort.TableSorter import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled import org.apache.paimon.spark.util.SparkRowUtils -import org.apache.paimon.spark.write.{PaimonDataWrite, WriteHelper, WriteTaskResult} +import org.apache.paimon.spark.write.{PaimonDataWrite, StreamingWrite, WriteHelper, WriteTaskResult} import org.apache.paimon.table.{FileStoreTable, SpecialFields} import org.apache.paimon.table.BucketMode._ import org.apache.paimon.table.sink._ @@ -45,6 +45,7 @@ import org.apache.paimon.types.RowKind import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory} import org.apache.spark.{Partitioner, TaskContext} +import org.apache.spark.internal.Logging import org.apache.spark.rdd.RDD import org.apache.spark.sql._ import org.apache.spark.sql.functions._ @@ -55,11 +56,20 @@ import java.util.Collections.singletonMap import scala.collection.JavaConverters._ +/** + * @param streaming + * the micro-batch this writer writes, when it writes for a streaming query. Such a write goes + * through the stream write API, the way a Flink streaming job does: it is committed under an + * identifier by the committer of the query, which recognises a replayed micro-batch. + */ case class PaimonSparkWriter( table: FileStoreTable, writeRowTracking: Boolean = false, - batchId: Option[Long] = None) - extends WriteHelper { + streaming: Option[StreamingWrite] = None) + extends WriteHelper + with Logging { + + private val commitIdentifier: Option[Long] = streaming.map(_.commitIdentifier) private lazy val tableSchema = table.schema @@ -89,8 +99,14 @@ case class PaimonSparkWriter( @transient private lazy val metricRegistry = SparkMetricRegistry() + /** + * A streaming write to a postpone bucket table always goes to bucket -2, to be bucketed by a + * later compaction, the same as a Flink streaming write: the fixed-bucket paths are for a batch + * job that ends with its commit. + */ val postponeBatchWriteFixedBucket: Boolean = - table.bucketMode() == POSTPONE_MODE && coreOptions.postponeBatchWriteFixedBucket() + streaming.isEmpty && table.bucketMode() == POSTPONE_MODE && + coreOptions.postponeBatchWriteFixedBucket() private val postponeBaseSnapshotId = if (postponeBatchWriteFixedBucket) @@ -102,25 +118,34 @@ case class PaimonSparkWriter( if (bucketNum.isPresent) Some(bucketNum.get().intValue()) else None } - val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder() + val writeBuilder: WriteBuilder = streaming match { + case Some(streamingWrite) => + // Every micro-batch is committed under the commit user of the query, which is stable across + // its runs, so that a replayed micro-batch can be recognised as already committed. + table.newStreamWriteBuilder().withCommitUser(streamingWrite.context.commitUser) + case None => table.newBatchWriteBuilder() + } def withOverwrite(): PaimonSparkWriter = withOverwrite(java.util.Collections.emptyMap()) def withOverwrite(partition: JMap[String, String]): PaimonSparkWriter = { overwritePartitionSpec = Some(partition.asScala.toMap) - if (!postponeBatchWriteFixedBucket) { - writeBuilder.withOverwrite(partition) + writeBuilder match { + case batchWriteBuilder: BatchWriteBuilder if !postponeBatchWriteFixedBucket => + batchWriteBuilder.withOverwrite(partition) + // A stream write builder is told at commit time; see [[commit]]. + case _ => } this } def writeOnly(): PaimonSparkWriter = { - PaimonSparkWriter(table.copy(singletonMap(WRITE_ONLY.key(), "true"))) + PaimonSparkWriter(table.copy(singletonMap(WRITE_ONLY.key(), "true")), streaming = streaming) } def withRowTracking(): PaimonSparkWriter = { if (coreOptions.rowTrackingEnabled()) { - PaimonSparkWriter(table, writeRowTracking = true) + PaimonSparkWriter(table, writeRowTracking = true, streaming = streaming) } else { this } @@ -140,7 +165,7 @@ case class PaimonSparkWriter( } else { None } - val activeWriteBuilder: BatchWriteBuilder = directPostponeBucketNum match { + val activeWriteBuilder: WriteBuilder = directPostponeBucketNum match { case Some(_) => val directWriteOptions = new java.util.HashMap[String, String]() directWriteOptions.put( @@ -176,9 +201,11 @@ case class PaimonSparkWriter( rowKindColIdx, writeRowTracking, fullCompactionDeltaCommits, - batchId, + commitIdentifier, uriReaderFactory, - postponePartitionBucketComputer + postponePartitionBucketComputer, + // The stream write builder does not know that the commit overwrites, see [[commit]]. + ignorePreviousFiles = streaming.isDefined && overwritePartitionSpec.isDefined ) def sparkParallelism = { @@ -458,6 +485,18 @@ case class PaimonSparkWriter( } def commit(commitMessages: Seq[CommitMessage], operation: Snapshot.Operation): Unit = { + streaming match { + case Some(streamingWrite) => + commitMicroBatch(streamingWrite, commitMessages, operation) + postCommit(commitMessages) + case None => + commitBatch(commitMessages, operation) + } + } + + private def commitBatch( + commitMessages: Seq[CommitMessage], + operation: Snapshot.Operation): Unit = { if (postponeBatchWriteFixedBucket && directPostponeWriteBuilder == null) { if (stagedSparkSession == null) { throw new IllegalStateException("Postpone staged write has no SparkSession.") @@ -471,8 +510,8 @@ case class PaimonSparkWriter( postCommit(finalMessages) return } - val activeWriteBuilder = - Option(directPostponeWriteBuilder).getOrElse(writeBuilder) + val activeWriteBuilder: BatchWriteBuilder = + Option(directPostponeWriteBuilder).getOrElse(writeBuilder.asInstanceOf[BatchWriteBuilder]) val tableCommit = activeWriteBuilder.newCommit() tableCommit.withMetricRegistry(metricRegistry) if (operation != null) { @@ -488,6 +527,34 @@ case class PaimonSparkWriter( postCommit(commitMessages) } + /** + * Commits a micro-batch through the committer of the query, which the first micro-batch of a run + * creates and every later one reuses. + */ + private def commitMicroBatch( + streamingWrite: StreamingWrite, + commitMessages: Seq[CommitMessage], + operation: Snapshot.Operation): Unit = { + streamingWrite.context.openCommitter { + val committer = writeBuilder + .asInstanceOf[StreamWriteBuilder] + .newCommit() + .asInstanceOf[InnerTableCommit] + committer.withMetricRegistry(metricRegistry) + // Both are the same for every micro-batch of a query: its output mode does not change, and + // a batch that wrote nothing is no reason to publish a snapshot, as for a batch write. + overwritePartitionSpec.foreach(spec => committer.withOverwrite(spec.asJava)) + committer.ignoreEmptyCommit( + Option(coreOptions.toConfiguration.get(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)) + .forall(_.booleanValue())) + if (operation != null) { + committer.withOperation(operation) + } + committer + } + streamingWrite.context.commit(streamingWrite, commitMessages, table) + } + private def baseSnapshotHasNoRealBuckets: Boolean = { postponeBaseSnapshotId.forall { snapshotId => diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala index 937a47c526e9..70279bb0ce25 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala @@ -24,7 +24,7 @@ import org.apache.paimon.options.Options import org.apache.paimon.spark._ import org.apache.paimon.spark.catalyst.analysis.ReplacePaimonFunctions import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper -import org.apache.paimon.spark.write.PaimonWriteOptions +import org.apache.paimon.spark.write.{PaimonWriteOptions, StreamingWrite} import org.apache.paimon.table.FileStoreTable import org.apache.spark.internal.Logging @@ -40,7 +40,7 @@ case class WriteIntoPaimonTable( saveMode: SaveMode, _data: DataFrame, options: Options, - batchId: Option[Long] = None) + streaming: Option[StreamingWrite] = None) extends RunnableCommand with ExpressionHelper with SchemaEvolutionHelper @@ -58,7 +58,7 @@ case class WriteIntoPaimonTable( updateTableWithOptions( Map(DYNAMIC_PARTITION_OVERWRITE.key -> dynamicPartitionOverwriteMode.toString)) - val writer = PaimonSparkWriter(table, batchId = batchId) + val writer = PaimonSparkWriter(table, streaming = streaming) if (overwritePartition != null) { writer.withOverwrite(overwritePartition.asJava) } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala index 9d0a1795b589..09db44d3e668 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala @@ -18,15 +18,24 @@ package org.apache.paimon.spark.sources +import org.apache.paimon.CoreOptions import org.apache.paimon.options.Options import org.apache.paimon.spark.{InsertInto, Overwrite} import org.apache.paimon.spark.commands.{SchemaEvolutionHelper, WriteIntoPaimonTable} +import org.apache.paimon.spark.write.{CommitMarker, StreamingWrite, StreamingWriteContext} import org.apache.paimon.table.FileStoreTable +import org.apache.spark.internal.Logging import org.apache.spark.sql.{DataFrame, PaimonUtils, SQLContext} import org.apache.spark.sql.execution.streaming.Sink import org.apache.spark.sql.sources.AlwaysTrue -import org.apache.spark.sql.streaming.OutputMode +import org.apache.spark.sql.streaming.{OutputMode, StreamingQueryListener} + +import java.nio.charset.StandardCharsets.UTF_8 +import java.util.UUID + +import scala.collection.JavaConverters._ +import scala.util.Try class PaimonSink( sqlContext: SQLContext, @@ -35,7 +44,167 @@ class PaimonSink( outputMode: OutputMode, options: Options) extends Sink - with SchemaEvolutionHelper { + with SchemaEvolutionHelper + with Logging { + + /** + * Structured Streaming replays a micro-batch with its original batch id when a query is restarted + * after failing between this sink returning from [[addBatch]] and Spark recording the batch as + * completed. Committing every batch under a commit user that is stable across restarts lets + * Paimon skip such a replay instead of committing its data twice. + * + * What the commit user has to identify is one incarnation of a checkpoint, since the + * micro-batches it numbers are only unique within one. Paimon skips a batch whose id a previous + * run committed under the same user, so reusing a user across two checkpoints drops the data of + * the second one, while changing it within one brings back the duplicate. The query id Spark + * persists in the checkpoint metadata is exactly that identity: it is new when a checkpoint is + * recreated, unchanged when a query resumes from one, and independent of how the location is + * spelled. It is therefore not configurable; 'commit.user-prefix' can name it. + * + * Resolved lazily: neither the query id nor the checkpoint location is available on the thread + * that constructs the sink. + */ + private lazy val commitUser: String = { + val user = withPrefix(identity) + logInfo(s"Streaming writes to ${originTable.name()} commit as '$user'.") + user + } + + /** The part of the commit user that identifies the checkpoint, without the prefix. */ + private lazy val identity: String = + queryId + .map(derivedIdentity("query", _)) + // Only reachable outside a stream execution, e.g. a direct addBatch call. A location cannot + // tell a recreated checkpoint from a resumed one, so it is a last resort. + .orElse(checkpointLocation.map(derivedIdentity("checkpoint", _))) + .getOrElse { + logWarning( + "This streaming write has neither a query id nor a checkpoint location to derive a " + + "stable commit user from, so a replayed micro-batch cannot be recognised and may " + + "be committed twice.") + UUID.randomUUID().toString + } + + // Spark hands the sink its options case-insensitively, but keeps whatever case the user wrote. + private def checkpointLocation: Option[String] = + options.toMap.asScala.collectFirst { + case (key, value) + if key.equalsIgnoreCase(PaimonSink.CHECKPOINT_LOCATION) && value != null && + value.nonEmpty => + value + } + + /** + * The id Spark persists in the checkpoint metadata. It is a thread local of the stream execution + * thread, so it can only be read from within [[addBatch]]. + */ + private def queryId: Option[String] = + Option(sqlContext.sparkContext.getLocalProperty(PaimonSink.QUERY_ID_KEY)).filter(_.nonEmpty) + + private def derivedIdentity(kind: String, value: String): String = + s"spark-$kind-${UUID.nameUUIDFromBytes(value.getBytes(UTF_8))}" + + /** + * 'commit.user-prefix' names the writers of a table, the way + * [[org.apache.paimon.CoreOptions#createCommitUser]] prefixes the random user of a batch write. + * It prefixes a derived user here too, so that a streaming job keeps the name its table was + * configured with. It cannot be the identity by itself, since every writer of the table shares + * it, so what follows it still has to be unique to the query. Being part of the commit user, it + * cannot change while a checkpoint is resumed; see [[checkPrefixUnchanged]]. + */ + private def withPrefix(user: String): String = + Seq( + options.get(CoreOptions.COMMIT_USER_PREFIX), + originTable.options().get(CoreOptions.COMMIT_USER_PREFIX.key)) + .find(prefix => prefix != null && prefix.nonEmpty) + .fold(user)(prefix => s"${prefix}_$user") + + /** + * The writer and the committer of this run of the query, from its first micro-batch until it + * terminates. Spark creates a sink per run, so a restarted query starts a new context and looks + * up once what the previous run committed. + */ + private var context: StreamingWriteContext = _ + + private def openContext(): StreamingWriteContext = synchronized { + if (context == null) { + val marker = + markerLocation.map( + new CommitMarker(_, sqlContext.sparkSession.sessionState.newHadoopConf())) + marker.foreach(checkPrefixUnchanged) + context = new StreamingWriteContext(commitUser, marker) + } + context + } + + /** Closes the context of this run, if it has created one. */ + private def closeContext(): Unit = synchronized { + if (context != null) { + context.close() + } + } + + /** + * A replayed micro-batch is only recognised under the commit user that committed it, so a run + * resuming a checkpoint under a different 'commit.user-prefix' would commit it twice. The marker + * left by the previous run names its commit user: one with the same identity but another prefix + * is that of this checkpoint, and the run is refused. A marker of another identity belongs to a + * query that used the location before, which says nothing about this one. + */ + private def checkPrefixUnchanged(marker: CommitMarker): Unit = + marker.commitUser.filter(_ != commitUser).foreach { + previous => + if (previous == identity || previous.endsWith(s"_$identity")) { + val previousPrefix = previous.stripSuffix(identity).stripSuffix("_") + throw new IllegalStateException( + s"The streaming query writing to ${originTable.name()} committed as '$previous' " + + s"before it was restarted, but now commits as '$commitUser': " + + s"'${CoreOptions.COMMIT_USER_PREFIX.key}' was changed. A micro-batch replayed by " + + "the restart would not be recognised and would be committed twice. " + + (if (previousPrefix.isEmpty) { + s"Unset '${CoreOptions.COMMIT_USER_PREFIX.key}'" + } else { + s"Set '${CoreOptions.COMMIT_USER_PREFIX.key}' back to '$previousPrefix'" + }) + + " to resume this checkpoint, or start the query from a new checkpoint.") + } + } + + /** + * Where the commit marker is kept: the checkpoint of the query, which the writer names only when + * it passes 'checkpointLocation' as an option. A checkpoint configured through + * 'spark.sql.streaming.checkpointLocation' and a query name never reaches the options, so it is + * taken from the running query instead. A query whose checkpoint cannot be found fails rather + * than write without a marker, which would let a replay after snapshot expiration be committed + * twice. Outside a stream execution there is no replay to guard. + */ + private def markerLocation: Option[String] = + checkpointLocation.orElse { + queryId.map { + id => + runningQueryCheckpoint(id).getOrElse { + throw new IllegalStateException( + s"Cannot find the checkpoint location of streaming query $id writing to " + + s"${originTable.name()}, which is needed to recognise a replayed micro-batch " + + s"after snapshot expiration. Set '${PaimonSink.CHECKPOINT_LOCATION}' as an " + + "option of the writer.") + } + } + } + + /** + * The checkpoint root Spark resolved for the running query. `StreamingQueryWrapper` and + * `StreamExecution` are internal and not in the same package across the supported Spark versions, + * so their public accessors are called by name. + */ + private def runningQueryCheckpoint(id: String): Option[String] = + Try { + val query = sqlContext.sparkSession.streams.get(UUID.fromString(id)) + val execution = query.getClass.getMethod("streamingQuery").invoke(query) + execution.getClass.getMethod("resolvedCheckpointRoot").invoke(execution).asInstanceOf[String] + }.toOption.filter(location => location != null && location.nonEmpty) + + private var closeRegistered = false override def addBatch(batchId: Long, data: DataFrame): Unit = { val saveMode = if (outputMode == OutputMode.Complete()) { @@ -44,7 +213,75 @@ class PaimonSink( InsertInto } val newData = PaimonUtils.createNewDataFrame(data) - WriteIntoPaimonTable(originTable, saveMode, newData, options, Some(batchId)).run( - sqlContext.sparkSession) + val closeAfterBatch = !registerCloseOnTermination() + try { + WriteIntoPaimonTable( + originTable, + saveMode, + newData, + options, + Some(StreamingWrite(openContext(), batchId))) + .run(sqlContext.sparkSession) + } finally { + // Outside a stream execution nothing ever terminates the query, so the committer of a batch + // cannot be kept for the next one. + if (closeAfterBatch) { + closeContext() + } + } } + + /** + * Closes the committer of this run when the query terminates, whether it completes, fails or is + * stopped. Returns whether the query is one whose termination Spark reports; it is not when + * [[addBatch]] is called outside a stream execution. + * + * A restarted query keeps its id and gets a new run id, and Spark reports terminations + * asynchronously, so the termination of the previous run may only be delivered once this run is + * writing. The listener therefore waits for the termination of this run only. + */ + private def registerCloseOnTermination(): Boolean = synchronized { + if (closeRegistered) { + return true + } + val streams = sqlContext.sparkSession.streams + // Spark's own query id is a uuid; anything else cannot be matched against a termination. + val run = for { + id <- queryId.flatMap(id => Try(UUID.fromString(id)).toOption) + query <- Option(streams.get(id)) + } yield (id, query.runId) + run match { + case Some((id, runId)) => + streams.addListener(new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {} + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = { + if (event.id == id && event.runId == runId) { + try { + closeContext() + } finally { + streams.removeListener(this) + } + } + } + }) + closeRegistered = true + true + case None => false + } + } +} + +object PaimonSink { + + private val CHECKPOINT_LOCATION = "checkpointLocation" + + /** + * `org.apache.spark.sql.execution.streaming.StreamExecution.QUERY_ID_KEY`, inlined because that + * class is not in the same package across all supported Spark versions. + */ + private val QUERY_ID_KEY = "sql.streaming.queryId" } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala new file mode 100644 index 000000000000..ea27d836d632 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/CommitMarker.scala @@ -0,0 +1,120 @@ +/* + * 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.spark.write + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path + +import java.io.IOException +import java.nio.charset.StandardCharsets.UTF_8 +import java.util.UUID + +import scala.io.Source +import scala.util.Try + +/** + * What a streaming query was about to commit, kept under its checkpoint location. + * + * A replayed micro-batch is recognised by the snapshots of its commit user, which snapshot + * expiration eventually removes. Before committing a micro-batch the sink records its batch id and + * the latest snapshot of the table at that moment: whatever that commit produced is newer than that + * snapshot, so as long as the table still retains the snapshot after it, a lookup of the commit + * user sees everything the micro-batch may have committed. Once expiration has gone past it, the + * lookup can no longer tell, and the replay must not be committed blindly. + * + * A marker is published before committing and is never replaced by a retry of the same batch. Only + * a later batch (or a new query) replaces it, after writing its complete contents to a temp file. A + * missing marker therefore means this batch never reached its commit. + */ +class CommitMarker(checkpointLocation: String, hadoopConf: Configuration) { + + val path: Path = new Path(new Path(checkpointLocation, "paimon"), "commit-marker") + + def write(commitUser: String, batchId: Long, latestSnapshotId: Long): Unit = { + if (latestSnapshotIdBefore(commitUser, batchId).isDefined) { + return + } + val fs = path.getFileSystem(hadoopConf) + val temporary = new Path(path.getParent, s".commit-marker-${UUID.randomUUID()}") + try { + val out = fs.create(temporary, false) + try { + out.write(s"$commitUser\n$batchId\n$latestSnapshotId\n".getBytes(UTF_8)) + } finally { + out.close() + } + // Spark has only one writer per checkpoint. The previous marker belongs to a different + // batch/query, so losing it here is safe: this batch has not reached its commit yet. Never + // delete a marker of this same batch, even if its first attempt has already committed. + if (fs.exists(path) && !fs.delete(path, false)) { + throw new IOException(s"Cannot replace commit marker $path") + } + if (!fs.rename(temporary, path)) { + throw new IOException(s"Cannot publish commit marker $path") + } + } finally { + fs.delete(temporary, false) + } + } + + /** + * The boundary of the first attempt of `batchId` of `commitUser`. A marker of another batch or + * query says nothing about this one. A corrupt marker, however, might have been truncated by an + * older sink retrying an already committed batch, so it must not be treated as a missing marker. + */ + def latestSnapshotIdBefore(commitUser: String, batchId: Long): Option[Long] = + read().map(lines => parse(lines).getOrElse(throw corrupt())).collect { + case (user, batch, snapshot) if user == commitUser && batch == batchId => snapshot + } + + /** + * The commit user of the micro-batch the marker was published for. A corrupt marker names none; + * the commit of a batch refuses it. + */ + def commitUser: Option[String] = read().flatMap(parse).map(_._1) + + private def read(): Option[List[String]] = { + val fs = path.getFileSystem(hadoopConf) + if (!fs.exists(path)) { + return None + } + val in = fs.open(path) + try { + Some(Source.fromInputStream(in, UTF_8.name()).getLines().toList) + } finally { + in.close() + } + } + + private def parse(lines: List[String]): Option[(String, Long, Long)] = + lines match { + case user :: batch :: snapshot :: Nil => + for { + id <- Try(batch.toLong).toOption.filter(_ >= 0) + before <- Try(snapshot.toLong).toOption.filter(_ >= 0) + if user.nonEmpty + } yield (user, id, before) + case _ => None + } + + private def corrupt(): IllegalStateException = + new IllegalStateException( + s"Cannot read commit marker $path. It may belong to an already committed micro-batch; " + + "restore a valid marker or verify the table before deleting it and restarting.") +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala index 603ec7ecff1c..3676201d5f1e 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataWrite.scala @@ -110,18 +110,22 @@ trait InnerTableV2DataWrite extends InnerTableDataWrite[InternalRow] with V2Data abstract class abstractInnerTableDataWrite[T] extends InnerTableDataWrite[T] with Logging { - val write: BatchTableWrite + val write: TableWrite val fullCompactionDeltaCommits: Option[Int] - /** For batch write, batchId is None, for streaming write, batchId is the current batch id (>= 0). */ - val batchId: Option[Long] + /** + * None for a batch write. For a streaming write, the identifier the micro-batch is committed + * under, which counts from 1. + */ + val commitIdentifier: Option[Long] private lazy val needFullCompaction: Boolean = { fullCompactionDeltaCommits match { case Some(deltaCommits) if deltaCommits > 0 => - batchId match { - case Some(id) => (id + 1) % deltaCommits == 0 + commitIdentifier match { + // The same rule by which a compacted-full scan recognises the full compaction. + case Some(identifier) => identifier % deltaCommits == 0 // When fullCompactionDeltaCommits is set, always trigger full compaction for batch write. case None => true } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala index af20144f5233..410953766731 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonDataWrite.scala @@ -30,15 +30,24 @@ import org.apache.spark.sql.Row import scala.collection.JavaConverters._ +/** + * @param writeBuilder + * a batch write builder, or the stream write builder of a streaming query. + * @param ignorePreviousFiles + * whether the write may ignore the files it would otherwise have to read first, because the + * commit overwrites them. A batch write builder decides this itself for an overwrite; a stream + * write builder does not know about overwrites. + */ case class PaimonDataWrite( - writeBuilder: BatchWriteBuilder, + writeBuilder: WriteBuilder, writeType: RowType, rowKindColIdx: Int = -1, writeRowTracking: Boolean = false, fullCompactionDeltaCommits: Option[Int], - batchId: Option[Long], + commitIdentifier: Option[Long], uriReaderFactory: UriReaderFactory, - postponePartitionBucketComputer: Option[BinaryRow => Integer]) + postponePartitionBucketComputer: Option[BinaryRow => Integer], + ignorePreviousFiles: Boolean = false) extends abstractInnerTableDataWrite[Row] with InnerTableV1DataWrite { @@ -50,6 +59,9 @@ case class PaimonDataWrite( if (writeRowTracking) { _write.withWriteType(writeType) } + if (ignorePreviousFiles) { + _write.withIgnorePreviousFiles(true) + } _write } @@ -72,7 +84,13 @@ case class PaimonDataWrite( } override def commitImpl(): Seq[CommitMessage] = { - write.prepareCommit().asScala.toSeq + val messages = commitIdentifier match { + // A Spark task does not outlive its micro-batch, so unlike a Flink writer it cannot leave a + // compaction running to be collected at the next commit. + case Some(identifier) => write.prepareCommit(true, identifier) + case None => write.prepareCommit() + } + messages.asScala.toSeq } override def close(): Unit = { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala index 05b6bfd307c4..5db5843173ab 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2DataWriter.scala @@ -38,7 +38,7 @@ case class PaimonV2DataWriter( dataSchema: StructType, coreOptions: CoreOptions, uriReaderFactory: UriReaderFactory, - batchId: Option[Long] = None, + commitIdentifier: Option[Long] = None, paimonWriteType: Option[RowType] = None, metadataSchema: Option[StructType] = None, plainWriteSchema: Option[StructType] = None) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala new file mode 100644 index 000000000000..00f3e1587396 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/StreamingWriteContext.scala @@ -0,0 +1,161 @@ +/* + * 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.spark.write + +import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.table.sink.{CommitMessage, InnerTableCommit} + +import org.apache.spark.internal.Logging + +import java.util.Collections + +import scala.collection.JavaConverters._ + +/** + * The committer of one run of a streaming query. It is created with the first micro-batch of the + * run and kept until the query terminates, the way a Flink committer lives across checkpoints: the + * maintenance a commit starts (tag creation, partition and snapshot expiration) can run on while + * the next micro-batch is written, and only a micro-batch that may be a replay has to look up what + * a previous run committed. + * + * @param commitUser + * the commit user of the query, stable across its runs, under which every micro-batch is + * committed so that a replayed one can be recognised. + * @param marker + * where the query records what it is about to commit, when it has a checkpoint location to keep + * it in. + */ +class StreamingWriteContext(val commitUser: String, @transient marker: Option[CommitMarker]) + extends AutoCloseable + with Serializable + with Logging { + + /** Driver-side state; the context itself travels with the write closures. */ + @transient private var committer: InnerTableCommit = _ + + /** The identifier of the last micro-batch this run committed, none before the first one. */ + @transient private var lastCommitted: Option[Long] = None + + /** The committer of the run, created by `create` on first use and kept afterwards. */ + def openCommitter(create: => InnerTableCommit): InnerTableCommit = + synchronized { + if (committer == null) { + committer = create + } + committer + } + + /** + * Commits a micro-batch under its identifier. + * + * Structured Streaming replays a micro-batch with its original batch id when a query is restarted + * after failing between the sink returning and Spark recording the batch as completed, so the + * first micro-batch of a run may already have been committed by a previous run. A batch that + * follows one this run committed cannot be a replay of it, and is committed without looking the + * history of the commit user up. + */ + def commit(write: StreamingWrite, messages: Seq[CommitMessage], table: FileStoreTable): Unit = + synchronized { + require(committer != null, "The committer of the streaming query has not been created.") + val identifier = write.commitIdentifier + val mayBeReplay = !lastCommitted.exists(identifier > _) + val previousAttempt = + if (mayBeReplay) marker.flatMap(_.latestSnapshotIdBefore(commitUser, write.batchId)) + else None + marker.foreach( + _.write( + commitUser, + write.batchId, + Option(table.snapshotManager().latestSnapshotId()).map(_.longValue()).getOrElse(0L))) + if (mayBeReplay) { + val commits = + committer.filterAndCommit( + Collections.singletonMap(Long.box(identifier), messages.asJava), + () => checkReplayIsRecognisable(previousAttempt, identifier, table)) + if (commits == 0) { + // Expected of a replay; the only trace of a misconfigured identity otherwise. + logInfo( + s"Micro-batch $identifier was already committed to ${table.name()} under commit " + + s"user '$commitUser' and is skipped as a replay.") + } + } else { + committer.commit(identifier, messages.asJava) + } + lastCommitted = Some(lastCommitted.fold(identifier)(Math.max(_, identifier))) + } + + /** + * Validate only after the committer has actually filtered the replay. Checking beforehand would + * race expiration between two independent lookups of the commit user's snapshots. Once filtering + * says this batch needs committing, its original boundary must still be retained; otherwise the + * lookup may have missed an expired commit. A batch with no previous marker never reached commit. + */ + private def checkReplayIsRecognisable( + previousAttempt: Option[Long], + identifier: Long, + table: FileStoreTable): Unit = { + val snapshots = table.snapshotManager() + previousAttempt.foreach { + before => + // Check the file itself: expiration deletes snapshots before updating the earliest hint. + if ( + Option(snapshots.latestSnapshotId()).exists(_.longValue() > before) && + !snapshots.snapshotExists(before + 1) + ) { + throw new IllegalStateException( + s"Cannot tell whether micro-batch $identifier of this query was committed to " + + s"${table.name()}. A previous attempt of it was about to commit when the latest " + + s"snapshot was $before, but snapshot ${before + 1} is no longer retained, so " + + s"the lookup of commit user '$commitUser' may have missed its commit. " + + "Committing it again could write it twice. Retain snapshots for longer than a " + + "query may be down. If the table shows that the micro-batch was not committed, " + + s"delete ${marker.get.path} and restart the query to commit it; otherwise restart " + + "the query from a new checkpoint.") + } + } + } + + override def close(): Unit = + synchronized { + if (committer != null) { + val toClose = committer + committer = null + toClose.close() + } + } +} + +/** + * One micro-batch of a streaming query, written and committed through the context of the query's + * current run. + * + * @param batchId + * the id Spark gives the micro-batch, counted from 0. + */ +case class StreamingWrite(context: StreamingWriteContext, batchId: Long) { + + /** + * Paimon numbers the commits of a stream from 1, the way Flink numbers its checkpoints, and + * [[org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner]] recognises a full + * compaction by an identifier that is a multiple of 'full-compaction.delta-commits'. Spark + * numbers its micro-batches from 0, so the third one is batch 2; both the full compaction + * schedule and the identifier it is published under have to count it as 3. + */ + val commitIdentifier: Long = batchId + 1 +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala new file mode 100644 index 000000000000..9d733f857062 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkIdempotencyTest.scala @@ -0,0 +1,1513 @@ +/* + * 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.spark + +import org.apache.paimon.catalog.{Catalog, CatalogLoader, DelegateCatalog, Identifier} +import org.apache.paimon.data.{BinaryString, GenericRow} +import org.apache.paimon.manifest.ManifestCommittable +import org.apache.paimon.options.Options +import org.apache.paimon.spark.sources.PaimonSink +import org.apache.paimon.spark.write.{CommitMarker, StreamingWrite, StreamingWriteContext} +import org.apache.paimon.table.{CatalogEnvironment, FileStoreTableFactory} +import org.apache.paimon.table.sink.{CommitCallback, InnerTableCommit} + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{Path, RawLocalFileSystem} +import org.apache.hadoop.fs.permission.FsPermission +import org.apache.spark.SparkConf +import org.apache.spark.sql.{DataFrame, Row} +import org.apache.spark.sql.paimon.shims.memstream.MemoryStream +import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, StreamingQueryListener, StreamTest} + +import java.io.{File, IOException, OutputStream} +import java.util.{Collections, List => JList, Map => JMap, UUID} +import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} + +import scala.collection.JavaConverters._ + +/** + * Structured Streaming guarantees exactly-once only if the sink is idempotent for a repeated + * batchId: when a query fails between the sink returning from `addBatch` and Spark recording the + * batch as completed, the restarted query replays that micro-batch with its original batchId. + */ +class PaimonSinkIdempotencyTest extends PaimonSparkTestBase with StreamTest { + + override protected def sparkConf: SparkConf = { + super.sparkConf.set("spark.sql.catalog.paimon.cache-enabled", "false") + } + + import testImplicits._ + + private def snapshotCount(tableName: String): Long = + loadTable(tableName).snapshotManager().snapshotCount() + + private def latestCommitUser(tableName: String): String = + loadTable(tableName).snapshotManager().latestSnapshot().commitUser() + + private def deleteRecursively(file: File): Unit = { + if (file.isDirectory) { + file.listFiles().foreach(deleteRecursively) + } + file.delete() + } + + private def causes(e: Throwable): Seq[Throwable] = + Iterator.iterate(e)(_.getCause).takeWhile(_ != null).take(20).toSeq + + private def runToCompletion(query: StreamingQuery): Unit = { + try { + query.processAllAvailable() + } finally { + query.stop() + } + } + + /** + * Leave the checkpoint in the state a driver failure leaves behind when it dies after the sink + * returned from `addBatch` but before Spark recorded the batch: the offset log still has the + * batch, the commit log does not. The restarted query replays it with the same batchId. + */ + private def dropCommitLogEntry(checkpointPath: String, batchId: Long): Unit = { + val commitsDir = new File(checkpointPath, "commits") + val names = Set(batchId.toString, s".$batchId.crc") + val entries = commitsDir.listFiles().filter(f => names.contains(f.getName)) + assert( + entries.exists(_.getName == batchId.toString), + s"no commit log entry for batch $batchId in $commitsDir") + entries.foreach(f => assert(f.delete())) + } + + test("Paimon Sink: replayed micro-batch must not be committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a"), (2, "b"), (3, "c")) + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + + val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert(snapshotCount("T") == 1) + assert( + latestCommitUser("T").startsWith("spark-query-"), + s"expected a commit user derived from the query id, " + + s"but got '${latestCommitUser("T")}'" + ) + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + // The replayed batch must be recognised as already committed. + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + } + } + + test("Paimon Sink: replay is recognised when only the query id is available") { + failAfter(streamingTimeout) { + withTempDir { + checkpointRoot => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val queryName = "paimon_idempotency" + // The location never reaches the sink options this way, so the commit user has to come + // from the query id that Spark persists in the checkpoint metadata. + val checkpointPath = new File(checkpointRoot, queryName).getCanonicalPath + + withSQLConf("spark.sql.streaming.checkpointLocation" -> checkpointRoot.getCanonicalPath) { + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a"), (2, "b"), (3, "c")) + + def start(): StreamingQuery = + df.writeStream + .queryName(queryName) + .format("paimon") + .start(location) + + runToCompletion(start()) + + val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert(snapshotCount("T") == 1) + assert( + latestCommitUser("T").startsWith("spark-query-"), + s"expected a commit user derived from the query id, " + + s"but got '${latestCommitUser("T")}'") + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + } + } + } + + test("Paimon Sink: replay of a batch that is not the first one is recognised") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + val query = start() + try { + inputData.addData((1, "a")) + query.processAllAvailable() + inputData.addData((2, "b")) + query.processAllAvailable() + inputData.addData((3, "c")) + query.processAllAvailable() + } finally { + query.stop() + } + + val expected = Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert(snapshotCount("T") == 3) + + dropCommitLogEntry(checkpointPath, 2) + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), expected) + assert( + snapshotCount("T") == 3, + s"replaying batch 2 created another snapshot (${snapshotCount("T")} in total)") + } + } + } + + test("Paimon Sink: replayed micro-batch of a complete mode query is not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (city STRING, population LONG)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData + .toDS() + .toDF("uid", "city") + .groupBy("city") + .count() + .toDF("city", "population") + inputData.addData((1, "HZ"), (2, "BJ"), (3, "BJ")) + + def start(): StreamingQuery = + df.writeStream + .outputMode("complete") + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + + val expected = Row("BJ", 2L) :: Row("HZ", 1L) :: Nil + checkAnswer(spark.sql("SELECT * FROM T ORDER BY city"), expected) + val snapshotsAfterFirstBatch = snapshotCount("T") + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T ORDER BY city"), expected) + assert( + snapshotCount("T") == snapshotsAfterFirstBatch, + s"replaying batch 0 created another snapshot (${snapshotCount("T")} in total, " + + s"$snapshotsAfterFirstBatch before the replay)" + ) + } + } + } + + test("Paimon Sink: complete mode replay on a postpone bucket table is not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A streaming write to a postpone bucket table writes bucket -2, the way a Flink + // streaming write does: the fixed bucket paths are for a batch job that ends with its + // commit, and 'postpone.batch-write-fixed-bucket' is not for a stream. + spark.sql( + "CREATE TABLE T (city STRING, population LONG) TBLPROPERTIES (" + + "'primary-key' = 'city', 'bucket' = '-2', 'postpone.default-bucket-num' = '1')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData + .toDS() + .toDF("uid", "city") + .groupBy("city") + .count() + .toDF("city", "population") + inputData.addData((1, "HZ"), (2, "BJ"), (3, "BJ")) + + def start(): StreamingQuery = + df.writeStream + .outputMode("complete") + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + + // The written rows wait in bucket -2 for a compaction, so the table still reads empty. + checkAnswer(spark.sql("SELECT * FROM T"), Nil) + checkAnswer(spark.sql("SELECT count(*) FROM `T$buckets` WHERE bucket = -2"), Row(1L)) + val snapshotsAfterFirstBatch = snapshotCount("T") + assert( + latestCommitUser("T").startsWith("spark-query-"), + s"expected the streaming committer to use the commit user derived from the query id, " + + s"but got '${latestCommitUser("T")}'" + ) + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + assert( + snapshotCount("T") == snapshotsAfterFirstBatch, + s"replaying batch 0 created another snapshot (${snapshotCount("T")} in total, " + + s"$snapshotsAfterFirstBatch before the replay)" + ) + + spark.sql("CALL sys.compact(table => 'T')") + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY city"), + Row("BJ", 2L) :: Row("HZ", 1L) :: Nil) + + // A later batch is not a replay and has to be committed: the replay lookup must not + // mistake a higher batch id for one that was already committed. Complete mode overwrites + // what the compaction has just materialized. + val snapshotsBeforeSecondBatch = snapshotCount("T") + inputData.addData((4, "SH")) + runToCompletion(start()) + + assert( + snapshotCount("T") == snapshotsBeforeSecondBatch + 1, + s"the batch after the replay was not committed (${snapshotCount("T")} snapshots)") + spark.sql("CALL sys.compact(table => 'T')") + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY city"), + Row("BJ", 2L) :: Row("HZ", 1L) :: Row("SH", 1L) :: Nil) + } + } + } + + test("Paimon Sink: a replay retries the partition registration of the batch it skips") { + withTempDir { + checkpointDir => + // A commit publishes its snapshot and only then registers the partition in the metastore. + // If that registration fails, the replay of the batch has to retry it: the snapshot + // already exists, so the replay must not commit again, but it must not report success + // before the partition is registered either. + spark.sql( + "CREATE TABLE T (city STRING, population LONG, dt STRING) PARTITIONED BY (dt) " + + "TBLPROPERTIES ('primary-key' = 'city,dt', 'bucket' = '-2', " + + "'postpone.default-bucket-num' = '1', 'metastore.partitioned-table' = 'true')") + val base = loadTable("T") + FailOnceRegistration.reset(paimonCatalog) + val environment = new CatalogEnvironment( + base.catalogEnvironment().identifier(), + base.catalogEnvironment().uuid(), + FailOnceRegistration.loader, + null, + null, + null, + false, + true) + val table = FileStoreTableFactory.create( + base.fileIO(), + base.location(), + base.schema(), + new Options(base.options()), + environment) + + def newSink(): PaimonSink = new PaimonSink( + spark.sqlContext, + table, + Nil, + OutputMode.Complete(), + Options.fromMap( + Collections.singletonMap("checkpointLocation", checkpointDir.getCanonicalPath))) + + val batch: DataFrame = Seq(("HZ", 1L, "2026-09-12")).toDF("city", "population", "dt") + + // The first attempt fails after the snapshot is published. + val failure = intercept[Exception](newSink().addBatch(0L, batch)) + assert( + failure.getMessage.contains("metastore unavailable") || + Option(failure.getCause).exists(_.getMessage.contains("metastore unavailable"))) + assert(snapshotCount("T") == 1) + assert(FailOnceRegistration.attempts == 1) + assert(FailOnceRegistration.registered.isEmpty) + + // The restarted query replays the batch. + newSink().addBatch(0L, batch) + + assert(snapshotCount("T") == 1, "the replay must not commit a second snapshot") + assert( + FailOnceRegistration.attempts == 2, + "the replay must retry the partition registration that failed after the snapshot") + assert( + FailOnceRegistration.registered.contains("2026-09-12"), + "the partition committed by the replayed batch must be registered") + } + } + + test("Paimon Sink: commit.user-prefix names the derived commit user") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // 'commit.user-prefix' names the writers of a table, as it does for a batch write. It + // only prefixes the derived user, so the identity of the query is unchanged and a + // replayed micro-batch is still recognised. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'my-job')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a")) + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + + assert( + latestCommitUser("T").startsWith("my-job_spark-query-"), + s"expected the prefix in front of the derived user, but the commit user is " + + s"'${latestCommitUser("T")}'" + ) + + dropCommitLogEntry(checkpointPath, 0) + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + } + } + + test("Paimon Sink: commit.user-prefix must not make two queries share a commit user") { + failAfter(streamingTimeout) { + withTempDir { + dir => + // The prefix belongs to every writer of the table, so what follows it still has to be + // unique to the query: two queries sharing the prefix must not drop each other's + // batches as replays. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'my-job')") + val location = loadTable("T").location().toString + + def runOneBatch(checkpoint: String, row: (Int, String)): Unit = { + val input = MemoryStream[(Int, String)] + input.addData(row) + runToCompletion( + input + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", new File(dir, checkpoint).getCanonicalPath) + .format("paimon") + .start(location)) + } + + runOneBatch("a", (1, "a")) + val userA = latestCommitUser("T") + runOneBatch("b", (2, "b")) + val userB = latestCommitUser("T") + + assert(userA != userB, s"both queries committed as '$userA'") + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil) + } + } + } + + /** + * Commits batch 0 under 'commit.user-prefix' = 'old-job', leaves it to be replayed, and resumes + * the checkpoint under 'new-job' with `start`, which applies the prefix it is given. + */ + private def checkPrefixChangeIsRefused( + checkpointPath: String, + start: String => StreamingQuery): Unit = { + runToCompletion(start("old-job")) + assert(latestCommitUser("T").startsWith("old-job_spark-query-")) + dropCommitLogEntry(checkpointPath, 0) + + // Under another commit user the replay of batch 0 would not be recognised. + val failure = intercept[Exception](runToCompletion(start("new-job"))) + assert( + causes(failure).exists( + e => + e.isInstanceOf[IllegalStateException] && + e.getMessage.contains("Set 'commit.user-prefix' back to 'old-job'")), + s"expected the prefix change to be refused, but got $failure" + ) + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert(snapshotCount("T") == 1) + + // Under the previous prefix the replay is recognised. + runToCompletion(start("old-job")) + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert( + snapshotCount("T") == 1, + s"replaying batch 0 created a second snapshot (${snapshotCount("T")} in total)") + } + + test("Paimon Sink: a checkpoint is not resumed under another table commit.user-prefix") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a")) + + checkPrefixChangeIsRefused( + checkpointPath, + prefix => { + spark.sql(s"ALTER TABLE T SET TBLPROPERTIES ('commit.user-prefix' = '$prefix')") + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + } + ) + } + } + } + + test("Paimon Sink: a checkpoint is not resumed under another write commit.user-prefix") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a")) + + checkPrefixChangeIsRefused( + checkpointPath, + prefix => + df.writeStream + .option("checkpointLocation", checkpointPath) + .option("commit.user-prefix", prefix) + .format("paimon") + .start(location) + ) + } + } + } + + test("Paimon Sink: addBatch with a repeated batchId must be a no-op") { + withTempDir { + checkpointDir => + withTable("T2") { + spark.sql("CREATE TABLE T2 (a INT, b STRING)") + // Called outside a stream execution there is no query id, so this also covers the + // checkpoint location fallback. + val sink = new PaimonSink( + spark.sqlContext, + loadTable("T2"), + Nil, + OutputMode.Append(), + Options.fromMap( + Collections.singletonMap("checkpointLocation", checkpointDir.getCanonicalPath)) + ) + + val batch: DataFrame = Seq((1, "a"), (2, "b")).toDF("a", "b") + sink.addBatch(0L, batch) + sink.addBatch(0L, batch) + + checkAnswer(spark.sql("SELECT * FROM T2 ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil) + assert(snapshotCount("T2") == 1) + assert( + latestCommitUser("T2").startsWith("spark-checkpoint-"), + s"expected a commit user derived from the checkpoint location, " + + s"but got '${latestCommitUser("T2")}'" + ) + } + } + } + + test("Paimon Sink: a new query reusing a checkpoint location must not skip its batches") { + failAfter(streamingTimeout) { + withTempDir { + dir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointDir = new File(dir, "cp") + + def runOneBatch(row: (Int, String)): Unit = { + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData(row) + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location)) + } + + runOneBatch((1, "old")) + val firstCommitUser = latestCommitUser("T") + + // The checkpoint is dropped and an unrelated query starts at the same location. Its + // batch ids start at 0 again, so reusing the identity of the previous query would + // make Paimon skip its data as an already committed replay. + deleteRecursively(checkpointDir) + runOneBatch((2, "new")) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "old") :: Row(2, "new") :: Nil) + assert( + latestCommitUser("T") != firstCommitUser, + "a query that does not continue the previous checkpoint must not reuse its " + + "commit user") + } + } + } + + test("Paimon Sink: a new checkpoint must not skip its batches under a named commit user") { + failAfter(streamingTimeout) { + withTempDir { + dir => + // A new checkpoint numbers its batches from 0 again, so its identifiers collide with + // those the previous query published. Only the query id, which is new with the + // checkpoint, keeps the two apart: a name configured for the writers of the table, + // which both queries share, must not. + spark.sql("CREATE TABLE T (a INT, b STRING) TBLPROPERTIES ('commit.user-prefix' = 'job')") + val location = loadTable("T").location().toString + val checkpointDir = new File(dir, "cp") + + def runBatches(rows: (Int, String)*): Unit = { + val inputData = MemoryStream[(Int, String)] + val query = inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + // No longer an option; a leftover setting must not pin the identity either. + .option("write.stream.commit-user", "my-job") + .format("paimon") + .start(location) + try { + rows.foreach { + row => + inputData.addData(row) + query.processAllAvailable() + } + } finally { + query.stop() + } + } + + runBatches((1, "old-1"), (2, "old-2")) + val firstCommitUser = latestCommitUser("T") + deleteRecursively(checkpointDir) + runBatches((3, "new-1"), (4, "new-2"), (5, "new-3")) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "old-1") :: Row(2, "old-2") :: Row(3, "new-1") :: Row(4, "new-2") :: + Row(5, "new-3") :: Nil) + assert(latestCommitUser("T") != firstCommitUser) + assert(latestCommitUser("T").startsWith("job_spark-query-")) + } + } + } + + test("Paimon Sink: a replay whose snapshot may have expired is refused, not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A replay is recognised by the snapshots of its commit user. Once expiration has + // removed them, the sink cannot tell whether the batch was committed, and must not + // commit it again on the chance that it was not. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "stream")) + + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + runToCompletion(start()) + val streamSnapshot = loadTable("T").snapshotManager().latestSnapshotId() + + // Another writer commits, and its expiration removes the snapshot of the batch. + spark.sql("INSERT INTO T VALUES (2, 'batch')") + assert( + !loadTable("T").snapshotManager().snapshotExists(streamSnapshot), + "the snapshot of the streaming batch should have expired") + + // The driver failed before recording batch 0, which is replayed on restart. + dropCommitLogEntry(checkpointPath, 0) + val failure = intercept[Exception](runToCompletion(start())) + assert( + causes(failure).exists( + e => + e.getMessage != null && + e.getMessage.contains("Cannot tell whether micro-batch 1 of this query")), + s"expected the replay to be refused, but got $failure" + ) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "stream") :: Row(2, "batch") :: Nil) + } + } + } + + test("Paimon Sink: a replay is refused after expiration with a session-configured checkpoint") { + failAfter(streamingTimeout) { + withTempDir { + checkpointRoot => + // Reviewer P1 on #10105: a checkpoint configured through the session and a query name + // never reaches the sink options, so the marker has to be kept in the checkpoint the + // query actually uses. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + val queryName = "paimon_session_checkpoint" + val checkpointPath = new File(checkpointRoot, queryName).getCanonicalPath + + withSQLConf("spark.sql.streaming.checkpointLocation" -> checkpointRoot.getCanonicalPath) { + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "stream")) + + def start(): StreamingQuery = + df.writeStream + .queryName(queryName) + .format("paimon") + .start(location) + + runToCompletion(start()) + assert( + new File(checkpointPath, "paimon/commit-marker").exists(), + "the marker belongs in the checkpoint the query uses") + + spark.sql("INSERT INTO T VALUES (2, 'batch')") + dropCommitLogEntry(checkpointPath, 0) + val failure = intercept[Exception](runToCompletion(start())) + assert( + causes(failure).exists( + e => + e.getMessage != null && + e.getMessage.contains("Cannot tell whether micro-batch 1 of this query")), + s"expected the replay to be refused, but got $failure" + ) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "stream") :: Row(2, "batch") :: Nil) + } + } + } + } + + test("Paimon Sink: a streaming query whose checkpoint cannot be found fails closed") { + withTable("T2") { + // Inside a stream execution, but with no checkpoint in the options and no running query to + // take it from: writing without a marker could commit a replay twice, so the sink refuses. + spark.sql("CREATE TABLE T2 (a INT, b STRING)") + val sink = + new PaimonSink(spark.sqlContext, loadTable("T2"), Nil, OutputMode.Append(), new Options()) + val batch: DataFrame = Seq((1, "a")).toDF("a", "b") + + spark.sparkContext.setLocalProperty( + "sql.streaming.queryId", + java.util.UUID.randomUUID().toString) + try { + val failure = intercept[IllegalStateException](sink.addBatch(0L, batch)) + assert(failure.getMessage.contains("Cannot find the checkpoint location")) + } finally { + spark.sparkContext.setLocalProperty("sql.streaming.queryId", null) + } + assert(snapshotCount("T2") == 0) + } + } + + test("Paimon Sink: the replay check refuses only what it cannot tell") { + withTempDir { + dir => + // The first micro-batch of a run may be a replay. With no snapshot of its commit user + // left, the marker its previous attempt wrote before committing decides: no marker, or + // one from another batch, means that attempt never committed; a marker older than every + // retained snapshot means it may have, and the table cannot say. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + spark.sql("INSERT INTO T VALUES (0, 'seed')") + val hadoopConf = spark.sessionState.newHadoopConf() + + def attempt(name: String, prepare: CommitMarker => Unit, tableName: String = "T"): Unit = { + val table = loadTable(tableName) + val user = s"user-$name" + val marker = new CommitMarker(new File(dir, name).getCanonicalPath, hadoopConf) + prepare(marker) + val context = new StreamingWriteContext(user, Some(marker)) + val builder = table.newStreamWriteBuilder().withCommitUser(user) + val write = builder.newWrite() + try { + write.write(GenericRow.of(Integer.valueOf(1), BinaryString.fromString(name))) + val messages = write.prepareCommit(true, 6).asScala.toSeq + context.openCommitter(builder.newCommit().asInstanceOf[InnerTableCommit]) + context.commit(StreamingWrite(context, 5), messages, table) + } finally { + write.close() + context.close() + } + } + + def latestId(): Long = loadTable("T").snapshotManager().latestSnapshotId() + + // No marker: the previous attempt never got to its commit. + attempt("none", _ => ()) + // A marker of another batch says nothing about this one. + attempt("other", _.write("user-other", 4, 0)) + // A marker of this batch that no expiration has gone past: a commit would be retained. + attempt("kept", m => m.write("user-kept", 5, latestId())) + assert( + spark.sql("SELECT b FROM T WHERE a = 1").collect().map(_.getString(0)).toSet == + Set("none", "other", "kept"), + "each of these attempts should have been committed" + ) + + // A marker of this batch on a table without any snapshot: nothing can have expired. + withTable("T3") { + spark.sql("CREATE TABLE T3 (a INT, b STRING)") + attempt("empty", m => m.write("user-empty", 5, 0), "T3") + checkAnswer(spark.sql("SELECT * FROM T3"), Row(1, "empty") :: Nil) + } + + // A marker of this batch older than every retained snapshot: refused. + val refused = intercept[IllegalStateException]( + attempt("expired", m => m.write("user-expired", 5, latestId() - 2))) + assert(refused.getMessage.contains("Cannot tell whether micro-batch 6 of this query")) + assert(spark.sql("SELECT * FROM T WHERE b = 'expired'").collect().isEmpty) + + // What the error says to do once the table shows the batch was not committed: delete + // the marker it names, and the next attempt commits the batch. + val named = "(/\\S+/paimon/commit-marker)".r.findFirstIn(refused.getMessage) + assert(named.isDefined, s"the error should name the marker: ${refused.getMessage}") + assert(new File(named.get).delete(), s"cannot delete ${named.get}") + attempt("expired", _ => ()) + checkAnswer(spark.sql("SELECT a FROM T WHERE b = 'expired'"), Row(1) :: Nil) + } + } + + private def commitReplayAttempt(marker: CommitMarker, batchId: Long = 0): Unit = { + val table = loadTable("T") + val user = "recovery-user" + val context = new StreamingWriteContext(user, Some(marker)) + val builder = table.newStreamWriteBuilder().withCommitUser(user) + val write = builder.newWrite() + try { + write.write(GenericRow.of(Integer.valueOf(1), BinaryString.fromString("stream"))) + val messages = write.prepareCommit(true, batchId + 1).asScala.toSeq + context.openCommitter(builder.newCommit().asInstanceOf[InnerTableCommit]) + context.commit(StreamingWrite(context, batchId), messages, table) + } finally { + write.close() + context.close() + } + } + + test("Paimon Sink: repeated recovery preserves the first attempt's marker") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + val original = loadTable("T").snapshotManager().latestSnapshotId() + val interrupted = new CommitMarker(dir.getCanonicalPath, conf) { + override def write(user: String, batch: Long, latest: Long): Unit = { + super.write(user, batch, latest) + throw new IllegalStateException("driver failed before filtering") + } + } + (1 to 2).foreach { + _ => + assert( + intercept[IllegalStateException](commitReplayAttempt(interrupted)).getMessage + .contains("driver failed")) + } + assert(marker.latestSnapshotIdBefore("recovery-user", 0).contains(0L)) + spark.sql("INSERT INTO T VALUES (2, 'other')") + assert(!loadTable("T").snapshotManager().snapshotExists(original)) + assert( + intercept[IllegalStateException](commitReplayAttempt(marker)).getMessage + .contains("Cannot tell whether")) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + + Seq(false, true).foreach { + staleHint => + test(s"Paimon Sink: expiration before replay filtering is refused (stale hint: $staleHint)") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + val original = loadTable("T").snapshotManager().latestSnapshotId() + val concurrentExpiration = new CommitMarker(dir.getCanonicalPath, conf) { + override def write(user: String, batch: Long, latest: Long): Unit = { + super.write(user, batch, latest) + // Interleave another writer after the old precheck but before actual filtering. + spark.sql("INSERT INTO T VALUES (2, 'other')") + val snapshots = loadTable("T").snapshotManager() + assert(!snapshots.snapshotExists(original)) + if (staleHint) { + // Expiration removes snapshot files before advancing the earliest hint. + snapshots.commitEarliestHint(original) + } + } + } + assert( + intercept[IllegalStateException](commitReplayAttempt(concurrentExpiration)).getMessage + .contains("Cannot tell whether")) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + } + + test("Paimon Sink: a corrupt marker must not be treated as an uncommitted batch") { + withTempDir { + dir => + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val conf = spark.sessionState.newHadoopConf() + val marker = new CommitMarker(dir.getCanonicalPath, conf) + commitReplayAttempt(marker) + spark.sql("INSERT INTO T VALUES (2, 'other')") + // An older sink could truncate the committed batch's marker during recovery. + val out = marker.path.getFileSystem(conf).create(marker.path, true) + out.write("recovery-user\n0".getBytes("UTF-8")) + out.close() + val failure = intercept[IllegalStateException](commitReplayAttempt(marker)) + assert(failure.getMessage.contains(marker.path.toString)) + checkAnswer(spark.sql("SELECT * FROM T WHERE b = 'stream'"), Row(1, "stream") :: Nil) + } + } + + test("Paimon Sink: an interrupted marker update keeps the previous batch's marker") { + withTempDir { + dir => + val conf = new Configuration(spark.sessionState.newHadoopConf()) + conf.setClass( + "fs.file.impl", + classOf[FailingMarkerFileSystem], + classOf[org.apache.hadoop.fs.FileSystem]) + conf.setBoolean("fs.file.impl.disable.cache", true) + val marker = new CommitMarker(dir.getCanonicalPath, conf) + marker.write("recovery-user", 0, 0) + FailingMarkerFileSystem.failWrites = true + try { + // A same-batch retry must not write, even when the filesystem rejects new writes. + marker.write("recovery-user", 0, 1) + intercept[IOException](marker.write("recovery-user", 1, 2)) + } finally { + FailingMarkerFileSystem.failWrites = false + } + assert(marker.latestSnapshotIdBefore("recovery-user", 0).contains(0L)) + marker.write("recovery-user", 1, 2) + assert(marker.latestSnapshotIdBefore("recovery-user", 1).contains(2L)) + // Recreating the checkpoint gives the same batch number a different identity. + marker.write("another-query", 0, 3) + assert(marker.latestSnapshotIdBefore("another-query", 0).contains(3L)) + } + } + + // Every kind of table the sink writes to: the write and commit paths differ by bucket mode and + // by what a commit also produces (compactions, deletion vectors, changelog, row ids). + Seq( + ("dynamic bucket", "'primary-key' = 'k,pt', 'bucket' = '-1'", false), + ("cross partition", "'primary-key' = 'k', 'bucket' = '-1'", false), + ( + "postpone bucket with real buckets", + "'primary-key' = 'k,pt', 'bucket' = '-2', 'postpone.default-bucket-num' = '1'", + true), + ( + "deletion vector", + "'primary-key' = 'k,pt', 'bucket' = '1', 'deletion-vectors.enabled' = 'true'", + false), + ( + "lookup changelog", + "'primary-key' = 'k,pt', 'bucket' = '1', 'changelog-producer' = 'lookup'", + false), + ("row tracking", "'row-tracking.enabled' = 'true'", false) + ).foreach { + case (kind, properties, postpone) => + test(s"Paimon Sink: a replay into a $kind table is not committed twice") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql( + "CREATE TABLE T (k INT, v STRING, pt STRING) PARTITIONED BY (pt) " + + s"TBLPROPERTIES ($properties)") + val seed = if (postpone) { + // A batch write puts this into a real bucket, so the streaming batches below + // are appended to a table that already has buckets. + spark.sql("INSERT INTO T VALUES (0, 'seed', 'p1')") + Row(0, "seed", "p1") :: Nil + } else { + Nil + } + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String, String)] + val df = inputData.toDS().toDF("k", "v", "pt") + def run(): Unit = + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location)) + + inputData.addData((1, "a", "p1"), (2, "b", "p2")) + run() + inputData.addData((3, "c", "p1")) + run() + if (postpone) { + checkAnswer( + spark.sql("SELECT count(*) > 0 FROM `T$buckets` WHERE bucket = -2"), + Row(true)) + } + + // A primary key table would hide a duplicate behind its merge, so what shows a + // replay committed again is a new snapshot. + val snapshots = snapshotCount("T") + dropCommitLogEntry(checkpointPath, 1) + run() + assert( + snapshotCount("T") == snapshots, + s"replaying batch 1 created ${snapshotCount("T") - snapshots} more snapshots") + + if (postpone) { + spark.sql("CALL sys.compact(table => 'T')") + } + checkAnswer( + spark.sql("SELECT k, v, pt FROM T ORDER BY k"), + seed ++ (Row(1, "a", "p1") :: Row(2, "b", "p2") :: Row(3, "c", "p1") :: Nil)) + } + } + } + } + + test("Paimon Sink: a batch that failed before its commit is committed once on restart") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The query fails while writing batch 1, before anything is committed. The run's + // committer has to be closed although the query did not stop cleanly, and the restart + // replays batch 1, which, never committed, must be committed once. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData + .toDS() + .map { + row => + if (row._1 == 2 && FailBeforeCommit.armed) { + throw new RuntimeException("injected failure before the commit") + } + row + } + .toDF("a", "b") + def start(): StreamingQuery = + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location) + + FailBeforeCommit.armed = true + try { + val query = start() + try { + inputData.addData((1, "a")) + query.processAllAvailable() + inputData.addData((2, "b")) + intercept[Exception](query.processAllAvailable()) + } finally { + query.stop() + } + } finally { + FailBeforeCommit.armed = false + } + assert(snapshotCount("T") == 1, "only batch 0 was committed before the failure") + // The query terminates asynchronously; bounded by the timeout of this test. + while (CountingCommitCallback.closed < CountingCommitCallback.created) { + Thread.sleep(50) + } + + runToCompletion(start()) + + checkAnswer(spark.sql("SELECT * FROM T ORDER BY a"), Row(1, "a") :: Row(2, "b") :: Nil) + assert(snapshotCount("T") == 2) + } + } + } + + test("Paimon Sink: a query upgraded from per-batch commits keeps committing") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // Before this sink, every micro-batch was a batch write: a random commit user, the + // batch commit identifier, and nothing under the checkpoint. A query resuming such a + // checkpoint has a commit user with no history and no marker, on a table whose older + // snapshots may have expired; its next batch is new and must be committed. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.num-retained.min' = '1', 'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + + val before = df.writeStream + .option("checkpointLocation", checkpointPath) + .foreachBatch { + (batch: DataFrame, _: Long) => + batch.write.format("paimon").mode("append").save(location) + } + .start() + try { + inputData.addData((1, "a")) + before.processAllAvailable() + inputData.addData((2, "b")) + before.processAllAvailable() + } finally { + before.stop() + } + assert(loadTable("T").snapshotManager().earliestSnapshotId() > 1) + + inputData.addData((3, "c")) + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location)) + + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + assert(latestCommitUser("T").startsWith("spark-query-")) + } + } + } + + test("Paimon Sink: an equivalent spelling of the checkpoint location keeps the identity") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql("CREATE TABLE T (a INT, b STRING)") + val location = loadTable("T").location().toString + val checkpointPath = checkpointDir.getCanonicalPath + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + inputData.addData((1, "a")) + + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath) + .format("paimon") + .start(location)) + val firstCommitUser = latestCommitUser("T") + + dropCommitLogEntry(checkpointPath, 0) + // The same checkpoint, written with a trailing separator. + runToCompletion( + df.writeStream + .option("checkpointLocation", checkpointPath + "/") + .format("paimon") + .start(location)) + + checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil) + assert( + latestCommitUser("T") == firstCommitUser, + "the same query resuming the same checkpoint must keep its commit user") + } + } + } + + test("Paimon Sink: maintenance of a micro-batch runs while the query is alive") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The committer of a query lives until the query terminates, so the maintenance a + // commit starts runs on while the next micro-batch is written. A committer closed after + // every micro-batch would cancel an async expiration before it ever ran. + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + "'snapshot.expire.execution-mode' = 'async', " + + "'snapshot.num-retained.min' = '1', " + + "'snapshot.num-retained.max' = '1')") + val location = loadTable("T").location().toString + + val inputData = MemoryStream[(Int, String)] + val df = inputData.toDS().toDF("a", "b") + val query = df.writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + try { + for (i <- 1 to 4) { + inputData.addData((i, s"v$i")) + query.processAllAvailable() + } + // Bounded by the timeout of this test. + while (snapshotCount("T") > 1) { + Thread.sleep(50) + } + } finally { + query.stop() + } + } + } + } + + test("Paimon Sink: one committer commits every micro-batch of a query and closes with it") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // The committer of a query is created with its first micro-batch and closed when the + // query terminates, the way a Flink committer lives across checkpoints, so that the + // maintenance a commit starts is not cancelled and only the first micro-batch of a run + // has to look up what a previous run committed. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + + val inputData = MemoryStream[(Int, String)] + val query = inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + try { + for (i <- 1 to 3) { + inputData.addData((i, s"v$i")) + query.processAllAvailable() + } + assert( + CountingCommitCallback.commits == 3, + s"expected three committed micro-batches, found ${CountingCommitCallback.commits}") + assert( + CountingCommitCallback.created == 1, + s"expected one committer for the whole query, found " + + s"${CountingCommitCallback.created}" + ) + assert( + CountingCommitCallback.closed == 0, + "the committer must not be closed while the query is running") + } finally { + query.stop() + } + + // The query terminates asynchronously; bounded by the timeout of this test. + while (CountingCommitCallback.closed == 0) { + Thread.sleep(50) + } + assert(CountingCommitCallback.created == 1) + } + } + } + test("Paimon Sink: a late termination of the previous run leaves the restarted run open") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + // A restarted query keeps its id, and Spark delivers terminations asynchronously: the + // termination of the previous run may arrive while the restarted run is writing. It + // must close the committer of the previous run only. + CountingCommitCallback.reset() + spark.sql( + "CREATE TABLE T (a INT, b STRING) TBLPROPERTIES (" + + s"'commit.callbacks' = '${classOf[CountingCommitCallback].getName}')") + val location = loadTable("T").location().toString + + val inputData = MemoryStream[(Int, String)] + def start(): StreamingQuery = + inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + + // Holds back a progress event of the first run, and with it the termination of that run + // that Spark queues behind it. + val firstRun = new AtomicReference[UUID]() + val release = new CountDownLatch(1) + val blocker = new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = { + if (event.progress.runId == firstRun.get()) { + release.await(streamingTimeout.toMillis, TimeUnit.MILLISECONDS) + } + } + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = {} + } + // Registered after the sink listeners, so it sees a termination once they have. + val delivered = Collections.synchronizedSet(new java.util.HashSet[UUID]()) + val observer = new StreamingQueryListener { + override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {} + + override def onQueryProgress( + event: StreamingQueryListener.QueryProgressEvent): Unit = {} + + override def onQueryTerminated( + event: StreamingQueryListener.QueryTerminatedEvent): Unit = { + delivered.add(event.runId) + } + } + def awaitDelivered(runId: UUID): Unit = { + // Bounded by the timeout of this test. + while (!delivered.contains(runId)) { + Thread.sleep(50) + } + } + + spark.streams.addListener(blocker) + try { + val first = start() + firstRun.set(first.runId) + inputData.addData((1, "a")) + first.processAllAvailable() + first.stop() + + val restarted = start() + try { + assert(restarted.id == first.id && restarted.runId != first.runId) + inputData.addData((2, "b")) + restarted.processAllAvailable() + spark.streams.addListener(observer) + + release.countDown() + awaitDelivered(first.runId) + assert(restarted.isActive) + assert( + CountingCommitCallback.created == 2 && CountingCommitCallback.closed == 1, + "the termination of the first run must close its own committer only, but " + + s"${CountingCommitCallback.closed} of ${CountingCommitCallback.created} " + + "committers are closed" + ) + + inputData.addData((3, "c")) + restarted.processAllAvailable() + } finally { + restarted.stop() + } + awaitDelivered(restarted.runId) + assert( + CountingCommitCallback.created == 2 && CountingCommitCallback.closed == 2, + s"${CountingCommitCallback.closed} of ${CountingCommitCallback.created} " + + "committers are closed after both runs terminated" + ) + checkAnswer( + spark.sql("SELECT * FROM T ORDER BY a"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + } finally { + release.countDown() + spark.streams.removeListener(blocker) + spark.streams.removeListener(observer) + } + } + } + } +} + +/** Counts the committers of a table, and what they commit. */ +private[spark] class CountingCommitCallback extends CommitCallback { + + CountingCommitCallback.createdCount.incrementAndGet() + + override def call(context: CommitCallback.Context): Unit = { + CountingCommitCallback.commitCount.incrementAndGet() + } + + override def retry(committable: ManifestCommittable): Unit = {} + + override def close(): Unit = { + CountingCommitCallback.closedCount.incrementAndGet() + } +} + +private[spark] object CountingCommitCallback { + + private[spark] val createdCount = new AtomicInteger(0) + private[spark] val commitCount = new AtomicInteger(0) + private[spark] val closedCount = new AtomicInteger(0) + + def reset(): Unit = { + createdCount.set(0) + commitCount.set(0) + closedCount.set(0) + } + + def created: Int = createdCount.get() + + def commits: Int = commitCount.get() + + def closed: Int = closedCount.get() +} + +/** Makes a streaming write fail while writing, before anything is committed. */ +private[spark] object FailBeforeCommit { + @volatile var armed: Boolean = false +} + +/** A catalog whose first partition registration fails, the way a metastore RPC can. */ +private[spark] object FailOnceRegistration { + + @volatile private var wrapped: Catalog = _ + @volatile var attempts: Int = 0 + val registered: java.util.Set[String] = + Collections.synchronizedSet(new java.util.HashSet[String]()) + + def reset(catalog: Catalog): Unit = { + wrapped = catalog + attempts = 0 + registered.clear() + } + + // Refers to this object only, so that the loader stays serializable with the table. + val loader: CatalogLoader = () => new FailOnceCatalog(wrapped) + + private class FailOnceCatalog(catalog: Catalog) extends DelegateCatalog(catalog) { + + override def catalogLoader(): CatalogLoader = loader + + override def createPartitions( + identifier: Identifier, + partitions: JList[JMap[String, String]]): Unit = { + attempts += 1 + if (attempts == 1) { + throw new RuntimeException("metastore unavailable") + } + partitions.asScala.foreach(p => registered.add(p.get("dt"))) + } + + override def alterPartitions( + identifier: Identifier, + partitions: JList[org.apache.paimon.partition.PartitionStatistics]): Unit = {} + + override def dropPartitions( + identifier: Identifier, + partitions: JList[JMap[String, String]]): Unit = {} + } +} + +/** Fails after opening and partially writing a marker, as a storage failure can. */ +private[spark] class FailingMarkerFileSystem extends RawLocalFileSystem { + override protected def createOutputStreamWithMode( + path: Path, + append: Boolean, + permission: FsPermission): OutputStream = { + val out = super.createOutputStreamWithMode(path, append, permission) + if (FailingMarkerFileSystem.failWrites) { + out.write("partial".getBytes("UTF-8")) + out.close() + throw new IOException("marker write interrupted") + } + out + } +} + +private[spark] object FailingMarkerFileSystem { + @volatile var failWrites: Boolean = false +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala index f22dab4b9cd0..bbbfa7cc7081 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala @@ -364,6 +364,61 @@ class PaimonSinkTest extends PaimonSparkTestBase with StreamTest { } } + test("Paimon Sink: full compaction of a micro-batch is recognised by a compacted-full scan") { + failAfter(streamingTimeout) { + withTempDir { + checkpointDir => + spark.sql(s""" + |CREATE TABLE T (a INT, b INT) + |TBLPROPERTIES ( + | 'primary-key'='a', + | 'bucket'='1', + | 'full-compaction.delta-commits'='3' + |) + |""".stripMargin) + val table = loadTable("T") + val location = table.location().toString + + val inputData = MemoryStream[(Int, Int)] + val stream = inputData + .toDS() + .toDF("a", "b") + .writeStream + .option("checkpointLocation", checkpointDir.getCanonicalPath) + .format("paimon") + .start(location) + + try { + // The third micro-batch is scheduled for a full compaction, the fourth is not. + for (b <- 100 to 103) { + inputData.addData((1, b)) + stream.processAllAvailable() + } + } finally { + stream.stop() + } + // Micro-batch n is committed under identifier n + 1, the way Flink numbers checkpoints; + // that is what a compacted-full scan recognises a scheduled full compaction by. + val snapshots = table.snapshotManager() + assert( + snapshots.latestSnapshot().commitKind == APPEND, + "the last micro-batch must not have compacted") + assert(snapshots.latestSnapshot().commitIdentifier() == 4) + val fullCompaction = snapshots.snapshot(snapshots.latestSnapshotId() - 1) + assert(fullCompaction.commitKind == COMPACT) + assert(fullCompaction.commitIdentifier() == 3) + + // A compacted-full scan reads the latest full compaction, that is the state after the + // third micro-batch, and must recognise it by the commit identifier the sink published. + val compactedFull = spark.read + .format("paimon") + .option("scan.mode", "compacted-full") + .load(location) + checkAnswer(compactedFull, Row(1, 102) :: Nil) + } + } + } + test("Paimon Sink: batch then stream should not overwrite batch data") { failAfter(streamingTimeout) { withTempDir {