diff --git a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java index 71cebe7f7a8d..df2124d4ea44 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java @@ -46,8 +46,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import static java.util.Collections.emptyList; - /** Compact manager for {@link AppendOnlyFileStore}. */ public class BucketedAppendCompactManager extends CompactFutureManager { @@ -115,15 +113,15 @@ private void triggerFullCompaction() { LOG.debug("Submit full compaction with these files {}", toCompact); } - taskFuture = - executor.submit( - new FullCompactTask( - dvMaintainer, - toCompact, - compactionFileSize, - forceRewriteAllFiles, - rewriter, - metricsReporter)); + submitTask( + executor, + new FullCompactTask( + dvMaintainer, + toCompact, + compactionFileSize, + forceRewriteAllFiles, + rewriter, + metricsReporter)); recordCompactionsQueuedRequest(); compacting = new ArrayList<>(toCompact); toCompact.clear(); @@ -147,10 +145,9 @@ private void triggerCompactionWithBestEffort() { LOG.debug("Submit normal compaction with these files {}", compacting); } - taskFuture = - executor.submit( - new AutoCompactTask( - dvMaintainer, compacting, rewriter, metricsReporter)); + submitTask( + executor, + new AutoCompactTask(dvMaintainer, compacting, rewriter, metricsReporter)); recordCompactionsQueuedRequest(); } } @@ -281,7 +278,7 @@ protected CompactResult doCompact() throws Exception { // do compaction if (dvMaintainer != null) { // if deletion vector enables, always trigger compaction. - return compact(dvMaintainer, toCompact, rewriter); + return compact(dvMaintainer, toCompact, rewriter, produced()); } else { // compute small files int big = 0; @@ -295,13 +292,18 @@ protected CompactResult doCompact() throws Exception { } if (forceRewriteAllFiles || (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE)) { - return compact(null, toCompact, rewriter); + return compact(null, toCompact, rewriter, produced()); } else { - return result(emptyList(), emptyList()); + return produced(); } } } + @Override + protected void deleteProduced(List files) { + rewriter.delete(files); + } + private boolean hasDeletionFile(DataFileMeta file) { return dvMaintainer != null && dvMaintainer.deletionVectorOf(file.fileName()).isPresent(); @@ -334,22 +336,28 @@ public AutoCompactTask( @Override protected CompactResult doCompact() throws Exception { - return compact(dvMaintainer, toCompact, rewriter); + return compact(dvMaintainer, toCompact, rewriter, produced()); + } + + @Override + protected void deleteProduced(List files) { + rewriter.delete(files); } } private static CompactResult compact( @Nullable BucketedDvMaintainer dvMaintainer, List toCompact, - CompactRewriter rewriter) + CompactRewriter rewriter, + CompactResult toUpdate) throws Exception { List rewrite = rewriter.rewrite(toCompact); - CompactResult result = result(toCompact, rewrite); + toUpdate.merge(result(toCompact, rewrite)); if (dvMaintainer != null) { toCompact.forEach(f -> dvMaintainer.removeDeletionVectorOf(f.fileName())); - result.setDeletionFile(CompactDeletionFile.generateFiles(dvMaintainer)); + toUpdate.setDeletionFile(CompactDeletionFile.generateFiles(dvMaintainer)); } - return result; + return toUpdate; } private static CompactResult result(List before, List after) { @@ -359,5 +367,11 @@ private static CompactResult result(List before, List rewrite(List compactBefore) throws Exception; + + /** + * Delete files produced by this rewriter, used when a compaction result is discarded and + * its files can never be committed. + */ + void delete(List files); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/cluster/BucketedAppendClusterManager.java b/paimon-core/src/main/java/org/apache/paimon/append/cluster/BucketedAppendClusterManager.java index c223084d3184..a55cd8b78bc4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/cluster/BucketedAppendClusterManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/cluster/BucketedAppendClusterManager.java @@ -151,7 +151,7 @@ private void submitCompaction(CompactUnit unit) { file.fileName(), file.level(), file.fileSize())) .collect(Collectors.joining(", "))); } - taskFuture = executor.submit(task); + submitTask(executor, task); } @Override @@ -202,7 +202,14 @@ public BucketedAppendClusterTask( @Override protected CompactResult doCompact() throws Exception { List rewrite = rewriter.rewrite(toCluster); - return new CompactResult(toCluster, upgrade(rewrite)); + CompactResult result = produced(); + result.merge(new CompactResult(toCluster, upgrade(rewrite))); + return result; + } + + @Override + protected void deleteProduced(List files) { + rewriter.delete(files); } protected List upgrade(List files) { @@ -215,5 +222,11 @@ protected List upgrade(List files) { /** Compact rewriter for append-only table. */ public interface CompactRewriter { List rewrite(List compactBefore) throws Exception; + + /** + * Delete files produced by this rewriter, used when a cluster result is discarded and its + * files can never be committed. + */ + void delete(List files); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java b/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java index e43bec01630d..61d586c69832 100644 --- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java @@ -20,9 +20,12 @@ import org.apache.paimon.annotation.VisibleForTesting; +import javax.annotation.Nullable; + import java.util.Optional; import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; /** Base implementation of {@link CompactManager} which runs compaction in a separate thread. */ @@ -30,10 +33,22 @@ public abstract class CompactFutureManager implements CompactManager { protected Future taskFuture; + @Nullable private CompactTask task; + + protected void submitTask(ExecutorService executor, CompactTask task) { + this.task = task; + this.taskFuture = executor.submit(task); + } + @Override public void cancelCompaction() { - // TODO this method may leave behind orphan files if compaction is actually finished - // but some CPU work still needs to be done + if (task != null) { + // Tell the task that its output is not needed anymore before interrupting it, so that + // it deletes the files it produced no matter whether it observes the interruption. + // See CompactTask#cancel for the invariant that this must not be followed by + // prepareCommit on the same writer/maintainer. + task.cancel(); + } if (taskFuture != null && !taskFuture.isCancelled()) { taskFuture.cancel(true); } @@ -48,15 +63,21 @@ protected final Optional innerGetCompactionResult(boolean blockin throws ExecutionException, InterruptedException { if (taskFuture != null) { if (blocking || taskFuture.isDone()) { - CompactResult result; + CompactTask currentTask = task; try { - result = obtainCompactResult(); + return Optional.of(obtainCompactResult()); } catch (CancellationException e) { - return Optional.empty(); + // Cancellation may have won the race against the completion of the task, in + // which case the future has dropped a result whose files are already on disk. + // Report it so that the caller can account for them. If the task instead + // observed the cancellation, it has deleted its own output and there is + // nothing to report here. + return Optional.ofNullable( + currentTask == null ? null : currentTask.completedResult()); } finally { taskFuture = null; + task = null; } - return Optional.of(result); } } return Optional.empty(); diff --git a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java index 229b25324b2b..05c9b24179e9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java +++ b/paimon-core/src/main/java/org/apache/paimon/compact/CompactTask.java @@ -18,6 +18,7 @@ package org.apache.paimon.compact; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.operation.metrics.CompactionMetrics; import org.apache.paimon.operation.metrics.MetricUtils; @@ -27,7 +28,10 @@ import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.Callable; /** Compact task. */ @@ -38,11 +42,85 @@ public abstract class CompactTask implements Callable { @Nullable private final CompactionMetrics.Reporter metricsReporter; private final String bucketInfo; + /** + * Output of the steps of this task which have already finished. Subclasses producing files in + * more than one step must accumulate into this result, otherwise files written by the finished + * steps are leaked when a later step fails. + */ + private final CompactResult produced = new CompactResult(); + + /** + * Makes publishing a result and declaring it cancelled mutually exclusive, so that a result is + * never both dropped by the canceller and kept by the task. Only ever held for field + * assignments, never across compaction or file deletion. + */ + private final Object publishLock = new Object(); + + private boolean cancelled = false; + @Nullable private CompactResult completedResult = null; + public CompactTask(@Nullable CompactionMetrics.Reporter metricsReporter, String bucketInfo) { this.metricsReporter = metricsReporter; this.bucketInfo = bucketInfo; } + protected CompactResult produced() { + return produced; + } + + /** + * Declare that the result of this task will be discarded, so that the task deletes the files it + * produced instead of leaving them behind. This must be invoked before interrupting the task, + * because a task doing pure CPU work may run to completion without ever observing the + * interruption. + * + *

Invariant: the caller must not let a cancelled/discarded result be consumed by the same + * writer's {@code prepareCommit}. Compaction may already have mutated shared in-memory state + * (for example {@code BucketedDvMaintainer} or clustering key index) before the result is + * published; {@link #discard} only deletes produced files and does not roll that state back. + * Today {@code cancelCompaction} is only invoked from writer {@code close()}, so the maintainer + * is thrown away with the writer; do not call this from a path that continues writing with the + * same maintainer. + */ + public void cancel() { + synchronized (publishLock) { + cancelled = true; + } + } + + /** + * Result of a task which ran to completion, {@code null} if this task has not produced a + * complete result. {@link java.util.concurrent.FutureTask} silently drops the value returned by + * {@link #call} when cancellation wins the race against its own completion, this field keeps + * the result reachable so that its files can still be accounted for. + */ + @Nullable + public CompactResult completedResult() { + synchronized (publishLock) { + return completedResult; + } + } + + /** + * Make the finished result available to {@link #completedResult}, unless this task has already + * been cancelled. Returns {@code false} when the result must be discarded by the task itself. + * + *

This is the single point where the ownership of the produced files is decided: either the + * task publishes first and {@link #cancel} arrives too late to drop the files silently, or + * {@link #cancel} wins and the task is the one which deletes them. There is no state in which + * both sides believe the other one takes care of the files. + */ + @VisibleForTesting + protected boolean publish(CompactResult result) { + synchronized (publishLock) { + if (cancelled) { + return false; + } + completedResult = result; + return true; + } + } + @Override public CompactResult call() throws Exception { MetricUtils.safeCall(this::startTimer, LOG); @@ -55,6 +133,18 @@ public CompactResult call() throws Exception { CompactResult result = doCompact(); long durationMs = System.currentTimeMillis() - startMillis; + // Publish before doing anything else, so that no work done afterwards (metrics, + // logging) can widen the window in which a concurrent cancellation goes unnoticed. + if (!publish(result)) { + LOG.info( + "Paimon compact task was cancelled after it finished: {}, taskType={}. " + + "Deleting its output because nobody will consume the result.", + bucketInfo, + getClass().getSimpleName()); + discard(); + return new CompactResult(); + } + MetricUtils.safeCall( () -> { if (metricsReporter != null) { @@ -95,6 +185,7 @@ public CompactResult call() throws Exception { bucketInfo, getClass().getSimpleName(), e); + discard(); throw e; } finally { MetricUtils.safeCall(this::stopTimer, LOG); @@ -102,6 +193,53 @@ public CompactResult call() throws Exception { } } + /** + * Delete the files of a result which will never be committed. Note that an output file can be + * the very same physical file as one of the inputs when it is only upgraded to another level, + * such a file is still required by previous snapshots and must be kept. + * + *

Only file-side cleanup is performed here. In-memory side effects applied during {@link + * #doCompact} (deletion-vector removals, clustering key-index updates, and similar) are not + * restored; see {@link #cancel()} for the caller invariant that makes this safe. + */ + private void discard() { + Set inputs = new HashSet<>(); + for (DataFileMeta file : produced.before()) { + inputs.add(file.fileName()); + } + List files = new ArrayList<>(produced.changelog()); + for (DataFileMeta file : produced.after()) { + if (!inputs.contains(file.fileName())) { + files.add(file); + } + } + + // Cancellation interrupts this thread, and file systems backed by RPC (HDFS, object + // stores) fail their calls immediately while the interrupt flag is set. Clear it for the + // duration of the deletions, otherwise the cleanup silently does nothing and leaves + // exactly the orphan files it is supposed to remove. + boolean wasInterrupted = Thread.interrupted(); + try { + deleteProduced(files); + if (produced.deletionFile() != null) { + produced.deletionFile().clean(); + } + } catch (Exception e) { + LOG.warn( + "Failed to delete the output of a discarded compact task: {}, taskType={}", + bucketInfo, + getClass().getSimpleName(), + e); + } finally { + if (wasInterrupted) { + Thread.currentThread().interrupt(); + } + } + } + + /** Delete files produced by this task, invoked when its result is discarded. */ + protected abstract void deleteProduced(List files); + private void decreaseCompactionsQueuedCount() { if (metricsReporter != null) { metricsReporter.decreaseCompactionsQueuedCount(); @@ -133,7 +271,8 @@ protected String logMetric( } /** - * Perform compaction. + * Perform compaction. Implementations must accumulate their output into {@link #produced()} and + * return it, so that a partial result of a failed task is still known and can be cleaned up. * * @return {@link CompactResult} of compact before and compact after files. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java index af58720d9667..a0c649c8afd8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java @@ -53,4 +53,10 @@ CompactResult rewrite(int outputLevel, boolean dropDelete, List> * @throws Exception exception */ CompactResult upgrade(int outputLevel, DataFileMeta file) throws Exception; + + /** + * Delete files produced by this rewriter, used when a compaction result is discarded and its + * files can never be committed. + */ + void deleteProduced(List files); } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java index 0e52dbf14c9a..a7d8c9de00fa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FileRewriteCompactTask.java @@ -59,7 +59,7 @@ public FileRewriteCompactTask( @Override protected CompactResult doCompact() throws Exception { - CompactResult result = new CompactResult(); + CompactResult result = produced(); for (DataFileMeta file : files) { rewriteFile(file, result); } @@ -67,6 +67,11 @@ protected CompactResult doCompact() throws Exception { return result; } + @Override + protected void deleteProduced(List producedFiles) { + rewriter.deleteProduced(producedFiles); + } + private void rewriteFile(DataFileMeta file, CompactResult toUpdate) throws Exception { List> candidate = singletonList(singletonList(SortedRun.fromSingle(file))); toUpdate.merge(rewriter.rewrite(outputLevel, dropDelete, candidate)); diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java index 708515bd014f..8dff3aaec287 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java @@ -255,7 +255,7 @@ private void submitCompaction(CompactUnit unit, boolean dropDelete) { file.fileName(), file.level(), file.fileSize())) .collect(Collectors.joining(", "))); } - taskFuture = executor.submit(task); + submitTask(executor, task); if (metricsReporter != null) { metricsReporter.increaseCompactionsQueuedCount(); metricsReporter.increaseCompactionsTotalCount(); diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java index 9eecce5248c4..bab2df7be3e9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactRewriter.java @@ -127,6 +127,11 @@ protected RecordReader readerForMergeTree( mergeSorter); } + @Override + public void deleteProduced(List files) { + files.forEach(writerFactory::deleteFile); + } + protected void notifyRewriteCompactBefore(List files) {} protected List notifyRewriteCompactAfter(List files) { diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java index db6d8e23e831..9947236ed14e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java @@ -82,7 +82,7 @@ public MergeTreeCompactTask( @Override protected CompactResult doCompact() throws Exception { List> candidate = new ArrayList<>(); - CompactResult result = new CompactResult(); + CompactResult result = produced(); // Checking the order and compacting adjacent and contiguous files // Note: can't skip an intermediate file to compact, this will destroy the overall @@ -113,6 +113,11 @@ protected CompactResult doCompact() throws Exception { return result; } + @Override + protected void deleteProduced(List files) { + rewriter.deleteProduced(files); + } + @Override protected String logMetric( long startMillis, List compactBefore, List compactAfter) { diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManager.java index 8d094209577f..d503a5890fd1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringCompactManager.java @@ -184,22 +184,25 @@ public void triggerCompaction(boolean fullCompaction) { if (taskFuture != null) { return; } - taskFuture = - executor.submit( - new CompactTask(metricsReporter, "") { - @Override - protected CompactResult doCompact() throws Exception { - return compact(fullCompaction); - } - }); + submitTask( + executor, + new CompactTask(metricsReporter, "") { + @Override + protected CompactResult doCompact() throws Exception { + return compact(fullCompaction, produced()); + } + + @Override + protected void deleteProduced(List files) { + fileRewriter.deleteProduced(files); + } + }); } - private CompactResult compact(boolean fullCompaction) throws Exception { + private CompactResult compact(boolean fullCompaction, CompactResult result) throws Exception { KeyValueSerializer kvSerializer = new KeyValueSerializer(keyType, valueType); RowType kvSchemaType = KeyValue.schema(keyType, valueType); - CompactResult result = new CompactResult(); - // Phase 1: Sort and rewrite all unsorted (level 0) files List unsortedFiles = fileLevels.unsortedFiles(); // Snapshot sorted files before Phase 1 to avoid including newly created files in Phase 2 diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringFileRewriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringFileRewriter.java index 451d7c75005e..07a92af1e230 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringFileRewriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/clustering/ClusteringFileRewriter.java @@ -111,6 +111,14 @@ public ClusteringFileRewriter( this.compression = compression; } + /** + * Delete files produced by this rewriter, used when a clustering result is discarded and its + * files can never be committed. + */ + public void deleteProduced(List files) { + files.forEach(writerFactory::deleteFile); + } + /** * Sort and rewrite unsorted file by clustering columns. Reads all KeyValue records, sorts them * using an external sort buffer, and writes to new level-1 files. Checks the key index inline diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java index cb2dd42cfa53..46b3204ad8bf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java @@ -260,6 +260,7 @@ public List compactRewrite( } } if (collectedExceptions != null) { + rewriter.abort(); throw collectedExceptions; } return rewriter.result(); @@ -297,12 +298,23 @@ public List clusterRewrite( } if (collectedExceptions != null) { + rewriter.abort(); throw collectedExceptions; } return rewriter.result(); } + /** + * Delete data files which can never be committed, for example the output of a compaction whose + * result has been discarded. + */ + protected void deleteFiles(BinaryRow partition, int bucket, List files) { + DataFilePathFactory dataPathFactory = + pathFactory.createDataFilePathFactory(partition, bucket); + files.forEach(file -> file.collectFiles(dataPathFactory).forEach(fileIO::deleteQuietly)); + } + private RowDataRollingFileWriter createRollingFileWriter( BinaryRow partition, int bucket, Supplier seqNumCounterSupplier) { return new RowDataRollingFileWriter( diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java index 1e1ee6abe058..b947309fa2f9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketedAppendFileStoreWrite.java @@ -109,7 +109,18 @@ protected CompactManager getCompactManager( options.sortedRunSizeRatio(), options.numSortedRunCompactionTrigger(), options.numLevels(), - files -> clusterRewrite(partition, bucket, files)); + new BucketedAppendClusterManager.CompactRewriter() { + @Override + public List rewrite(List toCluster) + throws Exception { + return clusterRewrite(partition, bucket, toCluster); + } + + @Override + public void delete(List files) { + deleteFiles(partition, bucket, files); + } + }); } else { Function dvFactory = dvMaintainer != null @@ -123,7 +134,18 @@ protected CompactManager getCompactManager( options.targetFileSize(false), options.compactionFileSize(false), options.forceRewriteAllFiles(), - files -> compactRewrite(partition, bucket, dvFactory, files), + new BucketedAppendCompactManager.CompactRewriter() { + @Override + public List rewrite(List toCompact) + throws Exception { + return compactRewrite(partition, bucket, dvFactory, toCompact); + } + + @Override + public void delete(List files) { + deleteFiles(partition, bucket, files); + } + }, compactionMetrics == null ? null : compactionMetrics.createReporter(partition, bucket)); diff --git a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java index b670ffa48f0b..ddb012d44331 100644 --- a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java +++ b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java @@ -363,8 +363,11 @@ public List commitDataImpl( .forEach( w -> { try { - // wait for compaction to end, otherwise orphan files may occur - // see CompactManager#cancelCompaction for more info + // Wait for compaction to end before closing. Closing cancels an + // in-flight compaction and returns as soon as its future is + // cancelled, while the compaction thread is still unwinding and + // holding its readers open, which makes the assertions on open + // streams of the tests racy. w.sync(); w.close(); } catch (Exception e) { diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java index 63324e7a841c..4f3b502efdcf 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java @@ -192,7 +192,7 @@ public void testBinaryColumnStatsRoundTrip() throws Exception { true, true, Collections.emptyList(), - compactBefore -> Collections.emptyList(), + rewriter(compactBefore -> Collections.emptyList()), options) .getKey(); @@ -1221,12 +1221,14 @@ private Pair> createWriter( spillable, hasIoManager, scannedFiles, - compactBefore -> { - latch.await(); - return compactBefore.isEmpty() - ? Collections.emptyList() - : Collections.singletonList(generateCompactAfter(compactBefore)); - }, + rewriter( + compactBefore -> { + latch.await(); + return compactBefore.isEmpty() + ? Collections.emptyList() + : Collections.singletonList( + generateCompactAfter(compactBefore)); + }), options); } @@ -1244,11 +1246,31 @@ private AppendOnlyWriter createVectorStoreWriter( false, true, Collections.emptyList(), - compactBefore -> Collections.emptyList(), + rewriter(compactBefore -> Collections.emptyList()), options) .getKey(); } + /** + * A rewrite function whose output is not backed by real files, so nothing has to be deleted. + */ + @FunctionalInterface + private interface Rewrite { + List rewrite(List compactBefore) throws Exception; + } + + private static BucketedAppendCompactManager.CompactRewriter rewriter(Rewrite rewrite) { + return new BucketedAppendCompactManager.CompactRewriter() { + @Override + public List rewrite(List compactBefore) throws Exception { + return rewrite.rewrite(compactBefore); + } + + @Override + public void delete(List files) {} + }; + } + private Pair> createWriterBase( long targetFileSize, FileFormat vectorFileFormat, diff --git a/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java b/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java index 2877247145dc..061e26808c7e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/FullCompactTaskTest.java @@ -134,25 +134,33 @@ public MockFullCompactTask( } private BucketedAppendCompactManager.CompactRewriter rewriter() { - return compactBefore -> { - List compactAfter = new ArrayList<>(); - long totalFileSize = 0L; - long minSeq = -1L; - for (int i = 0; i < compactBefore.size(); i++) { - DataFileMeta file = compactBefore.get(i); - if (i == 0) { - minSeq = file.minSequenceNumber(); - } - totalFileSize += file.fileSize(); - if (totalFileSize >= TARGET_FILE_SIZE) { - compactAfter.add(newFile(minSeq, minSeq + TARGET_FILE_SIZE - 1)); - minSeq += TARGET_FILE_SIZE; - } - if (i == compactBefore.size() - 1 && minSeq <= file.maxSequenceNumber()) { - compactAfter.add(newFile(minSeq, file.maxSequenceNumber())); + return new BucketedAppendCompactManager.CompactRewriter() { + @Override + public List rewrite(List compactBefore) { + List compactAfter = new ArrayList<>(); + long totalFileSize = 0L; + long minSeq = -1L; + for (int i = 0; i < compactBefore.size(); i++) { + DataFileMeta file = compactBefore.get(i); + if (i == 0) { + minSeq = file.minSequenceNumber(); + } + totalFileSize += file.fileSize(); + if (totalFileSize >= TARGET_FILE_SIZE) { + compactAfter.add(newFile(minSeq, minSeq + TARGET_FILE_SIZE - 1)); + minSeq += TARGET_FILE_SIZE; + } + if (i == compactBefore.size() - 1 && minSeq <= file.maxSequenceNumber()) { + compactAfter.add(newFile(minSeq, file.maxSequenceNumber())); + } } + return compactAfter; + } + + @Override + public void delete(List files) { + // the output of this rewriter is not backed by real files } - return compactAfter; }; } } diff --git a/paimon-core/src/test/java/org/apache/paimon/append/cluster/BucketedAppendClusterManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/cluster/BucketedAppendClusterManagerTest.java index 5efdee2d7017..37074f862ca0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/cluster/BucketedAppendClusterManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/cluster/BucketedAppendClusterManagerTest.java @@ -30,6 +30,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.operation.BaseAppendFileStoreWrite; import org.apache.paimon.reader.RecordReaderIterator; import org.apache.paimon.schema.Schema; @@ -76,6 +77,27 @@ public void before() throws Exception { } } + private BucketedAppendClusterManager.CompactRewriter clusterRewriter() { + return new BucketedAppendClusterManager.CompactRewriter() { + @Override + public List rewrite(List compactBefore) throws Exception { + return write.clusterRewrite(BinaryRow.EMPTY_ROW, 0, compactBefore); + } + + @Override + public void delete(List files) { + DataFilePathFactory pathFactory = + table.store() + .pathFactory() + .createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0); + files.forEach( + file -> + file.collectFiles(pathFactory) + .forEach(table.fileIO()::deleteQuietly)); + } + }; + } + @Test public void testBucketedAppendClusterTask() throws Exception { List toCluster = @@ -83,7 +105,7 @@ public void testBucketedAppendClusterTask() throws Exception { BucketedAppendClusterManager.BucketedAppendClusterTask task = new BucketedAppendClusterManager.BucketedAppendClusterTask( - toCluster, 5, files -> write.clusterRewrite(BinaryRow.EMPTY_ROW, 0, files)); + toCluster, 5, clusterRewriter()); CompactResult result = task.doCompact(); assertThat(result.before().size()).isEqualTo(9); @@ -121,7 +143,7 @@ public void testTriggerCompaction() throws Exception { options.sortedRunSizeRatio(), options.numSortedRunCompactionTrigger(), options.numLevels(), - files -> write.clusterRewrite(BinaryRow.EMPTY_ROW, 0, files)); + clusterRewriter()); assertThat(manager.levels().levelSortedRuns().size()).isEqualTo(9); manager.triggerCompaction(false); diff --git a/paimon-core/src/test/java/org/apache/paimon/compact/CompactCancellationTest.java b/paimon-core/src/test/java/org/apache/paimon/compact/CompactCancellationTest.java new file mode 100644 index 000000000000..c0f5aafcda14 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/compact/CompactCancellationTest.java @@ -0,0 +1,406 @@ +/* + * 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.compact; + +import org.apache.paimon.io.DataFileMeta; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +import static java.util.Collections.singletonList; +import static org.apache.paimon.io.DataFileTestUtils.newFile; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests that the output of a compaction whose result is discarded does not become an orphan file. + */ +public class CompactCancellationTest { + + private static final DataFileMeta INPUT = newFile("input", 0, 0, 10, 1); + private static final DataFileMeta OUTPUT = newFile("output", 1, 0, 10, 1); + private static final DataFileMeta CHANGELOG = newFile("changelog", 0, 0, 10, 1); + + private ExecutorService executor; + + @BeforeEach + public void before() { + executor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + public void after() { + executor.shutdownNow(); + } + + @Test + public void testCancelledTaskDeletesItsOutput() throws Exception { + TestTask task = new TestTask(result -> result.merge(rewriteResult())); + task.cancel(); + + assertThat(task.call().after()).isEmpty(); + assertThat(task.completedResult()).isNull(); + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + } + + @Test + public void testFailedTaskDeletesOutputOfFinishedSteps() { + TestTask task = + new TestTask( + result -> { + result.merge(rewriteResult()); + throw new IllegalStateException("failed in a later step"); + }); + + assertThatThrownBy(task::call).hasMessageContaining("failed in a later step"); + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + } + + @Test + public void testUpgradedFileIsNotDeleted() { + // an upgraded file is the very same physical file as its input, it is still required by + // previous snapshots + TestTask task = + new TestTask( + result -> { + result.merge(new CompactResult(INPUT, INPUT.upgrade(3))); + throw new IllegalStateException("boom"); + }); + + assertThatThrownBy(task::call).hasMessageContaining("boom"); + assertThat(task.deleted).isEmpty(); + } + + @Test + public void testDeletionFileIsCleanedUp() throws Exception { + TestDeletionFile deletionFile = new TestDeletionFile(); + TestTask task = new TestTask(result -> result.setDeletionFile(deletionFile)); + task.cancel(); + + task.call(); + assertThat(deletionFile.cleaned).isTrue(); + } + + @Test + public void testFinishedResultIsNotLostWhenCancellationWinsTheRace() throws Exception { + TestManager manager = new TestManager(); + TestTask task = new TestTask(result -> result.merge(rewriteResult())); + manager.submit(task); + task.awaitExit(); + + // the task has finished, but FutureTask dropped its result because cancellation won the + // race against the completion of the task + manager.simulateCancellation = true; + Optional result = manager.getCompactionResult(true); + + assertThat(result).isPresent(); + assertThat(names(result.get().after())).containsExactly("output"); + // the caller is now aware of the files, the task must not have deleted them + assertThat(task.deleted).isEmpty(); + } + + @Test + public void testCancelledTaskCleansUpAfterTheCallerGaveUpOnTheResult() throws Exception { + TestManager manager = new TestManager(); + CountDownLatch blocked = new CountDownLatch(1); + TestTask task = + new TestTask( + result -> { + result.merge(rewriteResult()); + blocked.countDown(); + // interrupted by the cancellation below + Thread.sleep(Long.MAX_VALUE); + }); + manager.submit(task); + blocked.await(); + + manager.simulateCancellation = true; + manager.cancelCompaction(); + + // the caller cannot wait for a task which may be doing a long piece of CPU work + assertThat(manager.getCompactionResult(true)).isEmpty(); + + // so the task itself is responsible for the files it has produced + task.awaitExit(); + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + } + + @Test + public void testCancellationArrivingWhenTheTaskIsAboutToPublishItsResult() throws Exception { + TestManager manager = new TestManager(); + CountDownLatch submitted = new CountDownLatch(1); + AtomicReference> reported = new AtomicReference<>(); + TestTask task = + new TestTask( + result -> { + assertThat(submitted.await(1, TimeUnit.MINUTES)).isTrue(); + result.merge(rewriteResult()); + }); + // the writer cancels the compaction and gives up on its result in the very moment the + // task is about to publish it + task.beforePublish = + () -> { + manager.cancelCompaction(); + reported.set(manager.getCompactionResult(true)); + }; + + manager.submit(task); + submitted.countDown(); + task.awaitExit(); + + // nobody else can account for the files, so the task must have deleted them + assertThat(reported.get()).isEmpty(); + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + } + + @Test + public void testOutputIsNeverLostWhenCancellationRacesWithCompletion() throws Exception { + for (int i = 0; i < 500; i++) { + TestManager manager = new TestManager(); + CountDownLatch started = new CountDownLatch(1); + int spins = i % 8; + TestTask task = + new TestTask( + result -> { + started.countDown(); + // shift the phase between the two threads to hit different + // interleavings around the publication of the result + for (int j = 0; j < spins; j++) { + Thread.yield(); + } + result.merge(rewriteResult()); + }); + manager.submit(task); + // cancelling a task which the executor has not picked up yet keeps it from running + // at all, so wait until it is really in flight + assertThat(started.await(1, TimeUnit.MINUTES)).isTrue(); + + manager.cancelCompaction(); + Optional reported = manager.getCompactionResult(true); + task.awaitExit(); + + // whoever ends up owning the files, they are never left behind unnoticed: the + // caller either learns about them or the task has deleted them itself + boolean reportedToCaller = reported.isPresent() && !reported.get().after().isEmpty(); + if (reportedToCaller) { + assertThat(names(reported.get().after())).containsExactly("output"); + assertThat(task.deleted).isEmpty(); + } else { + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + } + } + } + + @Test + public void testCleanupIsNotSkippedByTheCancellationInterrupt() throws Exception { + TestManager manager = new TestManager(); + CountDownLatch blocked = new CountDownLatch(1); + TestTask task = + new TestTask( + result -> { + result.merge(rewriteResult()); + blocked.countDown(); + // a task busy with CPU work observes the interruption without + // clearing it, just like a file system whose RPC fails while the + // interrupt flag is set + long deadline = System.currentTimeMillis() + 60_000; + while (!Thread.currentThread().isInterrupted() + && System.currentTimeMillis() < deadline) { + Thread.yield(); + } + }); + manager.submit(task); + assertThat(blocked.await(1, TimeUnit.MINUTES)).isTrue(); + + manager.cancelCompaction(); + task.awaitExit(); + + // deleting files goes through the file IO, whose calls fail immediately on an + // interrupted thread, so the flag must be cleared for the cleanup + assertThat(task.interruptedDuringCleanup).isFalse(); + assertThat(deletedNames(task)).containsExactlyInAnyOrder("output", "changelog"); + // and restored afterwards, the interruption must not be swallowed + assertThat(task.interruptedAfterCall).isTrue(); + } + + private static CompactResult rewriteResult() { + return new CompactResult( + singletonList(INPUT), singletonList(OUTPUT), singletonList(CHANGELOG)); + } + + private static List deletedNames(TestTask task) { + return names(task.deleted); + } + + private static List names(List files) { + return files.stream().map(DataFileMeta::fileName).collect(Collectors.toList()); + } + + /** A body of a {@link CompactTask} which is allowed to fail. */ + @FunctionalInterface + private interface TaskBody { + void run(CompactResult produced) throws Exception; + } + + /** An action running at a given point of a {@link CompactTask}, allowed to fail. */ + @FunctionalInterface + private interface Hook { + void run() throws Exception; + } + + private static class TestTask extends CompactTask { + + private final TaskBody body; + private final CountDownLatch exited = new CountDownLatch(1); + private final List deleted = new ArrayList<>(); + + @Nullable private volatile Hook beforePublish = null; + private volatile boolean interruptedDuringCleanup = false; + private volatile boolean interruptedAfterCall = false; + + private TestTask(TaskBody body) { + super(null, ""); + this.body = body; + } + + @Override + public CompactResult call() throws Exception { + try { + return super.call(); + } finally { + interruptedAfterCall = Thread.currentThread().isInterrupted(); + exited.countDown(); + } + } + + @Override + protected boolean publish(CompactResult result) { + Hook hook = beforePublish; + if (hook != null) { + try { + hook.run(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return super.publish(result); + } + + private void awaitExit() throws InterruptedException { + assertThat(exited.await(1, TimeUnit.MINUTES)).isTrue(); + } + + @Override + protected CompactResult doCompact() throws Exception { + body.run(produced()); + return produced(); + } + + @Override + protected void deleteProduced(List files) { + interruptedDuringCleanup = Thread.currentThread().isInterrupted(); + deleted.addAll(files); + } + } + + private class TestManager extends CompactFutureManager { + + private volatile boolean simulateCancellation = false; + + private void submit(CompactTask task) { + submitTask(executor, task); + } + + @Override + protected CompactResult obtainCompactResult() + throws InterruptedException, ExecutionException { + if (simulateCancellation) { + throw new CancellationException(); + } + return super.obtainCompactResult(); + } + + @Override + public Optional getCompactionResult(boolean blocking) + throws ExecutionException, InterruptedException { + return innerGetCompactionResult(blocking); + } + + @Override + public boolean shouldWaitForLatestCompaction() { + return false; + } + + @Override + public boolean shouldWaitForPreparingCheckpoint() { + return false; + } + + @Override + public void addNewFile(DataFileMeta file) {} + + @Override + public Collection allFiles() { + return new ArrayList<>(); + } + + @Override + public void triggerCompaction(boolean fullCompaction) {} + + @Override + public void close() throws IOException {} + } + + private static class TestDeletionFile implements CompactDeletionFile { + + private boolean cleaned = false; + + @Override + public Optional getOrCompute() { + return Optional.empty(); + } + + @Override + public CompactDeletionFile mergeOldFile(CompactDeletionFile old) { + throw new UnsupportedOperationException(); + } + + @Override + public void clean() { + cleaned = true; + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java index 67e787770f8e..a1043edd5c5c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java @@ -23,6 +23,7 @@ import org.apache.paimon.CoreOptions.SortEngine; import org.apache.paimon.KeyValue; import org.apache.paimon.compact.CompactResult; +import org.apache.paimon.compact.CompactUnit; import org.apache.paimon.compression.CompressOptions; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericRow; @@ -43,8 +44,10 @@ import org.apache.paimon.mergetree.compact.CompactRewriter; import org.apache.paimon.mergetree.compact.CompactStrategy; import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction; +import org.apache.paimon.mergetree.compact.FileRewriteCompactTask; import org.apache.paimon.mergetree.compact.IntervalPartition; import org.apache.paimon.mergetree.compact.MergeTreeCompactManager; +import org.apache.paimon.mergetree.compact.MergeTreeCompactRewriter; import org.apache.paimon.mergetree.compact.ReducerMergeFunctionWrapper; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; @@ -85,6 +88,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; @@ -92,6 +96,7 @@ import static org.apache.paimon.mergetree.compact.UniversalCompactionTest.ofTesting; import static org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link MergeTreeReaders} and {@link MergeTreeWriter}. */ public abstract class MergeTreeTestBase { @@ -325,6 +330,63 @@ interface RunnableWithException { void run() throws Exception; } + @Test + public void testDiscardedCompactionDeletesTheFilesItProduced() throws Exception { + List inputs = generateDataFileToCommit(); + assertThat(inputs.size()).isGreaterThan(1); + + // Rewrite the first input file successfully and fail on the second one, so that the files + // written by the finished step are only reachable through the accumulated result of the + // task. + AtomicInteger rewrites = new AtomicInteger(); + List produced = new ArrayList<>(); + // the production rewriter, so that the deletion really goes through its writer factory + CompactRewriter rewriter = + new MergeTreeCompactRewriter( + compactReaderFactory, + compactWriterFactory, + comparator, + null, + DeduplicateMergeFunction.factory(), + new MergeSorter(options, null, null, null)) { + @Override + public CompactResult rewrite( + int outputLevel, boolean dropDelete, List> sections) + throws Exception { + if (rewrites.getAndIncrement() > 0) { + throw new IOException("mock a failure in a later step"); + } + CompactResult result = super.rewrite(outputLevel, dropDelete, sections); + produced.addAll(result.after()); + return result; + } + }; + + FileRewriteCompactTask task = + new FileRewriteCompactTask( + rewriter, + CompactUnit.fromFiles(options.numLevels() - 1, inputs, true), + false, + null, + () -> null, + ""); + assertThatThrownBy(task::call).hasMessageContaining("mock a failure in a later step"); + + LocalFileIO fileIO = LocalFileIO.create(); + assertThat(produced).isNotEmpty(); + for (DataFileMeta file : produced) { + assertThat(fileIO.exists(pathOf(file))).isFalse(); + } + // the inputs are still required by the previous snapshots + for (DataFileMeta file : inputs) { + assertThat(fileIO.exists(pathOf(file))).isTrue(); + } + } + + private Path pathOf(DataFileMeta file) { + return writerFactory.pathFactory(file.level()).toPath(file); + } + @ParameterizedTest @ValueSource(longs = {1, 1024 * 1024}) public void testCloseUpgrade(long targetFileSize) throws Exception { @@ -631,6 +693,11 @@ public CompactResult rewrite( writer.close(); return new CompactResult(extractFilesFromSections(sections), writer.result()); } + + @Override + public void deleteProduced(List files) { + files.forEach(writerFactory::deleteFile); + } } private static class TestRecord { diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java index 05c041521b4d..ad4999da7d3f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java @@ -584,6 +584,11 @@ public CompactResult rewrite( extractFilesFromSections(sections), Collections.singletonList(newFile(outputLevel, minKey, maxKey, maxSequence))); } + + @Override + public void deleteProduced(List files) { + // the output of this rewriter is not backed by real files + } } private static class LevelMinMax { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreTestUtils.java index 54fef645cb45..59916575a40f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreTestUtils.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreTestUtils.java @@ -117,8 +117,11 @@ public static void commitData( .forEach( w -> { try { - // wait for compaction to end, otherwise orphan files may occur - // see CompactManager#cancelCompaction for more info + // Wait for compaction to end before closing. Closing cancels an + // in-flight compaction and returns as soon as its future is + // cancelled, while the compaction thread is still unwinding and + // holding its readers open, which makes the assertions on open + // streams of the tests racy. w.sync(); w.close(); } catch (Exception e) {