From 4cdd4c4cfccd53625117757ea1f448d1d6272862 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Wed, 26 Aug 2026 16:47:33 +0800 Subject: [PATCH 1/5] [api] Read a bounded batch's input lazily and keep its running work Let sequentialBatchedExecuteCloseable take an iterator, so a caller that discovers its work by listing storage does not have to list all of it before the first task can start, and refill the window as a slot frees rather than a batch at a time. Add a variant whose close waits for a task that has already started instead of interrupting it. Interrupting a delete or a write halfway leaves the caller unable to say whether it took effect, so a caller that changes stored state cannot let close cancel what it has already handed out. Give a worker its thread's classloader back, and clear the interrupt a cancelled task may leave behind, so that neither reaches whatever the shared pool runs next. --- .../apache/paimon/utils/ThreadPoolUtils.java | 122 ++++++++++++++---- .../paimon/utils/ThreadPoolUtilsTest.java | 91 ++++++++++++- .../paimon/utils/ManifestReadThreadPool.java | 2 +- 3 files changed, 183 insertions(+), 32 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java index b5c28a19a08a..03b47b1e7239 100644 --- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java +++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java @@ -134,7 +134,7 @@ private void advanceIfNeeded() { } /** - * Parallel processes one bounded batch at a time and returns results in input order. + * Parallel processes a bounded number of inputs at a time and returns results in input order. * *

The caller must close the iterator to cancel unstarted tasks and wait for running tasks. */ @@ -143,10 +143,53 @@ public static CloseableBatchIterator sequentialBatchedExecuteCloseable Function> processor, List input, int queueSize) { + return sequentialBatchedExecuteCloseable(executor, processor, input.iterator(), queueSize); + } + + /** + * Parallel processes a lazily supplied input, keeping at most {@code queueSize} tasks in flight + * and returning results in input order. The input is read only as slots free, so a caller that + * discovers its work by listing storage does not have to list all of it up front. + * + *

The caller must close the iterator to stop reading input, cancel unstarted tasks and wait + * for running tasks. An input that fails to produce its next element reports that by throwing + * an unchecked exception, as a processor does. + */ + public static CloseableBatchIterator sequentialBatchedExecuteCloseable( + ExecutorService executor, + Function> processor, + Iterator input, + int queueSize) { + return newSequentialBatchIterator(executor, processor, input, queueSize, true); + } + + /** + * As {@link #sequentialBatchedExecuteCloseable}, but closing waits for a task that has already + * started instead of interrupting it. + * + *

Use this when a task changes stored state. Interrupting a delete or a write halfway leaves + * the caller unable to say whether it took effect, so a caller that has to know the outcome of + * everything it handed out cannot let close cancel work that is already running. + */ + public static CloseableBatchIterator sequentialBatchedExecuteNonCancellable( + ExecutorService executor, + Function> processor, + Iterator input, + int queueSize) { + return newSequentialBatchIterator(executor, processor, input, queueSize, false); + } + + private static CloseableBatchIterator newSequentialBatchIterator( + ExecutorService executor, + Function> processor, + Iterator input, + int queueSize, + boolean cancelRunningOnClose) { if (queueSize <= 0) { throw new NegativeArraySizeException("queue size should not be negative"); } - return new SequentialBatchIterator<>(executor, processor, input, queueSize); + return new SequentialBatchIterator<>( + executor, processor, input, queueSize, cancelRunningOnClose); } public static void randomlyOnlyExecute( @@ -224,7 +267,9 @@ private static class SequentialBatchIterator implements CloseableBatchIter private final ExecutorService executor; private final Function> processor; - private final Queue> batches; + private final Iterator input; + private final int queueSize; + private final boolean cancelRunningOnClose; private final Queue> activeTasks = new ArrayDeque<>(); private Iterator activeResults = Collections.emptyList().iterator(); @@ -234,11 +279,14 @@ private static class SequentialBatchIterator implements CloseableBatchIter private SequentialBatchIterator( ExecutorService executor, Function> processor, - List input, - int queueSize) { + Iterator input, + int queueSize, + boolean cancelRunningOnClose) { this.executor = executor; this.processor = processor; - this.batches = new ArrayDeque<>(Lists.partition(input, queueSize)); + this.input = input; + this.queueSize = queueSize; + this.cancelRunningOnClose = cancelRunningOnClose; } @Override @@ -263,30 +311,31 @@ private void advanceIfNeeded() { while (next == null) { if (activeResults.hasNext()) { next = activeResults.next(); - } else if (!activeTasks.isEmpty()) { - BatchTask task = activeTasks.peek(); - try { - List results = task.result(); + continue; + } + fillWindow(); + if (activeTasks.isEmpty()) { + return; + } + BatchTask task = activeTasks.peek(); + try { + List results = task.result(); + activeTasks.poll(); + activeResults = results.iterator(); + } catch (RuntimeException | Error failure) { + if (task.failureReported()) { activeTasks.poll(); - activeResults = results.iterator(); - } catch (RuntimeException | Error failure) { - if (task.failureReported()) { - activeTasks.poll(); - } - throw failure; } - } else if (batches.isEmpty()) { - return; - } else { - submitBatch(batches.poll()); + throw failure; } } } - private void submitBatch(List batch) { + /** Hands out work until the window is full or the input is exhausted. */ + private void fillWindow() { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); - for (U input : batch) { - BatchTask task = new BatchTask<>(processor, input, classLoader); + while (activeTasks.size() < queueSize && input.hasNext()) { + BatchTask task = new BatchTask<>(processor, input.next(), classLoader); executor.execute(task); activeTasks.add(task); } @@ -298,13 +347,16 @@ public synchronized void close() { return; } closed = true; - batches.clear(); Throwable failure = null; boolean interrupted = Thread.interrupted(); for (BatchTask task : activeTasks) { try { - task.cancel(); + if (cancelRunningOnClose) { + task.cancel(); + } else { + task.cancelIfUnstarted(); + } } catch (Throwable cleanupFailure) { failure = firstOrSuppressed(cleanupFailure, failure); } @@ -372,12 +424,16 @@ public void run() { runner = Thread.currentThread(); } + Thread currentThread = Thread.currentThread(); + ClassLoader originalClassLoader = currentThread.getContextClassLoader(); try { - Thread.currentThread().setContextClassLoader(classLoader); + currentThread.setContextClassLoader(classLoader); result = processor.apply(input); } catch (RuntimeException | Error taskFailure) { failure = taskFailure; } finally { + currentThread.setContextClassLoader(originalClassLoader); + Thread.interrupted(); synchronized (this) { runner = null; state = FINISHED; @@ -387,12 +443,22 @@ public void run() { } private synchronized void cancel() { + if (cancelIfUnstarted()) { + return; + } + if (state == RUNNING) { + runner.interrupt(); + } + } + + /** Gives up on a task only while it can still be given up on without interrupting it. */ + private synchronized boolean cancelIfUnstarted() { if (state == CREATED) { state = CANCELLED; completion.countDown(); - } else if (state == RUNNING) { - runner.interrupt(); + return true; } + return false; } private List result() { diff --git a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java index 65a3331fc77b..0e931e88165b 100644 --- a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java @@ -25,6 +25,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Iterator; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -82,10 +83,13 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti results.add(firstResult.get(3, TimeUnit.SECONDS)); assertThat(iterator.hasNext()).isTrue(); results.add(iterator.next()); - assertThat(workers.getTaskCount()).isEqualTo(2); - + // Taking the first result freed a slot, and the window refills as slots free rather + // than waiting for every task it handed out to be consumed. + // The window refills as a slot frees rather than a batch at a time, so how many tasks + // have been handed to the pool at a given moment is no longer aligned to batches, and + // ThreadPoolExecutor only reports that count approximately. What the iterator owes the + // caller is the bound asserted above and the order asserted below. assertThat(iterator.hasNext()).isTrue(); - assertThat(workers.getTaskCount()).isEqualTo(4); results.add(iterator.next()); assertThat(iterator.hasNext()).isTrue(); results.add(iterator.next()); @@ -101,6 +105,87 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti } } + @Test + public void testLazyInputIsReadOnlyAsSlotsFree() throws Exception { + ExecutorService workers = Executors.newFixedThreadPool(2); + CountDownLatch releaseFirst = new CountDownLatch(1); + AtomicInteger read = new AtomicInteger(); + Iterator input = + new Iterator() { + private int next; + + @Override + public boolean hasNext() { + return next < 100; + } + + @Override + public Integer next() { + read.incrementAndGet(); + return next++; + } + }; + try (CloseableBatchIterator iterator = + ThreadPoolUtils.sequentialBatchedExecuteCloseable( + workers, + value -> { + if (value == 0) { + await(releaseFirst); + } + return Collections.singletonList(value); + }, + input, + 4)) { + // A caller that discovers its work by listing storage must not be forced to list all + // of it: only a window's worth of input may be read before the first result is taken. + assertThat(read.get()).isLessThanOrEqualTo(4); + releaseFirst.countDown(); + assertThat(iterator.next()).isEqualTo(0); + assertThat(read.get()).isLessThan(100); + } finally { + workers.shutdownNow(); + } + } + + @Test + public void testWorkerRunsWithTheCallersClassLoaderAndGivesItBack() throws Exception { + ExecutorService workers = Executors.newFixedThreadPool(1); + ClassLoader callerClassLoader = new ClassLoader(getClass().getClassLoader()) {}; + AtomicReference poolClassLoader = new AtomicReference<>(); + workers.submit(() -> poolClassLoader.set(Thread.currentThread().getContextClassLoader())) + .get(10, TimeUnit.SECONDS); + + ClassLoader original = Thread.currentThread().getContextClassLoader(); + List seen = new ArrayList<>(); + try { + Thread.currentThread().setContextClassLoader(callerClassLoader); + try (CloseableBatchIterator iterator = + ThreadPoolUtils.sequentialBatchedExecuteCloseable( + workers, + value -> { + seen.add(Thread.currentThread().getContextClassLoader()); + return Collections.singletonList(value); + }, + Arrays.asList(0, 1), + 1)) { + while (iterator.hasNext()) { + iterator.next(); + } + } + } finally { + Thread.currentThread().setContextClassLoader(original); + } + + assertThat(seen).containsExactly(callerClassLoader, callerClassLoader); + // The pool is shared, so a worker that keeps a caller's loader would hand it to whatever + // runs on that thread next. + AtomicReference afterwards = new AtomicReference<>(); + workers.submit(() -> afterwards.set(Thread.currentThread().getContextClassLoader())) + .get(10, TimeUnit.SECONDS); + assertThat(afterwards.get()).isSameAs(poolClassLoader.get()); + workers.shutdownNow(); + } + @Test public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exception { ThreadPoolExecutor workers = (ThreadPoolExecutor) Executors.newFixedThreadPool(1); diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java index 0ef818762d40..ffd7a8fe0c37 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java @@ -61,7 +61,7 @@ public static Iterable sequentialBatchedExecute( executor, processor, input, effectiveThreadNum(threadNum, executor)); } - /** This method parallel processes one bounded batch and waits for it when closed. */ + /** This method parallel processes a bounded number of inputs and waits for them when closed. */ public static ThreadPoolUtils.CloseableBatchIterator sequentialBatchedExecuteCloseable( Function> processor, List input, @Nullable Integer threadNum) { From 5f10ffe7c2a048913fa8220378a31a48dc15b3a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dapeng=20Sun=28=E5=AD=99=E5=A4=A7=E9=B9=8F=29?= Date: Wed, 26 Aug 2026 16:47:33 +0800 Subject: [PATCH 2/5] [core] Parallelize format table overwrite cleanup Add bounded cleanup concurrency for catalog-managed partitioned Format Tables while preserving serial behavior for other table paths. Deletes are handed out through the shared bounded runner, so the listing stays lazy, everything accepted is waited for before the commit fails, and failures keep their input order. --- docs/generated/core_configuration.html | 6 + .../java/org/apache/paimon/CoreOptions.java | 20 + .../table/format/FormatBatchWriteBuilder.java | 7 +- .../table/format/FormatTableCommit.java | 228 ++- .../org/apache/paimon/CoreOptionsTest.java | 37 + .../FormatTableCommitStatisticsTest.java | 3 +- .../table/format/FormatTableCommitTest.java | 1322 ++++++++++++++++- 7 files changed, 1567 insertions(+), 56 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 3fb25ebce15f..c2aaf6d6fe01 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -746,6 +746,12 @@ String Format table commit hive sync uri. + +

format-table.commit.cleanup-thread-num
+ 64 + Integer + The maximum number of concurrent old-data file deletions during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. +
format-table.file.compression
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index a19d1818c247..430b5e362902 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2654,6 +2654,16 @@ public String toString() { .noDefaultValue() .withDescription("Format table commit hive sync uri."); + public static final ConfigOption FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM = + key("format-table.commit.cleanup-thread-num") + .intType() + .defaultValue(64) + .withDescription( + "The maximum number of concurrent old-data file deletions during " + + "overwrite commits for an internal Format Table with " + + "catalog-managed partitions. Supported values are 1 through " + + "64. Other Format Tables use serial cleanup."); + @Immutable public static final ConfigOption BLOB_FIELD = key("blob-field") @@ -3302,6 +3312,16 @@ public String formatTableCommitSyncPartitionHiveUri() { return options.get(FORMAT_TABLE_COMMIT_HIVE_SYNC_URI); } + public int formatTableCommitCleanupThreadNum() { + int threadNum = options.get(FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM); + checkArgument( + threadNum >= 1 && threadNum <= 64, + "Option %s must be between 1 and 64, but was %s.", + FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), + threadNum); + return threadNum; + } + public MemorySize fileReaderAsyncThreshold() { return options.get(FILE_READER_ASYNC_THRESHOLD); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java index 73c11d774166..4c8beca9c175 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java @@ -78,6 +78,10 @@ public BatchTableCommit newCommit() { CoreOptions options = new CoreOptions(table.options()); boolean formatTablePartitionOnlyValueInPath = options.formatTablePartitionOnlyValueInPath(); String syncHiveUri = options.formatTableCommitSyncPartitionHiveUri(); + int cleanupThreadNum = + table.partitionManager() != null && !table.partitionKeys().isEmpty() + ? options.formatTableCommitCleanupThreadNum() + : 1; return new FormatTableCommit( table.location(), table.partitionKeys(), @@ -90,7 +94,8 @@ public BatchTableCommit newCommit() { syncHiveUri, table.catalogContext(), table.partitionManager(), - options.dynamicPartitionOverwrite()); + options.dynamicPartitionOverwrite(), + cleanupThreadNum); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index be2105930a57..70dbe38135f0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -38,6 +38,7 @@ import org.apache.paimon.table.sink.TableCommit; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.ThreadPoolUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,24 +47,36 @@ import java.io.FileNotFoundException; import java.io.IOException; +import java.io.UncheckedIOException; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.NoSuchElementException; import java.util.Set; +import java.util.concurrent.ExecutorService; import java.util.stream.Collectors; import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition; +import static org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator; +import static org.apache.paimon.utils.ThreadPoolUtils.sequentialBatchedExecuteNonCancellable; /** Commit for Format Table. */ public class FormatTableCommit implements BatchTableCommit { private static final Logger LOG = LoggerFactory.getLogger(FormatTableCommit.class); + private static final int MAX_CLEANUP_THREAD_NUM = 64; + + private static final ExecutorService CLEANUP_EXECUTOR = + ThreadPoolUtils.createCachedThreadPool( + MAX_CLEANUP_THREAD_NUM, "FORMAT-TABLE-COMMIT-CLEANUP-THREAD-POOL"); + private String location; private final boolean formatTablePartitionOnlyValueInPath; private final String defaultPartName; @@ -75,6 +88,7 @@ public class FormatTableCommit implements BatchTableCommit { private Identifier tableIdentifier; @Nullable private final FormatTablePartitionManager partitionManager; private final boolean dynamicPartitionOverwrite; + private final int cleanupThreadNum; public FormatTableCommit( String location, @@ -88,7 +102,14 @@ public FormatTableCommit( @Nullable String syncHiveUri, CatalogContext catalogContext, @Nullable FormatTablePartitionManager partitionManager, - boolean dynamicPartitionOverwrite) { + boolean dynamicPartitionOverwrite, + int cleanupThreadNum) { + if (cleanupThreadNum < 1 || cleanupThreadNum > MAX_CLEANUP_THREAD_NUM) { + throw new IllegalArgumentException( + String.format( + "Format Table cleanup thread number must be between 1 and %s, but was %s.", + MAX_CLEANUP_THREAD_NUM, cleanupThreadNum)); + } this.location = location; this.fileIO = fileIO; this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath; @@ -100,6 +121,7 @@ public FormatTableCommit( this.tableIdentifier = tableIdentifier; this.partitionManager = partitionManager; this.dynamicPartitionOverwrite = dynamicPartitionOverwrite; + this.cleanupThreadNum = cleanupThreadNum; if (syncHiveUri != null) { try { Options options = new Options(); @@ -151,32 +173,31 @@ public void commit(List commitMessages) { // A static partition may name only the leading keys, in which case the path // is a prefix and the partition directories of the remaining keys sit below. clearedPartitionPaths.addAll( - deletePreviousDataFile( - partitionPath, partitionKeys.size() - staticPartitions.size())); + deletePreviousDataFiles( + Collections.singletonList(partitionPath), + partitionKeys.size() - staticPartitions.size(), + cleanupThreadNum)); } if (!fileIO.exists(partitionPath)) { fileIO.mkdirs(partitionPath); } } else if (overwrite) { if (replacesOnlyWrittenPartitions()) { - Set partitionPaths = new HashSet<>(); + Set partitionPaths = new LinkedHashSet<>(); for (TwoPhaseCommitMessage message : messages) { partitionPaths.add(message.getCommitter().targetPath().getParent()); } - for (Path p : partitionPaths) { - // The parent of a written file is a complete partition directory - the - // table directory itself when the table is unpartitioned - so there is no - // partition level below it to descend, and it is a partition this commit - // writes anyway. - deletePreviousDataFile(p, 0); - } + // The parent of a written file is a complete partition directory - the table + // directory itself when the table is unpartitioned - so there is no partition + // level below it to descend. Collect every selected directory before deleting + // so many small partitions can share the same cleanup concurrency window. + deletePreviousDataFiles(new ArrayList<>(partitionPaths), 0, cleanupThreadNum); } else { // Overwriting without naming a partition replaces the table, so what has to go // is everything the table holds rather than the files this commit happens to // write: a statement whose query returns nothing still empties the table. - for (Path dataDirectory : tableDataDirectories()) { - clearedPartitionPaths.addAll(deletePreviousDataFile(dataDirectory, 0)); - } + clearedPartitionPaths.addAll( + deletePreviousDataFiles(tableDataDirectories(), 0, cleanupThreadNum)); } } @@ -244,9 +265,26 @@ public void commit(List commitMessages) { } } - } catch (Exception e) { - this.abort(commitMessages); - throw new RuntimeException(e); + } catch (Throwable failure) { + // Cleanup restores the caller's interrupt before failing. Clear it only while aborting + // staging output, then restore it; an abort failure is secondary to the commit failure + // that made abort necessary. + boolean interrupted = Thread.interrupted(); + try { + this.abort(commitMessages); + } catch (Throwable abortFailure) { + if (failure != abortFailure) { + failure.addSuppressed(abortFailure); + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + if (failure instanceof Error) { + throw (Error) failure; + } + throw new RuntimeException(failure); } } @@ -506,42 +544,142 @@ private List, Path>> partitionsInTheFileSyste */ private Set deletePreviousDataFile(Path partitionPath, int partitionLevels) throws IOException { + return deletePreviousDataFiles( + Collections.singletonList(partitionPath), partitionLevels, 1); + } + + private Set deletePreviousDataFiles( + List partitionPaths, int partitionLevels, int threadNum) throws IOException { + PreviousDataFiles dataFiles = new PreviousDataFiles(partitionPaths, partitionLevels); Set clearedPartitionPaths = new HashSet<>(); - if (fileIO.exists(partitionPath)) { - // Committed data files only: what sits under a staging directory is another writer's - // uncommitted output, whatever its name looks like. - for (FileStatus file : - FormatTableScan.listDataFiles( - fileIO, - partitionPath, - partitionLevels, - formatTablePartitionOnlyValueInPath, - defaultPartName)) { - boolean deleted; - try { - deleted = fileIO.delete(file.getPath(), false); - } catch (FileNotFoundException ignore) { - continue; - } catch (IOException e) { - throw new RuntimeException(e); + try { + if (threadNum == 1) { + while (dataFiles.hasNext()) { + FileStatus file = dataFiles.next(); + if (deleteDataFile(file)) { + clearedPartitionPaths.add(file.getPath().getParent()); + } } - if (deleted) { - // Only what this commit removed: a file another writer deleted first would - // have every concurrent writer report the whole subtree. - clearedPartitionPaths.add(file.getPath().getParent()); - } else if (fileIO.exists(file.getPath())) { - // A refusal is not that race: the file is still readable, and going on would - // report the partition as holding nothing while its rows are still there. - throw new IOException( - String.format( - "Failed to delete data file %s of table %s.", - file.getPath(), tableIdentifier.getFullName())); + return clearedPartitionPaths; + } + // Listing lazily keeps the memory of an overwrite that replaces the table + // proportional to one partition rather than to everything the table holds. Closing + // the iterator is what stops new deletes and waits for the ones already handed out, + // so a failure cannot leave a worker still deleting after this method returns. + try (CloseableBatchIterator cleared = + sequentialBatchedExecuteNonCancellable( + CLEANUP_EXECUTOR, this::deleteAndReportCleared, dataFiles, threadNum)) { + while (cleared.hasNext()) { + clearedPartitionPaths.add(cleared.next()); } } + } catch (UncheckedIOException e) { + throw (IOException) withoutUncheckedWrapper(e); } return clearedPartitionPaths; } + /** + * Unwraps the {@link UncheckedIOException} a worker has to throw to cross a {@link + * java.util.function.Function}, so a caller sees the {@link IOException} the file system raised + * rather than the wrapper. Whatever else failed while the accepted deletes drained is attached + * to the wrapper, so it has to travel with the cause rather than be dropped with it. + */ + private static Throwable withoutUncheckedWrapper(Throwable failure) { + if (!(failure instanceof UncheckedIOException)) { + return failure; + } + Throwable unwrapped = failure.getCause(); + for (Throwable suppressed : failure.getSuppressed()) { + unwrapped.addSuppressed(withoutUncheckedWrapper(suppressed)); + } + return unwrapped; + } + + /** Deletes one file and reports the partition it emptied, for a worker that cannot throw. */ + private List deleteAndReportCleared(FileStatus file) { + try { + return deleteDataFile(file) + ? Collections.singletonList(file.getPath().getParent()) + : Collections.emptyList(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + /** Deletes one listed data file and reports whether this commit removed it. */ + private boolean deleteDataFile(FileStatus file) throws IOException { + boolean deleted; + try { + deleted = fileIO.delete(file.getPath(), false); + } catch (FileNotFoundException ignore) { + return false; + } + if (deleted) { + return true; + } + if (fileIO.exists(file.getPath())) { + // A refusal is not a concurrent-delete race: the file is still readable, and going on + // would report the partition as holding nothing while its rows are still there. + throw new IOException( + String.format( + "Failed to delete data file %s of table %s.", + file.getPath(), tableIdentifier.getFullName())); + } + return false; + } + + /** The committed data files below the given paths, listed one partition at a time. */ + private final class PreviousDataFiles implements Iterator { + + private final List partitionPaths; + private final int partitionLevels; + private List currentFiles = Collections.emptyList(); + private int nextPartition; + private int nextFile; + + private PreviousDataFiles(List partitionPaths, int partitionLevels) { + this.partitionPaths = partitionPaths; + this.partitionLevels = partitionLevels; + } + + @Override + public boolean hasNext() { + while (nextFile >= currentFiles.size()) { + if (nextPartition >= partitionPaths.size()) { + return false; + } + Path partitionPath = partitionPaths.get(nextPartition++); + try { + if (!fileIO.exists(partitionPath)) { + continue; + } + // Committed data files only: what sits under a staging directory is another + // writer's uncommitted output, whatever its name looks like. + currentFiles = + FormatTableScan.listDataFiles( + fileIO, + partitionPath, + partitionLevels, + formatTablePartitionOnlyValueInPath, + defaultPartName); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + nextFile = 0; + } + return true; + } + + @Override + public FileStatus next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return currentFiles.get(nextFile++); + } + } + @Override public void truncateTable() { // Data files only. The partition directories stay, and so do their catalog registrations: diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index ac71ff2552aa..f3e81179d733 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -288,4 +288,41 @@ public void testLocalKvDbBlockSize() { .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("local-kv-db.block-size"); } + + @Test + public void testFormatTableCommitCleanupThreadNumDefaultsTo64AndAcceptsBounds() { + Options conf = new Options(); + assertThat(new CoreOptions(conf).formatTableCommitCleanupThreadNum()).isEqualTo(64); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, 1); + assertThat(new CoreOptions(conf).formatTableCommitCleanupThreadNum()).isEqualTo(1); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, 64); + assertThat(new CoreOptions(conf).formatTableCommitCleanupThreadNum()).isEqualTo(64); + } + + @Test + public void testFormatTableCommitCleanupThreadNumRejectsValuesOutsideSupportedRange() { + Options conf = new Options(); + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, 0); + assertThatThrownBy(() -> new CoreOptions(conf).formatTableCommitCleanupThreadNum()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("format-table.commit.cleanup-thread-num") + .hasMessageContaining("1") + .hasMessageContaining("64"); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, -1); + assertThatThrownBy(() -> new CoreOptions(conf).formatTableCommitCleanupThreadNum()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("format-table.commit.cleanup-thread-num") + .hasMessageContaining("1") + .hasMessageContaining("64"); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, 65); + assertThatThrownBy(() -> new CoreOptions(conf).formatTableCommitCleanupThreadNum()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("format-table.commit.cleanup-thread-num") + .hasMessageContaining("1") + .hasMessageContaining("64"); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java index f5ac14607065..1a52eaab887d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java @@ -674,7 +674,8 @@ private FormatTableCommit commit( null, null, partitionManager, - dynamicPartitionOverwrite); + dynamicPartitionOverwrite, + /* cleanupThreadNum */ 1); } /** An overwrite that names no partition: INSERT OVERWRITE without a PARTITION clause. */ diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java index 75b141d947f2..97e45a16afe1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java @@ -18,25 +18,47 @@ package org.apache.paimon.table.format; +import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.RenamingTwoPhaseOutputStream; import org.apache.paimon.fs.TwoPhaseOutputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionStatistics; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.PartitionPathUtils; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.mockito.ArgumentCaptor; +import java.io.FileNotFoundException; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME; import static org.assertj.core.api.Assertions.assertThat; @@ -46,6 +68,7 @@ import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -86,7 +109,8 @@ void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception null, null, partitionManager, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); CommitMessage message = new TwoPhaseCommitMessage(committer); assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) @@ -119,7 +143,8 @@ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception { null, null, partitionManager, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); CommitMessage message = new TwoPhaseCommitMessage(committer); assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) @@ -215,7 +240,8 @@ void testOverwriteKeepsFilesOfConcurrentWritersStagingTrees() throws Exception { null, null, null, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); @@ -264,7 +290,8 @@ void testOverwritingAPrefixKeepsStagingTreesSittingAtAPartitionLevel() throws Ex null, null, null, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); commit.commit(Collections.emptyList()); @@ -308,7 +335,8 @@ void testOverwritingAPrefixClearsTheDefaultPartitionDirectory() throws Exception null, null, null, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); commit.commit(Collections.emptyList()); @@ -356,7 +384,8 @@ void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception { null, null, null, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); assertThatThrownBy(() -> commit.commit(Collections.emptyList())) .isInstanceOf(RuntimeException.class) @@ -632,6 +661,669 @@ void testOverwritingTheWholeTableLeavesADirectoryThatIsNoPartitionOfIt() throws assertThat(fileIO.exists(new Path(tablePath, "loose.csv"))).isTrue(); } + @Test + void testCatalogManagedBuilderUses64WayCleanupByDefault() throws Exception { + ParallelDeleteFileIO fileIO = new ParallelDeleteFileIO(64, true); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 65); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + doAnswer( + invocation -> { + assertThat(fileIO.activeDeletes()).isZero(); + return null; + }) + .when(committer) + .commit(fileIO); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, + fileIO, + partitionManager, + Collections.emptyMap(), + Collections.singletonMap("part", "p")); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = + executor.submit( + () -> + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer)))); + + assertThat(fileIO.awaitFirstWave()).isTrue(); + assertThat(fileIO.deleteCalls()).isEqualTo(64); + assertThat(fileIO.awaitUnexpectedExtraDelete()).isFalse(); + verify(committer, never()).commit(fileIO); + + fileIO.releaseFirstWave(); + result.get(10, TimeUnit.SECONDS); + assertThat(fileIO.deleteCalls()).isEqualTo(65); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(64); + verify(committer).commit(fileIO); + } finally { + fileIO.releaseFirstWave(); + executor.shutdownNow(); + } + } + + @Test + void testCatalogManagedBuilderHonorsConfiguredSerialCleanup() throws Exception { + SerialProbeFileIO fileIO = new SerialProbeFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 3); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, + fileIO, + partitionManager, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "1"), + Collections.singletonMap("part", "p")); + + commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); + + assertThat(fileIO.deleteCalls()).isEqualTo(3); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(1); + } + + @Test + void testCatalogManagedBuilderPropagatesConfiguredCleanupConcurrency() throws Exception { + ParallelDeleteFileIO fileIO = new ParallelDeleteFileIO(7, true); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 8); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, + fileIO, + partitionManager, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "7"), + Collections.singletonMap("part", "p")); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> commit.commit(Collections.emptyList())); + + assertThat(fileIO.awaitFirstWave()).isTrue(); + assertThat(fileIO.deleteCalls()).isEqualTo(7); + assertThat(fileIO.awaitUnexpectedExtraDelete()).isFalse(); + + fileIO.releaseFirstWave(); + result.get(10, TimeUnit.SECONDS); + assertThat(fileIO.deleteCalls()).isEqualTo(8); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(7); + } finally { + fileIO.releaseFirstWave(); + executor.shutdownNow(); + } + } + + @Test + void testFilesystemDiscoveredFormatTableCleanupRemainsSerial() throws Exception { + SerialProbeFileIO fileIO = new SerialProbeFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 3); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, + fileIO, + null, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "64"), + Collections.singletonMap("part", "p")); + + commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); + + assertThat(fileIO.deleteCalls()).isEqualTo(3); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(1); + } + + @Test + void testCleanupIsAHardBarrierBeforePublishingNewFiles() throws Exception { + PartialBarrierDeleteFileIO fileIO = new PartialBarrierDeleteFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 2); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + doAnswer( + invocation -> { + assertThat(fileIO.activeDeletes()).isZero(); + return null; + }) + .when(committer) + .commit(fileIO); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = + executor.submit( + () -> + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer)))); + + assertThat(fileIO.awaitBothDeletesStarted()).isTrue(); + verify(committer, never()).commit(fileIO); + + fileIO.releaseFirstDelete(); + assertThat(fileIO.awaitFirstDeleteReturned()).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + verify(committer, never()).commit(fileIO); + + fileIO.releaseSecondDelete(); + result.get(10, TimeUnit.SECONDS); + verify(committer).commit(fileIO); + } finally { + fileIO.releaseFirstDelete(); + fileIO.releaseSecondDelete(); + executor.shutdownNow(); + } + } + + @Test + void testUnpartitionedCatalogManagedFormatTableCleanupRemainsSerial() throws Exception { + SerialProbeFileIO fileIO = new SerialProbeFileIO(); + Path tablePath = new Path(tempDir.toUri()); + writeOldFiles(fileIO, tablePath, 3); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(tablePath, "data-new.csv")); + FormatTableCommit commit = + builderUnpartitionedOverwriteCommit( + tablePath, + fileIO, + partitionManager, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "64")); + + commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); + + assertThat(fileIO.deleteCalls()).isEqualTo(3); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(1); + } + + @Test + void testCleanupFailureStopsNewSubmissionsAndDrainsTheAlreadyRunningDelete() throws Exception { + FailureDrainFileIO fileIO = new FailureDrainFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 6); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = + executor.submit( + () -> + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer)))); + + assertThat(fileIO.awaitFailureAttempted()).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + assertThat(fileIO.attemptedFiles()) + .containsExactlyInAnyOrder("data-000.csv", "data-001.csv"); + + fileIO.releaseSuccessfulSibling(); + assertThat(rootCause(awaitFailure(result))) + .hasMessage("delete failed at input position 0"); + assertThat(fileIO.attemptedFiles()) + .containsExactlyInAnyOrder("data-000.csv", "data-001.csv"); + assertThat(fileIO.successfulFiles()).containsExactly("data-001.csv"); + verify(committer, never()).commit(fileIO); + } finally { + fileIO.releaseSuccessfulSibling(); + executor.shutdownNow(); + } + } + + @Test + void testLaterPartitionListingFailureDrainsAcceptedDeletesBeforeAbort() throws Exception { + Path tablePath = new Path(new Path(tempDir.toUri()), "listing-failure"); + Path firstPartition = new Path(tablePath, "part=p0"); + Path failingPartition = new Path(tablePath, "part=p1"); + LaterRootListingFailureFileIO fileIO = + new LaterRootListingFailureFileIO(failingPartition, 2); + writeOldFiles(fileIO, firstPartition, 2); + writeOldFiles(fileIO, failingPartition, 1); + + AtomicInteger discardCalls = new AtomicInteger(); + ConcurrentLinkedQueue activeDeletesAtDiscard = new ConcurrentLinkedQueue<>(); + List messages = new ArrayList<>(); + List committers = new ArrayList<>(); + for (Path partition : Arrays.asList(firstPartition, failingPartition)) { + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partition, "data-new.csv")); + doAnswer( + invocation -> { + activeDeletesAtDiscard.add(fileIO.activeDeletes()); + discardCalls.incrementAndGet(); + return null; + }) + .when(committer) + .discard(fileIO); + committers.add(committer); + messages.add(new TwoPhaseCommitMessage(committer)); + } + FormatTableCommit commit = newCleanupCommit(tablePath, fileIO, null, null, 3); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit(() -> commit.commit(messages)); + try { + assertThat(fileIO.awaitListingFailure()).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + assertThat(discardCalls).hasValue(0); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer, never()).commit(fileIO); + verify(committer, never()).discard(fileIO); + } + + fileIO.releaseFirstDelete(); + assertThat(fileIO.awaitFirstDeleteReturned()).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + assertThat(discardCalls).hasValue(0); + + fileIO.releaseSecondDelete(); + assertThat(rootCause(awaitFailure(result))) + .hasMessage("Failed to list the later partition root."); + assertThat(discardCalls).hasValue(2); + assertThat(activeDeletesAtDiscard).containsExactly(0, 0); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer, never()).commit(fileIO); + verify(committer).discard(fileIO); + } + } finally { + fileIO.releaseFirstDelete(); + fileIO.releaseSecondDelete(); + try { + if (!result.isDone()) { + try { + result.get(10, TimeUnit.SECONDS); + } catch (ExecutionException ignored) { + // The test expects the listing failure above. + } + } + } finally { + executor.shutdownNow(); + } + } + } + + @Test + void testCleanupSelectsLowestInputFailureAndSuppressesTheOtherFailure() throws Exception { + OrderedDualFailureFileIO fileIO = new OrderedDualFailureFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 2); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> commit.commit(Collections.emptyList())); + assertThat(fileIO.awaitHigherPositionFailure()).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + + fileIO.releaseLowerPositionFailure(); + Throwable primary = rootCause(awaitFailure(result)); + assertThat(primary).hasMessage("delete failed at input position 0"); + assertThat(primary.getSuppressed()) + .extracting(Throwable::getMessage) + .containsExactly("delete failed at input position 1"); + } finally { + fileIO.releaseLowerPositionFailure(); + executor.shutdownNow(); + } + } + + @Test + void testInterruptDrainsCleanupRestoresFlagAndNeverPublishes() throws Exception { + BlockingDeleteFileIO fileIO = new BlockingDeleteFileIO(2); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 2); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + AtomicReference failure = new AtomicReference<>(); + AtomicBoolean interruptRestored = new AtomicBoolean(); + CountDownLatch commitReturned = new CountDownLatch(1); + Thread commitThread = + new Thread( + () -> { + try { + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer))); + } catch (Throwable t) { + failure.set(t); + } finally { + interruptRestored.set(Thread.currentThread().isInterrupted()); + commitReturned.countDown(); + } + }, + "format-cleanup-interrupted-caller"); + + commitThread.start(); + try { + assertThat(fileIO.awaitDeletesStarted()).isTrue(); + commitThread.interrupt(); + assertThat(commitReturned.await(300, TimeUnit.MILLISECONDS)).isFalse(); + + fileIO.releaseDeletes(); + assertThat(commitReturned.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(failure.get()).isNotNull(); + assertThat(causeChain(failure.get())).anyMatch(InterruptedException.class::isInstance); + assertThat(interruptRestored).isTrue(); + verify(committer, never()).commit(fileIO); + } finally { + fileIO.releaseDeletes(); + commitThread.interrupt(); + commitThread.join(TimeUnit.SECONDS.toMillis(10)); + } + assertThat(commitThread.isAlive()).isFalse(); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + void testCleanupStatisticsClaimOnlyFilesDeletedByThisCommit() throws Exception { + MixedOwnershipFileIO fileIO = new MixedOwnershipFileIO(); + Path tablePath = new Path(tempDir.toUri()); + writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=00"), 1); + writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=01"), 1); + writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=02"), 1); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + PARTITION_DEFAULT_NAME.defaultValue(), + true, + Identifier.create("cleanup_db", "cleanup_table"), + Collections.singletonMap("year", "2025"), + null, + null, + partitionManager, + /* dynamicPartitionOverwrite */ true, + 2); + + commit.commit(Collections.emptyList()); + + Map owned = partitionSpec("2025", "00"); + ArgumentCaptor>> specs = + ArgumentCaptor.forClass((Class) List.class); + ArgumentCaptor> statistics = + ArgumentCaptor.forClass((Class) List.class); + verify(partitionManager) + .createPartitions(specs.capture(), eq(true), statistics.capture(), eq(true)); + assertThat(specs.getValue()).containsExactly(owned); + assertThat(statistics.getValue()) + .singleElement() + .satisfies( + stat -> { + assertThat(stat.spec()).isEqualTo(owned); + assertThat(stat.recordCount()).isZero(); + assertThat(stat.fileSizeInBytes()).isZero(); + assertThat(stat.fileCount()).isZero(); + }); + assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=00/data-000.csv"))).isFalse(); + assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=01/data-000.csv"))).isFalse(); + assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=02/data-000.csv"))).isFalse(); + } + + @Test + void testCleanupRejectsFalseWhenTheOldDataFileStillExists() throws Exception { + RefusingDeleteFileIO fileIO = new RefusingDeleteFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 1); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + + assertThatThrownBy( + () -> + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer)))) + .hasRootCauseMessage( + "Failed to delete data file " + + new Path(partitionPath, "data-000.csv") + + " of table cleanup_db.cleanup_table."); + verify(committer, never()).commit(fileIO); + verify(committer).discard(fileIO); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + @Test + void testConcurrentCleanupReportsCompleteStatisticsAfterBarrier() throws Exception { + ParallelDeleteFileIO fileIO = new ParallelDeleteFileIO(4); + Path tablePath = new Path(tempDir.toUri()); + for (int month = 0; month < 8; month++) { + writeOldFiles( + fileIO, new Path(tablePath, String.format("year=2025/month=%02d", month)), 1); + } + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + doAnswer( + invocation -> { + assertThat(fileIO.activeDeletes()).isZero(); + return null; + }) + .when(partitionManager) + .createPartitions(anyList(), eq(true), anyList(), eq(true)); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + PARTITION_DEFAULT_NAME.defaultValue(), + true, + Identifier.create("cleanup_db", "cleanup_table"), + Collections.singletonMap("year", "2025"), + null, + null, + partitionManager, + /* dynamicPartitionOverwrite */ true, + 4); + + commit.commit(Collections.emptyList()); + + List> expectedSpecs = new ArrayList<>(); + for (int month = 0; month < 8; month++) { + expectedSpecs.add(partitionSpec("2025", String.format("%02d", month))); + } + ArgumentCaptor>> specs = + ArgumentCaptor.forClass((Class) List.class); + ArgumentCaptor> statistics = + ArgumentCaptor.forClass((Class) List.class); + verify(partitionManager) + .createPartitions(specs.capture(), eq(true), statistics.capture(), eq(true)); + assertThat(specs.getValue()).containsExactlyInAnyOrderElementsOf(expectedSpecs); + assertThat(statistics.getValue()) + .hasSize(8) + .extracting(PartitionStatistics::spec) + .containsExactlyInAnyOrderElementsOf(expectedSpecs); + assertThat(statistics.getValue()) + .allSatisfy( + stat -> { + assertThat(stat.recordCount()).isZero(); + assertThat(stat.fileSizeInBytes()).isZero(); + assertThat(stat.fileCount()).isZero(); + }); + } + + @Test + void testCatalogManagedOverwriteCleanupSpansPartitionDirectories() throws Exception { + assertOverwriteCleanupSpansPartitions(/* dynamicPartitionOverwrite */ true); + assertOverwriteCleanupSpansPartitions(/* dynamicPartitionOverwrite */ false); + } + + @Test + void testCleanupDoesNotListEveryPartitionBeforeTheFirstDeleteWindowCompletes() + throws Exception { + Path tablePath = new Path(new Path(tempDir.toUri()), "lazy-root-listing"); + Path firstPartition = new Path(tablePath, "part=p0"); + Path deferredPartition = new Path(tablePath, "part=p1"); + LazyRootListingFileIO fileIO = new LazyRootListingFileIO(2, deferredPartition); + writeOldFiles(fileIO, firstPartition, 2); + writeOldFiles(fileIO, deferredPartition, 2); + + List messages = new ArrayList<>(); + for (Path partition : Arrays.asList(firstPartition, deferredPartition)) { + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partition, "data-new.csv")); + messages.add(new TwoPhaseCommitMessage(committer)); + } + Map options = new LinkedHashMap<>(); + options.put(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "2"); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, + fileIO, + mock(FormatTablePartitionManager.class), + options, + /* staticPartition */ null); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = null; + try { + result = executor.submit(() -> commit.commit(messages)); + assertThat(fileIO.awaitFirstWave()).isTrue(); + assertThat(fileIO.deferredRootListed()).isFalse(); + } finally { + fileIO.releaseFirstWave(); + if (result != null) { + result.get(10, TimeUnit.SECONDS); + } + executor.shutdownNow(); + } + + assertThat(fileIO.deferredRootListed()).isTrue(); + assertThat(fileIO.deleteCalls()).isEqualTo(4); + } + + @Test + void testBuilderCleanupConcurrencyDoesNotApplyToTruncateOperations() throws Exception { + SerialProbeFileIO tableFileIO = new SerialProbeFileIO(); + Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-table"); + Path tablePartitionPath = new Path(tablePath, "part=p"); + writeOldFiles(tableFileIO, tablePartitionPath, 3); + FormatTablePartitionManager tableManager = mock(FormatTablePartitionManager.class); + when(tableManager.listPartitions(Collections.emptyMap(), null)) + .thenReturn( + Collections.singletonList( + new Partition( + Collections.singletonMap("part", "p"), + 0, + 0, + 0, + 0, + -1, + false))); + builderTruncateCommit(tablePath, tableFileIO, tableManager).truncateTable(); + + SerialProbeFileIO partitionFileIO = new SerialProbeFileIO(); + Path partitionsPath = new Path(new Path(tempDir.toUri()), "truncate-partitions"); + Path namedPartitionPath = new Path(partitionsPath, "part=p"); + writeOldFiles(partitionFileIO, namedPartitionPath, 3); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + when(partitionManager.listPartitionsByNames(anyList())) + .thenReturn( + Collections.singletonList( + new Partition( + Collections.singletonMap("part", "p"), + 0, + 0, + 0, + 0, + -1, + false))); + builderTruncateCommit(partitionsPath, partitionFileIO, partitionManager) + .truncatePartitions( + Collections.singletonList(Collections.singletonMap("part", "p"))); + + assertThat(tableFileIO.maxConcurrentDeletes()).isEqualTo(1); + assertThat(partitionFileIO.maxConcurrentDeletes()).isEqualTo(1); + } + + @Test + void testAbortFailureDoesNotReplaceInterruptedCleanupFailure() throws Exception { + BlockingDeleteFileIO fileIO = new BlockingDeleteFileIO(2); + Path tablePath = new Path(new Path(tempDir.toUri()), "abort-failure"); + Path partitionPath = new Path(tablePath, "part=p"); + writeOldFiles(fileIO, partitionPath, 2); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + doThrow(new IOException("discard failed after cleanup interruption")) + .when(committer) + .discard(fileIO); + FormatTableCommit commit = + newCleanupCommit(tablePath, fileIO, null, Collections.singletonMap("part", "p"), 2); + AtomicReference failure = new AtomicReference<>(); + AtomicBoolean interruptRestored = new AtomicBoolean(); + Thread commitThread = + new Thread( + () -> { + try { + commit.commit( + Collections.singletonList( + new TwoPhaseCommitMessage(committer))); + } catch (Throwable t) { + failure.set(t); + } finally { + interruptRestored.set(Thread.currentThread().isInterrupted()); + } + }, + "format-cleanup-abort-failure-caller"); + + commitThread.start(); + try { + assertThat(fileIO.awaitDeletesStarted()).isTrue(); + commitThread.interrupt(); + fileIO.releaseDeletes(); + commitThread.join(TimeUnit.SECONDS.toMillis(10)); + } finally { + fileIO.releaseDeletes(); + commitThread.interrupt(); + commitThread.join(TimeUnit.SECONDS.toMillis(10)); + } + + assertThat(commitThread.isAlive()).isFalse(); + assertThat(failure.get()).isNotNull(); + assertThat(rootCause(failure.get())).isInstanceOf(InterruptedException.class); + assertThat(failureTree(failure.get())) + .extracting(Throwable::getMessage) + .contains("discard failed after cleanup interruption"); + assertThat(interruptRestored).isTrue(); + verify(committer).discard(fileIO); + verify(committer, never()).commit(fileIO); + } + /** * An overwrite that names no partition: what INSERT OVERWRITE without a PARTITION clause is. */ @@ -652,7 +1344,617 @@ private FormatTableCommit overwritingCommit( null, null, null, - dynamicPartitionOverwrite); + dynamicPartitionOverwrite, + /* cleanupThreadNum */ 1); + } + + private void assertOverwriteCleanupSpansPartitions(boolean dynamicPartitionOverwrite) + throws Exception { + ParallelDeleteFileIO fileIO = new ParallelDeleteFileIO(4); + Path tablePath = + new Path( + new Path(tempDir.toUri()), + dynamicPartitionOverwrite ? "dynamic-roots" : "whole-roots"); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + List partitions = new ArrayList<>(); + List messages = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + String value = "p" + i; + Path partitionPath = new Path(tablePath, "part=" + value); + writeOldFiles(fileIO, partitionPath, 1); + partitions.add( + new Partition(Collections.singletonMap("part", value), 0, 0, 0, 0, -1, false)); + if (dynamicPartitionOverwrite) { + TwoPhaseOutputStream.Committer committer = + mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-new.csv")); + messages.add(new TwoPhaseCommitMessage(committer)); + } + } + if (!dynamicPartitionOverwrite) { + when(partitionManager.listPartitions(Collections.emptyMap(), null)) + .thenReturn(partitions); + } + Map options = new LinkedHashMap<>(); + options.put(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(), "4"); + options.put( + CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(), + Boolean.toString(dynamicPartitionOverwrite)); + FormatTableCommit commit = + builderOverwriteCommit( + tablePath, fileIO, partitionManager, options, /* staticPartition */ null); + + commit.commit(messages); + + assertThat(fileIO.deleteCalls()).isEqualTo(4); + assertThat(fileIO.maxConcurrentDeletes()).isEqualTo(4); + } + + private FormatTableCommit builderOverwriteCommit( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + Map options, + Map staticPartition) { + FormatTable table = formatTable(tablePath, fileIO, partitionManager, options); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + writeBuilder.withOverwrite(staticPartition); + return (FormatTableCommit) writeBuilder.newCommit(); + } + + private FormatTableCommit builderTruncateCommit( + Path tablePath, FileIO fileIO, FormatTablePartitionManager partitionManager) { + return (FormatTableCommit) + formatTable(tablePath, fileIO, partitionManager, Collections.emptyMap()) + .newBatchWriteBuilder() + .newCommit(); + } + + private FormatTable formatTable( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + Map options) { + RowType rowType = + RowType.builder() + .field("part", DataTypes.STRING()) + .field("id", DataTypes.INT()) + .build(); + FormatTable table = + FormatTable.builder() + .fileIO(fileIO) + .identifier(Identifier.create("cleanup_db", "cleanup_table")) + .rowType(rowType) + .partitionKeys(Collections.singletonList("part")) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options(options) + .partitionManager(partitionManager) + .build(); + return table; + } + + private FormatTableCommit builderUnpartitionedOverwriteCommit( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + Map options) { + FormatTable table = + FormatTable.builder() + .fileIO(fileIO) + .identifier(Identifier.create("cleanup_db", "cleanup_table")) + .rowType(RowType.builder().field("id", DataTypes.INT()).build()) + .partitionKeys(Collections.emptyList()) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options(options) + .partitionManager(partitionManager) + .build(); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + writeBuilder.withOverwrite(null); + return (FormatTableCommit) writeBuilder.newCommit(); + } + + private FormatTableCommit newCleanupCommit( + Path tablePath, + FileIO fileIO, + FormatTablePartitionManager partitionManager, + Map staticPartition, + int cleanupThreadNum) { + return new FormatTableCommit( + tablePath.toString(), + Collections.singletonList("part"), + fileIO, + false, + PARTITION_DEFAULT_NAME.defaultValue(), + true, + Identifier.create("cleanup_db", "cleanup_table"), + staticPartition, + null, + null, + partitionManager, + /* dynamicPartitionOverwrite */ true, + cleanupThreadNum); + } + + private static void writeOldFiles(LocalFileIO fileIO, Path partitionPath, int count) + throws IOException { + for (int i = 0; i < count; i++) { + fileIO.writeFile( + new Path(partitionPath, String.format("data-%03d.csv", i)), "old", false); + } + } + + private static ExecutionException awaitFailure(Future future) throws Exception { + try { + future.get(10, TimeUnit.SECONDS); + throw new AssertionError("Expected cleanup commit to fail"); + } catch (ExecutionException expected) { + return expected; + } + } + + private static Throwable rootCause(Throwable throwable) { + Throwable root = throwable; + while (root.getCause() != null) { + root = root.getCause(); + } + return root; + } + + private static List causeChain(Throwable throwable) { + List chain = new ArrayList<>(); + Throwable current = throwable; + while (current != null) { + chain.add(current); + current = current.getCause(); + } + return chain; + } + + private static List failureTree(Throwable throwable) { + List failures = new ArrayList<>(); + collectFailures(throwable, failures); + return failures; + } + + private static void collectFailures(Throwable throwable, List failures) { + if (throwable == null) { + return; + } + failures.add(throwable); + for (Throwable suppressed : throwable.getSuppressed()) { + collectFailures(suppressed, failures); + } + collectFailures(throwable.getCause(), failures); + } + + private static class ParallelDeleteFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + private final int firstWaveSize; + private final boolean holdFirstWave; + private final CountDownLatch firstWave; + private final CountDownLatch releaseFirstWave = new CountDownLatch(1); + private final CountDownLatch unexpectedExtraDelete = new CountDownLatch(1); + private final AtomicInteger deleteCalls = new AtomicInteger(); + private final AtomicInteger activeDeletes = new AtomicInteger(); + private final AtomicInteger maxConcurrentDeletes = new AtomicInteger(); + + private ParallelDeleteFileIO(int firstWaveSize) { + this(firstWaveSize, false); + } + + private ParallelDeleteFileIO(int firstWaveSize, boolean holdFirstWave) { + this.firstWaveSize = firstWaveSize; + this.holdFirstWave = holdFirstWave; + this.firstWave = new CountDownLatch(firstWaveSize); + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + int call = deleteCalls.incrementAndGet(); + int active = activeDeletes.incrementAndGet(); + maxConcurrentDeletes.updateAndGet(previous -> Math.max(previous, active)); + try { + if (call <= firstWaveSize) { + firstWave.countDown(); + if (!firstWave.await(10, TimeUnit.SECONDS)) { + throw new IOException("Expected cleanup delete calls did not overlap"); + } + if (holdFirstWave && !releaseFirstWave.await(10, TimeUnit.SECONDS)) { + throw new IOException("Test did not release the first cleanup wave"); + } + } else { + unexpectedExtraDelete.countDown(); + } + return super.delete(path, recursive); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while observing cleanup concurrency", e); + } finally { + activeDeletes.decrementAndGet(); + } + } + + protected int deleteCalls() { + return deleteCalls.get(); + } + + private int maxConcurrentDeletes() { + return maxConcurrentDeletes.get(); + } + + private int activeDeletes() { + return activeDeletes.get(); + } + + protected boolean awaitFirstWave() throws InterruptedException { + return firstWave.await(10, TimeUnit.SECONDS); + } + + private boolean awaitFirstWave(long timeout, TimeUnit unit) throws InterruptedException { + return firstWave.await(timeout, unit); + } + + private boolean awaitUnexpectedExtraDelete() throws InterruptedException { + return unexpectedExtraDelete.await(300, TimeUnit.MILLISECONDS); + } + + protected void releaseFirstWave() { + releaseFirstWave.countDown(); + } + } + + private static class LazyRootListingFileIO extends ParallelDeleteFileIO { + + private static final long serialVersionUID = 1L; + + private final Path deferredRoot; + private final AtomicBoolean deferredRootListed = new AtomicBoolean(); + + private LazyRootListingFileIO(int firstWaveSize, Path deferredRoot) { + super(firstWaveSize, true); + this.deferredRoot = deferredRoot; + } + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + if (deferredRoot.equals(path)) { + deferredRootListed.set(true); + } + return super.listStatus(path); + } + + private boolean deferredRootListed() { + return deferredRootListed.get(); + } + } + + private static class SerialProbeFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + private final CountDownLatch secondDeleteStarted = new CountDownLatch(1); + private final AtomicInteger deleteCalls = new AtomicInteger(); + private final AtomicInteger activeDeletes = new AtomicInteger(); + private final AtomicInteger maxConcurrentDeletes = new AtomicInteger(); + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + int call = deleteCalls.incrementAndGet(); + int active = activeDeletes.incrementAndGet(); + maxConcurrentDeletes.updateAndGet(previous -> Math.max(previous, active)); + try { + if (call == 1) { + secondDeleteStarted.await(300, TimeUnit.MILLISECONDS); + } else { + secondDeleteStarted.countDown(); + } + return super.delete(path, recursive); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while probing serial cleanup", e); + } finally { + activeDeletes.decrementAndGet(); + } + } + + private int deleteCalls() { + return deleteCalls.get(); + } + + private int maxConcurrentDeletes() { + return maxConcurrentDeletes.get(); + } + } + + private static class PartialBarrierDeleteFileIO extends SortedLocalFileIO { + + private static final long serialVersionUID = 1L; + + private final CountDownLatch bothDeletesStarted = new CountDownLatch(2); + private final CountDownLatch releaseFirstDelete = new CountDownLatch(1); + private final CountDownLatch releaseSecondDelete = new CountDownLatch(1); + private final CountDownLatch firstDeleteReturned = new CountDownLatch(1); + private final AtomicInteger activeDeletes = new AtomicInteger(); + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + activeDeletes.incrementAndGet(); + bothDeletesStarted.countDown(); + await(bothDeletesStarted, "both barrier delete calls"); + try { + if ("data-000.csv".equals(path.getName())) { + await(releaseFirstDelete, "first barrier delete release"); + return super.delete(path, recursive); + } + await(releaseSecondDelete, "second barrier delete release"); + return super.delete(path, recursive); + } finally { + activeDeletes.decrementAndGet(); + if ("data-000.csv".equals(path.getName())) { + firstDeleteReturned.countDown(); + } + } + } + + private boolean awaitBothDeletesStarted() throws InterruptedException { + return bothDeletesStarted.await(10, TimeUnit.SECONDS); + } + + private void releaseFirstDelete() { + releaseFirstDelete.countDown(); + } + + private void releaseSecondDelete() { + releaseSecondDelete.countDown(); + } + + private boolean awaitFirstDeleteReturned() throws InterruptedException { + return firstDeleteReturned.await(10, TimeUnit.SECONDS); + } + + private int activeDeletes() { + return activeDeletes.get(); + } + } + + private static class BlockingDeleteFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + private final CountDownLatch deletesStarted; + private final CountDownLatch releaseDeletes = new CountDownLatch(1); + private final AtomicInteger activeDeletes = new AtomicInteger(); + + private BlockingDeleteFileIO(int deleteCount) { + this.deletesStarted = new CountDownLatch(deleteCount); + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + activeDeletes.incrementAndGet(); + deletesStarted.countDown(); + try { + if (!releaseDeletes.await(10, TimeUnit.SECONDS)) { + throw new IOException("Test did not release blocked cleanup deletes"); + } + return super.delete(path, recursive); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while blocking cleanup delete", e); + } finally { + activeDeletes.decrementAndGet(); + } + } + + private boolean awaitDeletesStarted() throws InterruptedException { + return deletesStarted.await(10, TimeUnit.SECONDS); + } + + private void releaseDeletes() { + releaseDeletes.countDown(); + } + + private int activeDeletes() { + return activeDeletes.get(); + } + } + + private abstract static class SortedLocalFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + FileStatus[] statuses = super.listStatus(path); + Arrays.sort(statuses, Comparator.comparing(status -> status.getPath().toString())); + return statuses; + } + + protected static void await(CountDownLatch latch, String description) throws IOException { + try { + if (!latch.await(10, TimeUnit.SECONDS)) { + throw new IOException("Timed out waiting for " + description); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for " + description, e); + } + } + } + + private static class LaterRootListingFailureFileIO extends SortedLocalFileIO { + + private static final long serialVersionUID = 1L; + + private final Path failingRoot; + private final CountDownLatch deletesStarted; + private final CountDownLatch listingFailure = new CountDownLatch(1); + private final CountDownLatch releaseFirstDelete = new CountDownLatch(1); + private final CountDownLatch releaseSecondDelete = new CountDownLatch(1); + private final CountDownLatch firstDeleteReturned = new CountDownLatch(1); + private final AtomicInteger activeDeletes = new AtomicInteger(); + + private LaterRootListingFailureFileIO(Path failingRoot, int deleteCount) { + this.failingRoot = failingRoot; + this.deletesStarted = new CountDownLatch(deleteCount); + } + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + if (failingRoot.equals(path)) { + await(deletesStarted, "accepted deletes before later-root listing failure"); + listingFailure.countDown(); + throw new IOException("Failed to list the later partition root."); + } + return super.listStatus(path); + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + activeDeletes.incrementAndGet(); + deletesStarted.countDown(); + try { + if ("data-000.csv".equals(path.getName())) { + await(releaseFirstDelete, "release of the first accepted delete"); + } else { + await(releaseSecondDelete, "release of the second accepted delete"); + } + return super.delete(path, recursive); + } finally { + activeDeletes.decrementAndGet(); + if ("data-000.csv".equals(path.getName())) { + firstDeleteReturned.countDown(); + } + } + } + + private boolean awaitListingFailure() throws InterruptedException { + return listingFailure.await(10, TimeUnit.SECONDS); + } + + private void releaseFirstDelete() { + releaseFirstDelete.countDown(); + } + + private boolean awaitFirstDeleteReturned() throws InterruptedException { + return firstDeleteReturned.await(10, TimeUnit.SECONDS); + } + + private void releaseSecondDelete() { + releaseSecondDelete.countDown(); + } + + private int activeDeletes() { + return activeDeletes.get(); + } + } + + private static class FailureDrainFileIO extends SortedLocalFileIO { + + private static final long serialVersionUID = 1L; + + private final CountDownLatch firstPairStarted = new CountDownLatch(2); + private final CountDownLatch failureAttempted = new CountDownLatch(1); + private final CountDownLatch releaseSuccessfulSibling = new CountDownLatch(1); + private final ConcurrentLinkedQueue attemptedFiles = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue successfulFiles = new ConcurrentLinkedQueue<>(); + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + String name = path.getName(); + attemptedFiles.add(name); + if ("data-000.csv".equals(name)) { + firstPairStarted.countDown(); + await(firstPairStarted, "both initial deletes to start"); + failureAttempted.countDown(); + throw new IOException("delete failed at input position 0"); + } + if ("data-001.csv".equals(name)) { + firstPairStarted.countDown(); + await(firstPairStarted, "both initial deletes to start"); + await(releaseSuccessfulSibling, "release of in-flight sibling"); + boolean deleted = super.delete(path, recursive); + successfulFiles.add(name); + return deleted; + } + return super.delete(path, recursive); + } + + private boolean awaitFailureAttempted() throws InterruptedException { + return failureAttempted.await(10, TimeUnit.SECONDS); + } + + private void releaseSuccessfulSibling() { + releaseSuccessfulSibling.countDown(); + } + + private ConcurrentLinkedQueue attemptedFiles() { + return attemptedFiles; + } + + private ConcurrentLinkedQueue successfulFiles() { + return successfulFiles; + } + } + + private static class OrderedDualFailureFileIO extends SortedLocalFileIO { + + private static final long serialVersionUID = 1L; + + private final CountDownLatch firstPairStarted = new CountDownLatch(2); + private final CountDownLatch higherPositionFailure = new CountDownLatch(1); + private final CountDownLatch releaseLowerPositionFailure = new CountDownLatch(1); + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + firstPairStarted.countDown(); + await(firstPairStarted, "both failing deletes to start"); + if ("data-001.csv".equals(path.getName())) { + higherPositionFailure.countDown(); + throw new IOException("delete failed at input position 1"); + } + await(releaseLowerPositionFailure, "lower-position failure"); + throw new IOException("delete failed at input position 0"); + } + + private boolean awaitHigherPositionFailure() throws InterruptedException { + return higherPositionFailure.await(10, TimeUnit.SECONDS); + } + + private void releaseLowerPositionFailure() { + releaseLowerPositionFailure.countDown(); + } + } + + private static class MixedOwnershipFileIO extends SortedLocalFileIO { + + private static final long serialVersionUID = 1L; + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + boolean deleted = super.delete(path, recursive); + if (path.toString().contains("month=01")) { + throw new FileNotFoundException("concurrently deleted " + path); + } + if (path.toString().contains("month=02")) { + return false; + } + return deleted; + } + } + + private static class RefusingDeleteFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + @Override + public boolean delete(Path path, boolean recursive) { + return false; + } } private static Map partitionSpec(String year, String month) { @@ -681,7 +1983,8 @@ private FormatTableCommit truncatingCommit( null, null, partitionManager, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); } private FormatTablePartitionManager commitPartitionedFile( @@ -706,7 +2009,8 @@ private FormatTablePartitionManager commitPartitionedFile( null, null, partitionManager, - /* dynamicPartitionOverwrite */ true); + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); return partitionManager; } From 5265e1d00659248e63f05930ec222b812ab1063c Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Thu, 27 Aug 2026 00:33:16 +0800 Subject: [PATCH 3/5] [api][core] Harden format table cleanup concurrency --- docs/generated/core_configuration.html | 2 +- .../java/org/apache/paimon/CoreOptions.java | 5 +- .../apache/paimon/utils/ThreadPoolUtils.java | 91 ++++++------ .../paimon/utils/ThreadPoolUtilsTest.java | 130 ++++++++++++++++-- .../table/format/FormatTableCommit.java | 84 ++++------- .../table/format/FormatTableCommitTest.java | 68 ++------- 6 files changed, 209 insertions(+), 171 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index c2aaf6d6fe01..6919aeae01e7 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -750,7 +750,7 @@
format-table.commit.cleanup-thread-num
64 Integer - The maximum number of concurrent old-data file deletions during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. + The maximum number of concurrent old-data file deletions during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. This limit uses a separate thread pool and is independent of file-operation.thread-num, so the total file-operation concurrency in one process may be the sum of both limits.
format-table.file.compression
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 430b5e362902..07d89878cf05 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2662,7 +2662,10 @@ public String toString() { "The maximum number of concurrent old-data file deletions during " + "overwrite commits for an internal Format Table with " + "catalog-managed partitions. Supported values are 1 through " - + "64. Other Format Tables use serial cleanup."); + + "64. Other Format Tables use serial cleanup. This limit uses " + + "a separate thread pool and is independent of " + + "file-operation.thread-num, so the total file-operation " + + "concurrency in one process may be the sum of both limits."); @Immutable public static final ConfigOption BLOB_FIELD = diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java index 03b47b1e7239..855c0043745a 100644 --- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java +++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java @@ -143,24 +143,7 @@ public static CloseableBatchIterator sequentialBatchedExecuteCloseable Function> processor, List input, int queueSize) { - return sequentialBatchedExecuteCloseable(executor, processor, input.iterator(), queueSize); - } - - /** - * Parallel processes a lazily supplied input, keeping at most {@code queueSize} tasks in flight - * and returning results in input order. The input is read only as slots free, so a caller that - * discovers its work by listing storage does not have to list all of it up front. - * - *

The caller must close the iterator to stop reading input, cancel unstarted tasks and wait - * for running tasks. An input that fails to produce its next element reports that by throwing - * an unchecked exception, as a processor does. - */ - public static CloseableBatchIterator sequentialBatchedExecuteCloseable( - ExecutorService executor, - Function> processor, - Iterator input, - int queueSize) { - return newSequentialBatchIterator(executor, processor, input, queueSize, true); + return newSequentialBatchIterator(executor, processor, input.iterator(), queueSize, true); } /** @@ -271,10 +254,12 @@ private static class SequentialBatchIterator implements CloseableBatchIter private final int queueSize; private final boolean cancelRunningOnClose; private final Queue> activeTasks = new ArrayDeque<>(); + private final Object submissionLock = new Object(); private Iterator activeResults = Collections.emptyList().iterator(); private T next; private boolean closed; + private boolean submissionStopped; private SequentialBatchIterator( ExecutorService executor, @@ -334,10 +319,23 @@ private void advanceIfNeeded() { /** Hands out work until the window is full or the input is exhausted. */ private void fillWindow() { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); - while (activeTasks.size() < queueSize && input.hasNext()) { - BatchTask task = new BatchTask<>(processor, input.next(), classLoader); - executor.execute(task); - activeTasks.add(task); + while (activeTasks.size() < queueSize) { + synchronized (submissionLock) { + if (submissionStopped || !input.hasNext()) { + return; + } + BatchTask task = + new BatchTask<>( + processor, input.next(), classLoader, this::stopSubmission); + executor.execute(task); + activeTasks.add(task); + } + } + } + + private void stopSubmission() { + synchronized (submissionLock) { + submissionStopped = true; } } @@ -352,15 +350,20 @@ public synchronized void close() { boolean interrupted = Thread.interrupted(); for (BatchTask task : activeTasks) { try { - if (cancelRunningOnClose) { - task.cancel(); - } else { - task.cancelIfUnstarted(); - } + task.cancelIfUnstarted(); } catch (Throwable cleanupFailure) { failure = firstOrSuppressed(cleanupFailure, failure); } } + if (cancelRunningOnClose) { + for (BatchTask task : activeTasks) { + try { + task.interruptIfRunning(); + } catch (Throwable cleanupFailure) { + failure = firstOrSuppressed(cleanupFailure, failure); + } + } + } for (BatchTask task : activeTasks) { while (true) { try { @@ -398,6 +401,7 @@ private static class BatchTask implements Runnable { private final Function> processor; private final U input; private final ClassLoader classLoader; + private final Runnable stopSubmission; private final CountDownLatch completion = new CountDownLatch(1); private int state = CREATED; @@ -406,10 +410,15 @@ private static class BatchTask implements Runnable { private Throwable failure; private volatile boolean failureReported; - private BatchTask(Function> processor, U input, ClassLoader classLoader) { + private BatchTask( + Function> processor, + U input, + ClassLoader classLoader, + Runnable stopSubmission) { this.processor = processor; this.input = input; this.classLoader = classLoader; + this.stopSubmission = stopSubmission; } @Override @@ -425,15 +434,20 @@ public void run() { } Thread currentThread = Thread.currentThread(); + boolean wasInterrupted = currentThread.isInterrupted(); ClassLoader originalClassLoader = currentThread.getContextClassLoader(); try { currentThread.setContextClassLoader(classLoader); result = processor.apply(input); } catch (RuntimeException | Error taskFailure) { failure = taskFailure; + stopSubmission.run(); } finally { currentThread.setContextClassLoader(originalClassLoader); Thread.interrupted(); + if (wasInterrupted) { + currentThread.interrupt(); + } synchronized (this) { runner = null; state = FINISHED; @@ -442,31 +456,28 @@ public void run() { } } - private synchronized void cancel() { - if (cancelIfUnstarted()) { - return; - } + private synchronized void interruptIfRunning() { if (state == RUNNING) { runner.interrupt(); } } /** Gives up on a task only while it can still be given up on without interrupting it. */ - private synchronized boolean cancelIfUnstarted() { + private synchronized void cancelIfUnstarted() { if (state == CREATED) { state = CANCELLED; completion.countDown(); - return true; } - return false; } private List result() { - try { - completion.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); + if (completion.getCount() != 0) { + try { + completion.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } } if (failure != null) { failureReported = true; diff --git a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java index 0e931e88165b..6726c6902363 100644 --- a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java @@ -20,6 +20,9 @@ import org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator; +import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators; +import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors; + import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -31,6 +34,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -49,6 +53,7 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti ExecutorService consumer = Executors.newSingleThreadExecutor(); CountDownLatch firstStarted = new CountDownLatch(1); CountDownLatch secondFinished = new CountDownLatch(1); + CountDownLatch thirdStarted = new CountDownLatch(1); CountDownLatch releaseFirst = new CountDownLatch(1); CloseableBatchIterator iterator = ThreadPoolUtils.sequentialBatchedExecuteCloseable( @@ -59,6 +64,8 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti await(releaseFirst); } else if (input == 1) { secondFinished.countDown(); + } else if (input == 2) { + thirdStarted.countDown(); } return Collections.singletonList(input); }, @@ -75,20 +82,18 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti assertThat(firstStarted.await(3, TimeUnit.SECONDS)).isTrue(); assertThat(secondFinished.await(3, TimeUnit.SECONDS)).isTrue(); - assertThat(workers.getTaskCount()).isEqualTo(2); + workers.submit(() -> {}).get(3, TimeUnit.SECONDS); + assertThat(thirdStarted.getCount()).isOne(); assertThat(firstResult.isDone()).isFalse(); releaseFirst.countDown(); List results = new ArrayList<>(); results.add(firstResult.get(3, TimeUnit.SECONDS)); + // Consuming input 0 frees one slot. The next lookup refills it before input 1 is + // consumed, instead of waiting for the whole window to drain. assertThat(iterator.hasNext()).isTrue(); + assertThat(thirdStarted.await(3, TimeUnit.SECONDS)).isTrue(); results.add(iterator.next()); - // Taking the first result freed a slot, and the window refills as slots free rather - // than waiting for every task it handed out to be consumed. - // The window refills as a slot frees rather than a batch at a time, so how many tasks - // have been handed to the pool at a given moment is no longer aligned to batches, and - // ThreadPoolExecutor only reports that count approximately. What the iterator owes the - // caller is the bound asserted above and the order asserted below. assertThat(iterator.hasNext()).isTrue(); results.add(iterator.next()); assertThat(iterator.hasNext()).isTrue(); @@ -126,7 +131,7 @@ public Integer next() { } }; try (CloseableBatchIterator iterator = - ThreadPoolUtils.sequentialBatchedExecuteCloseable( + ThreadPoolUtils.sequentialBatchedExecuteNonCancellable( workers, value -> { if (value == 0) { @@ -136,17 +141,73 @@ public Integer next() { }, input, 4)) { - // A caller that discovers its work by listing storage must not be forced to list all - // of it: only a window's worth of input may be read before the first result is taken. - assertThat(read.get()).isLessThanOrEqualTo(4); releaseFirst.countDown(); assertThat(iterator.next()).isEqualTo(0); - assertThat(read.get()).isLessThan(100); + assertThat(read).hasValue(4); } finally { workers.shutdownNow(); } } + @Test + public void testWorkerFailureStopsNewSubmissions() throws Exception { + ExecutorService workers = Executors.newFixedThreadPool(2); + ExecutorService consumer = Executors.newSingleThreadExecutor(); + CountDownLatch secondFailed = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + AtomicInteger inputsRead = new AtomicInteger(); + RuntimeException workerFailure = new RuntimeException("worker failure"); + Iterator input = + Iterators.transform( + Arrays.asList(0, 1, 2).iterator(), + value -> { + inputsRead.incrementAndGet(); + return value; + }); + CloseableBatchIterator iterator = + ThreadPoolUtils.sequentialBatchedExecuteNonCancellable( + workers, + value -> { + if (value == 0) { + await(releaseFirst); + } else if (value == 1) { + secondFailed.countDown(); + throw workerFailure; + } + return Collections.singletonList(value); + }, + input, + 2); + + try { + Future result = + consumer.submit( + () -> + catchThrowable( + () -> { + assertThat(iterator.next()).isZero(); + iterator.hasNext(); + })); + + assertThat(secondFailed.await(3, TimeUnit.SECONDS)).isTrue(); + // With input 0 still gated, this can only run after the failed task has + // completely left BatchTask.run and published its failure state. + workers.submit(() -> {}).get(3, TimeUnit.SECONDS); + assertThat(inputsRead).hasValue(2); + + releaseFirst.countDown(); + assertThat(result.get(3, TimeUnit.SECONDS)).isSameAs(workerFailure); + assertThat(inputsRead).hasValue(2); + } finally { + releaseFirst.countDown(); + iterator.close(); + consumer.shutdownNow(); + workers.shutdownNow(); + assertThat(consumer.awaitTermination(3, TimeUnit.SECONDS)).isTrue(); + assertThat(workers.awaitTermination(3, TimeUnit.SECONDS)).isTrue(); + } + } + @Test public void testWorkerRunsWithTheCallersClassLoaderAndGivesItBack() throws Exception { ExecutorService workers = Executors.newFixedThreadPool(1); @@ -186,9 +247,47 @@ public void testWorkerRunsWithTheCallersClassLoaderAndGivesItBack() throws Excep workers.shutdownNow(); } + @Test + public void testDirectExecutorPreservesCallerInterrupt() { + ExecutorService workers = MoreExecutors.newDirectExecutorService(); + try { + Thread.currentThread().interrupt(); + try (CloseableBatchIterator iterator = + ThreadPoolUtils.sequentialBatchedExecuteCloseable( + workers, Collections::singletonList, Collections.singletonList(1), 1)) { + assertThat(iterator.next()).isOne(); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } + } finally { + Thread.interrupted(); + workers.shutdownNow(); + } + } + @Test public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exception { - ThreadPoolExecutor workers = (ThreadPoolExecutor) Executors.newFixedThreadPool(1); + LinkedBlockingQueue taskQueue = new LinkedBlockingQueue<>(); + AtomicBoolean runQueuedTaskOnInterrupt = new AtomicBoolean(); + ThreadPoolExecutor workers = + new ThreadPoolExecutor( + 1, + 1, + 0L, + TimeUnit.MILLISECONDS, + taskQueue, + runnable -> + new Thread(runnable) { + @Override + public void interrupt() { + super.interrupt(); + if (runQueuedTaskOnInterrupt.compareAndSet(true, false)) { + Runnable queuedTask = taskQueue.poll(); + if (queuedTask != null) { + queuedTask.run(); + } + } + } + }); ExecutorService closer = Executors.newSingleThreadExecutor(); CountDownLatch secondStarted = new CountDownLatch(1); CountDownLatch workerInterrupted = new CountDownLatch(1); @@ -218,7 +317,9 @@ public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exceptio assertThat(iterator.hasNext()).isTrue(); assertThat(iterator.next()).isZero(); assertThat(secondStarted.await(3, TimeUnit.SECONDS)).isTrue(); - assertThat(workers.getTaskCount()).isEqualTo(3); + assertThat(workers.getQueue()).hasSize(1); + assertThat(thirdExecuted).isFalse(); + runQueuedTaskOnInterrupt.set(true); Future closeResult = closer.submit( @@ -230,6 +331,7 @@ public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exceptio }); assertThat(closeStarted.await(3, TimeUnit.SECONDS)).isTrue(); assertThat(workerInterrupted.await(3, TimeUnit.SECONDS)).isTrue(); + assertThat(runQueuedTaskOnInterrupt).isFalse(); assertThat(closeResult.isDone()).isFalse(); closeThread.get().interrupt(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index 70dbe38135f0..1a36d78a2581 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -40,6 +40,8 @@ import org.apache.paimon.utils.PartitionPathUtils; import org.apache.paimon.utils.ThreadPoolUtils; +import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -57,7 +59,6 @@ import java.util.LinkedHashSet; import java.util.List; import java.util.Map; -import java.util.NoSuchElementException; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.stream.Collectors; @@ -550,7 +551,7 @@ private Set deletePreviousDataFile(Path partitionPath, int partitionLevels private Set deletePreviousDataFiles( List partitionPaths, int partitionLevels, int threadNum) throws IOException { - PreviousDataFiles dataFiles = new PreviousDataFiles(partitionPaths, partitionLevels); + Iterator dataFiles = previousDataFiles(partitionPaths, partitionLevels); Set clearedPartitionPaths = new HashSet<>(); try { if (threadNum == 1) { @@ -609,15 +610,13 @@ private List deleteAndReportCleared(FileStatus file) { /** Deletes one listed data file and reports whether this commit removed it. */ private boolean deleteDataFile(FileStatus file) throws IOException { - boolean deleted; try { - deleted = fileIO.delete(file.getPath(), false); + if (fileIO.delete(file.getPath(), false)) { + return true; + } } catch (FileNotFoundException ignore) { return false; } - if (deleted) { - return true; - } if (fileIO.exists(file.getPath())) { // A refusal is not a concurrent-delete race: the file is still readable, and going on // would report the partition as holding nothing while its rows are still there. @@ -630,54 +629,29 @@ private boolean deleteDataFile(FileStatus file) throws IOException { } /** The committed data files below the given paths, listed one partition at a time. */ - private final class PreviousDataFiles implements Iterator { - - private final List partitionPaths; - private final int partitionLevels; - private List currentFiles = Collections.emptyList(); - private int nextPartition; - private int nextFile; - - private PreviousDataFiles(List partitionPaths, int partitionLevels) { - this.partitionPaths = partitionPaths; - this.partitionLevels = partitionLevels; - } - - @Override - public boolean hasNext() { - while (nextFile >= currentFiles.size()) { - if (nextPartition >= partitionPaths.size()) { - return false; - } - Path partitionPath = partitionPaths.get(nextPartition++); - try { - if (!fileIO.exists(partitionPath)) { - continue; - } - // Committed data files only: what sits under a staging directory is another - // writer's uncommitted output, whatever its name looks like. - currentFiles = - FormatTableScan.listDataFiles( - fileIO, - partitionPath, - partitionLevels, - formatTablePartitionOnlyValueInPath, - defaultPartName); - } catch (IOException e) { - throw new UncheckedIOException(e); - } - nextFile = 0; - } - return true; - } - - @Override - public FileStatus next() { - if (!hasNext()) { - throw new NoSuchElementException(); - } - return currentFiles.get(nextFile++); - } + private Iterator previousDataFiles(List partitionPaths, int partitionLevels) { + return Iterators.concat( + Iterators.transform( + partitionPaths.iterator(), + partitionPath -> { + try { + if (!fileIO.exists(partitionPath)) { + return Collections.emptyList().iterator(); + } + // Committed data files only: what sits under a staging directory is + // another writer's uncommitted output, whatever its name looks + // like. + return FormatTableScan.listDataFiles( + fileIO, + partitionPath, + partitionLevels, + formatTablePartitionOnlyValueInPath, + defaultPartName) + .iterator(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + })); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java index 97e45a16afe1..77ac0ff03d3a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java @@ -61,6 +61,8 @@ import java.util.concurrent.atomic.AtomicReference; import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME; +import static org.apache.paimon.shade.guava30.com.google.common.base.Throwables.getCausalChain; +import static org.apache.paimon.shade.guava30.com.google.common.base.Throwables.getRootCause; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.entry; @@ -695,8 +697,6 @@ void testCatalogManagedBuilderUses64WayCleanupByDefault() throws Exception { new TwoPhaseCommitMessage(committer)))); assertThat(fileIO.awaitFirstWave()).isTrue(); - assertThat(fileIO.deleteCalls()).isEqualTo(64); - assertThat(fileIO.awaitUnexpectedExtraDelete()).isFalse(); verify(committer, never()).commit(fileIO); fileIO.releaseFirstWave(); @@ -754,7 +754,6 @@ void testCatalogManagedBuilderPropagatesConfiguredCleanupConcurrency() throws Ex Future result = executor.submit(() -> commit.commit(Collections.emptyList())); assertThat(fileIO.awaitFirstWave()).isTrue(); - assertThat(fileIO.deleteCalls()).isEqualTo(7); assertThat(fileIO.awaitUnexpectedExtraDelete()).isFalse(); fileIO.releaseFirstWave(); @@ -883,7 +882,7 @@ void testCleanupFailureStopsNewSubmissionsAndDrainsTheAlreadyRunningDelete() thr .containsExactlyInAnyOrder("data-000.csv", "data-001.csv"); fileIO.releaseSuccessfulSibling(); - assertThat(rootCause(awaitFailure(result))) + assertThat(getRootCause(awaitFailure(result))) .hasMessage("delete failed at input position 0"); assertThat(fileIO.attemptedFiles()) .containsExactlyInAnyOrder("data-000.csv", "data-001.csv"); @@ -944,7 +943,7 @@ void testLaterPartitionListingFailureDrainsAcceptedDeletesBeforeAbort() throws E assertThat(discardCalls).hasValue(0); fileIO.releaseSecondDelete(); - assertThat(rootCause(awaitFailure(result))) + assertThat(getRootCause(awaitFailure(result))) .hasMessage("Failed to list the later partition root."); assertThat(discardCalls).hasValue(2); assertThat(activeDeletesAtDiscard).containsExactly(0, 0); @@ -985,7 +984,7 @@ void testCleanupSelectsLowestInputFailureAndSuppressesTheOtherFailure() throws E .isInstanceOf(TimeoutException.class); fileIO.releaseLowerPositionFailure(); - Throwable primary = rootCause(awaitFailure(result)); + Throwable primary = getRootCause(awaitFailure(result)); assertThat(primary).hasMessage("delete failed at input position 0"); assertThat(primary.getSuppressed()) .extracting(Throwable::getMessage) @@ -1034,7 +1033,8 @@ void testInterruptDrainsCleanupRestoresFlagAndNeverPublishes() throws Exception fileIO.releaseDeletes(); assertThat(commitReturned.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(failure.get()).isNotNull(); - assertThat(causeChain(failure.get())).anyMatch(InterruptedException.class::isInstance); + assertThat(getCausalChain(failure.get())) + .anyMatch(InterruptedException.class::isInstance); assertThat(interruptRestored).isTrue(); verify(committer, never()).commit(fileIO); } finally { @@ -1315,7 +1315,7 @@ void testAbortFailureDoesNotReplaceInterruptedCleanupFailure() throws Exception assertThat(commitThread.isAlive()).isFalse(); assertThat(failure.get()).isNotNull(); - assertThat(rootCause(failure.get())).isInstanceOf(InterruptedException.class); + assertThat(getRootCause(failure.get())).isInstanceOf(InterruptedException.class); assertThat(failureTree(failure.get())) .extracting(Throwable::getMessage) .contains("discard failed after cleanup interruption"); @@ -1494,24 +1494,6 @@ private static ExecutionException awaitFailure(Future future) throws Exceptio } } - private static Throwable rootCause(Throwable throwable) { - Throwable root = throwable; - while (root.getCause() != null) { - root = root.getCause(); - } - return root; - } - - private static List causeChain(Throwable throwable) { - List chain = new ArrayList<>(); - Throwable current = throwable; - while (current != null) { - chain.add(current); - current = current.getCause(); - } - return chain; - } - private static List failureTree(Throwable throwable) { List failures = new ArrayList<>(); collectFailures(throwable, failures); @@ -1531,8 +1513,6 @@ private static void collectFailures(Throwable throwable, List failure private static class ParallelDeleteFileIO extends LocalFileIO { - private static final long serialVersionUID = 1L; - private final int firstWaveSize; private final boolean holdFirstWave; private final CountDownLatch firstWave; @@ -1594,10 +1574,6 @@ protected boolean awaitFirstWave() throws InterruptedException { return firstWave.await(10, TimeUnit.SECONDS); } - private boolean awaitFirstWave(long timeout, TimeUnit unit) throws InterruptedException { - return firstWave.await(timeout, unit); - } - private boolean awaitUnexpectedExtraDelete() throws InterruptedException { return unexpectedExtraDelete.await(300, TimeUnit.MILLISECONDS); } @@ -1609,8 +1585,6 @@ protected void releaseFirstWave() { private static class LazyRootListingFileIO extends ParallelDeleteFileIO { - private static final long serialVersionUID = 1L; - private final Path deferredRoot; private final AtomicBoolean deferredRootListed = new AtomicBoolean(); @@ -1634,8 +1608,6 @@ private boolean deferredRootListed() { private static class SerialProbeFileIO extends LocalFileIO { - private static final long serialVersionUID = 1L; - private final CountDownLatch secondDeleteStarted = new CountDownLatch(1); private final AtomicInteger deleteCalls = new AtomicInteger(); private final AtomicInteger activeDeletes = new AtomicInteger(); @@ -1672,8 +1644,6 @@ private int maxConcurrentDeletes() { private static class PartialBarrierDeleteFileIO extends SortedLocalFileIO { - private static final long serialVersionUID = 1L; - private final CountDownLatch bothDeletesStarted = new CountDownLatch(2); private final CountDownLatch releaseFirstDelete = new CountDownLatch(1); private final CountDownLatch releaseSecondDelete = new CountDownLatch(1); @@ -1723,11 +1693,8 @@ private int activeDeletes() { private static class BlockingDeleteFileIO extends LocalFileIO { - private static final long serialVersionUID = 1L; - private final CountDownLatch deletesStarted; private final CountDownLatch releaseDeletes = new CountDownLatch(1); - private final AtomicInteger activeDeletes = new AtomicInteger(); private BlockingDeleteFileIO(int deleteCount) { this.deletesStarted = new CountDownLatch(deleteCount); @@ -1735,7 +1702,6 @@ private BlockingDeleteFileIO(int deleteCount) { @Override public boolean delete(Path path, boolean recursive) throws IOException { - activeDeletes.incrementAndGet(); deletesStarted.countDown(); try { if (!releaseDeletes.await(10, TimeUnit.SECONDS)) { @@ -1745,8 +1711,6 @@ public boolean delete(Path path, boolean recursive) throws IOException { } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while blocking cleanup delete", e); - } finally { - activeDeletes.decrementAndGet(); } } @@ -1757,16 +1721,10 @@ private boolean awaitDeletesStarted() throws InterruptedException { private void releaseDeletes() { releaseDeletes.countDown(); } - - private int activeDeletes() { - return activeDeletes.get(); - } } private abstract static class SortedLocalFileIO extends LocalFileIO { - private static final long serialVersionUID = 1L; - @Override public FileStatus[] listStatus(Path path) throws IOException { FileStatus[] statuses = super.listStatus(path); @@ -1788,8 +1746,6 @@ protected static void await(CountDownLatch latch, String description) throws IOE private static class LaterRootListingFailureFileIO extends SortedLocalFileIO { - private static final long serialVersionUID = 1L; - private final Path failingRoot; private final CountDownLatch deletesStarted; private final CountDownLatch listingFailure = new CountDownLatch(1); @@ -1855,8 +1811,6 @@ private int activeDeletes() { private static class FailureDrainFileIO extends SortedLocalFileIO { - private static final long serialVersionUID = 1L; - private final CountDownLatch firstPairStarted = new CountDownLatch(2); private final CountDownLatch failureAttempted = new CountDownLatch(1); private final CountDownLatch releaseSuccessfulSibling = new CountDownLatch(1); @@ -1903,8 +1857,6 @@ private ConcurrentLinkedQueue successfulFiles() { private static class OrderedDualFailureFileIO extends SortedLocalFileIO { - private static final long serialVersionUID = 1L; - private final CountDownLatch firstPairStarted = new CountDownLatch(2); private final CountDownLatch higherPositionFailure = new CountDownLatch(1); private final CountDownLatch releaseLowerPositionFailure = new CountDownLatch(1); @@ -1932,8 +1884,6 @@ private void releaseLowerPositionFailure() { private static class MixedOwnershipFileIO extends SortedLocalFileIO { - private static final long serialVersionUID = 1L; - @Override public boolean delete(Path path, boolean recursive) throws IOException { boolean deleted = super.delete(path, recursive); @@ -1949,8 +1899,6 @@ public boolean delete(Path path, boolean recursive) throws IOException { private static class RefusingDeleteFileIO extends LocalFileIO { - private static final long serialVersionUID = 1L; - @Override public boolean delete(Path path, boolean recursive) { return false; From 6d3ef330416bb12bc6ef844a5b6219327333aebe Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Thu, 27 Aug 2026 00:48:13 +0800 Subject: [PATCH 4/5] [api][core] Clarify cleanup scheduler contracts Name the mutation scheduler after its close behavior, document interrupt restoration, and make the related tests and configuration wording precise. --- docs/generated/core_configuration.html | 2 +- .../java/org/apache/paimon/CoreOptions.java | 2 +- .../apache/paimon/utils/ThreadPoolUtils.java | 13 +++++----- .../paimon/utils/ThreadPoolUtilsTest.java | 26 +++++++++++-------- .../table/format/FormatTableCommit.java | 17 +++++------- .../paimon/utils/ManifestReadThreadPool.java | 2 +- 6 files changed, 31 insertions(+), 31 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 6919aeae01e7..e544bec90fb5 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -750,7 +750,7 @@

format-table.commit.cleanup-thread-num
64 Integer - The maximum number of concurrent old-data file deletions during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. This limit uses a separate thread pool and is independent of file-operation.thread-num, so the total file-operation concurrency in one process may be the sum of both limits. + The maximum number of concurrent deletions of old data files during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. This limit uses a separate thread pool and is independent of file-operation.thread-num, so the total file-operation concurrency in one process may be the sum of both limits.
format-table.file.compression
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 07d89878cf05..363aec22b29e 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2659,7 +2659,7 @@ public String toString() { .intType() .defaultValue(64) .withDescription( - "The maximum number of concurrent old-data file deletions during " + "The maximum number of concurrent deletions of old data files during " + "overwrite commits for an internal Format Table with " + "catalog-managed partitions. Supported values are 1 through " + "64. Other Format Tables use serial cleanup. This limit uses " diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java index 855c0043745a..e91ba169e304 100644 --- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java +++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java @@ -134,7 +134,7 @@ private void advanceIfNeeded() { } /** - * Parallel processes a bounded number of inputs at a time and returns results in input order. + * Processes a bounded number of inputs in parallel and returns results in input order. * *

The caller must close the iterator to cancel unstarted tasks and wait for running tasks. */ @@ -154,7 +154,7 @@ public static CloseableBatchIterator sequentialBatchedExecuteCloseable * the caller unable to say whether it took effect, so a caller that has to know the outcome of * everything it handed out cannot let close cancel work that is already running. */ - public static CloseableBatchIterator sequentialBatchedExecuteNonCancellable( + public static CloseableBatchIterator sequentialBatchedExecuteAwaitRunningTasksOnClose( ExecutorService executor, Function> processor, Iterator input, @@ -316,7 +316,7 @@ private void advanceIfNeeded() { } } - /** Hands out work until the window is full or the input is exhausted. */ + /** Does not consume more input than the active-task window can hold. */ private void fillWindow() { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); while (activeTasks.size() < queueSize) { @@ -434,7 +434,7 @@ public void run() { } Thread currentThread = Thread.currentThread(); - boolean wasInterrupted = currentThread.isInterrupted(); + boolean interruptedOnEntry = currentThread.isInterrupted(); ClassLoader originalClassLoader = currentThread.getContextClassLoader(); try { currentThread.setContextClassLoader(classLoader); @@ -444,8 +444,10 @@ public void run() { stopSubmission.run(); } finally { currentThread.setContextClassLoader(originalClassLoader); + // Reset the flag to its entry state so a cancelled task cannot leak an interrupt + // to a reused worker and a direct executor cannot clear its caller's interrupt. Thread.interrupted(); - if (wasInterrupted) { + if (interruptedOnEntry) { currentThread.interrupt(); } synchronized (this) { @@ -462,7 +464,6 @@ private synchronized void interruptIfRunning() { } } - /** Gives up on a task only while it can still be given up on without interrupting it. */ private synchronized void cancelIfUnstarted() { if (state == CREATED) { state = CANCELLED; diff --git a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java index 7aa1e97ec362..899c7842fffc 100644 --- a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java @@ -111,10 +111,10 @@ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws Excepti } @Test - public void testLazyInputIsReadOnlyAsSlotsFree() throws Exception { + public void testLazyInputIsConsumedOnlyAsSlotsFree() throws Exception { ExecutorService workers = Executors.newFixedThreadPool(2); CountDownLatch releaseFirst = new CountDownLatch(1); - AtomicInteger read = new AtomicInteger(); + AtomicInteger inputsRead = new AtomicInteger(); Iterator input = new Iterator() { private int next; @@ -126,12 +126,12 @@ public boolean hasNext() { @Override public Integer next() { - read.incrementAndGet(); + inputsRead.incrementAndGet(); return next++; } }; try (CloseableBatchIterator iterator = - ThreadPoolUtils.sequentialBatchedExecuteNonCancellable( + ThreadPoolUtils.sequentialBatchedExecuteAwaitRunningTasksOnClose( workers, value -> { if (value == 0) { @@ -143,7 +143,7 @@ public Integer next() { 4)) { releaseFirst.countDown(); assertThat(iterator.next()).isEqualTo(0); - assertThat(read).hasValue(4); + assertThat(inputsRead).hasValue(4); } finally { workers.shutdownNow(); } @@ -165,7 +165,7 @@ public void testWorkerFailureStopsNewSubmissions() throws Exception { return value; }); CloseableBatchIterator iterator = - ThreadPoolUtils.sequentialBatchedExecuteNonCancellable( + ThreadPoolUtils.sequentialBatchedExecuteAwaitRunningTasksOnClose( workers, value -> { if (value == 0) { @@ -209,7 +209,7 @@ public void testWorkerFailureStopsNewSubmissions() throws Exception { } @Test - public void testWorkerRunsWithTheCallersClassLoaderAndGivesItBack() throws Exception { + public void testWorkerUsesCallerClassLoaderAndRestoresPoolClassLoader() throws Exception { ExecutorService workers = Executors.newFixedThreadPool(1); ClassLoader callerClassLoader = new ClassLoader(getClass().getClassLoader()) {}; AtomicReference poolClassLoader = new AtomicReference<>(); @@ -240,10 +240,13 @@ public void testWorkerRunsWithTheCallersClassLoaderAndGivesItBack() throws Excep assertThat(seen).containsExactly(callerClassLoader, callerClassLoader); // The pool is shared, so a worker that keeps a caller's loader would hand it to whatever // runs on that thread next. - AtomicReference afterwards = new AtomicReference<>(); - workers.submit(() -> afterwards.set(Thread.currentThread().getContextClassLoader())) + AtomicReference restoredPoolClassLoader = new AtomicReference<>(); + workers.submit( + () -> + restoredPoolClassLoader.set( + Thread.currentThread().getContextClassLoader())) .get(10, TimeUnit.SECONDS); - assertThat(afterwards.get()).isSameAs(poolClassLoader.get()); + assertThat(restoredPoolClassLoader.get()).isSameAs(poolClassLoader.get()); workers.shutdownNow(); } @@ -268,6 +271,8 @@ public void testDirectExecutorPreservesCallerInterrupt() { public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exception { LinkedBlockingQueue taskQueue = new LinkedBlockingQueue<>(); AtomicBoolean runQueuedTaskOnInterrupt = new AtomicBoolean(); + // If close interrupts the worker before cancelling queued tasks, interrupt() runs the + // queued task and exposes the ordering bug. ThreadPoolExecutor workers = new ThreadPoolExecutor( 1, @@ -422,5 +427,4 @@ private static void awaitIgnoringInterrupts(CountDownLatch latch, CountDownLatch } } } - } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index 1a36d78a2581..e54f67b557f1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -65,7 +65,7 @@ import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition; import static org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator; -import static org.apache.paimon.utils.ThreadPoolUtils.sequentialBatchedExecuteNonCancellable; +import static org.apache.paimon.utils.ThreadPoolUtils.sequentialBatchedExecuteAwaitRunningTasksOnClose; /** Commit for Format Table. */ public class FormatTableCommit implements BatchTableCommit { @@ -568,31 +568,26 @@ private Set deletePreviousDataFiles( // the iterator is what stops new deletes and waits for the ones already handed out, // so a failure cannot leave a worker still deleting after this method returns. try (CloseableBatchIterator cleared = - sequentialBatchedExecuteNonCancellable( + sequentialBatchedExecuteAwaitRunningTasksOnClose( CLEANUP_EXECUTOR, this::deleteAndReportCleared, dataFiles, threadNum)) { while (cleared.hasNext()) { clearedPartitionPaths.add(cleared.next()); } } } catch (UncheckedIOException e) { - throw (IOException) withoutUncheckedWrapper(e); + throw (IOException) unwrapUncheckedIOException(e); } return clearedPartitionPaths; } - /** - * Unwraps the {@link UncheckedIOException} a worker has to throw to cross a {@link - * java.util.function.Function}, so a caller sees the {@link IOException} the file system raised - * rather than the wrapper. Whatever else failed while the accepted deletes drained is attached - * to the wrapper, so it has to travel with the cause rather than be dropped with it. - */ - private static Throwable withoutUncheckedWrapper(Throwable failure) { + /** Unwraps worker I/O failures while preserving recursively suppressed failures. */ + private static Throwable unwrapUncheckedIOException(Throwable failure) { if (!(failure instanceof UncheckedIOException)) { return failure; } Throwable unwrapped = failure.getCause(); for (Throwable suppressed : failure.getSuppressed()) { - unwrapped.addSuppressed(withoutUncheckedWrapper(suppressed)); + unwrapped.addSuppressed(unwrapUncheckedIOException(suppressed)); } return unwrapped; } diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java index ffd7a8fe0c37..1e106cb3d4e1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java @@ -61,7 +61,7 @@ public static Iterable sequentialBatchedExecute( executor, processor, input, effectiveThreadNum(threadNum, executor)); } - /** This method parallel processes a bounded number of inputs and waits for them when closed. */ + /** Processes a bounded number of inputs in parallel and waits for submitted tasks on close. */ public static ThreadPoolUtils.CloseableBatchIterator sequentialBatchedExecuteCloseable( Function> processor, List input, @Nullable Integer threadNum) { From 01fc733b7da61a8b683ab1dd19123f41f164629e Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Thu, 27 Aug 2026 00:37:09 +0800 Subject: [PATCH 5/5] [core] Parallelize format table file publication --- docs/generated/core_configuration.html | 6 + .../java/org/apache/paimon/CoreOptions.java | 20 ++ .../apache/paimon/utils/ThreadPoolUtils.java | 18 +- .../paimon/utils/ThreadPoolUtilsTest.java | 51 +++ .../table/format/FormatBatchWriteBuilder.java | 7 +- .../table/format/FormatTableCommit.java | 163 +++++++-- .../org/apache/paimon/CoreOptionsTest.java | 25 ++ .../table/format/FormatTableCommitTest.java | 327 +++++++++++++++++- 8 files changed, 591 insertions(+), 26 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index e544bec90fb5..4dca9eb1d6b7 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -752,6 +752,12 @@ Integer The maximum number of concurrent deletions of old data files during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. This limit uses a separate thread pool and is independent of file-operation.thread-num, so the total file-operation concurrency in one process may be the sum of both limits. + +

format-table.commit.publish-thread-num
+ 64 + Integer + The maximum number of concurrent file publications during commits for a partitioned Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables publish serially. +
format-table.file.compression
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 363aec22b29e..9d2b2c6b5927 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2667,6 +2667,16 @@ public String toString() { + "file-operation.thread-num, so the total file-operation " + "concurrency in one process may be the sum of both limits."); + public static final ConfigOption FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM = + key("format-table.commit.publish-thread-num") + .intType() + .defaultValue(64) + .withDescription( + "The maximum number of concurrent file publications during commits " + + "for a partitioned Format Table with catalog-managed " + + "partitions. Supported values are 1 through 64. Other Format " + + "Tables publish serially."); + @Immutable public static final ConfigOption BLOB_FIELD = key("blob-field") @@ -3325,6 +3335,16 @@ public int formatTableCommitCleanupThreadNum() { return threadNum; } + public int formatTableCommitPublishThreadNum() { + int threadNum = options.get(FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM); + checkArgument( + threadNum >= 1 && threadNum <= 64, + "Option %s must be between 1 and 64, but was %s.", + FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM.key(), + threadNum); + return threadNum; + } + public MemorySize fileReaderAsyncThreshold() { return options.get(FILE_READER_ASYNC_THRESHOLD); } diff --git a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java index e91ba169e304..37db694e93c3 100644 --- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java +++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java @@ -21,6 +21,9 @@ import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators; import org.apache.paimon.shade.guava30.com.google.common.collect.Lists; +import java.security.AccessControlContext; +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; @@ -319,6 +322,7 @@ private void advanceIfNeeded() { /** Does not consume more input than the active-task window can hold. */ private void fillWindow() { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + AccessControlContext accessControlContext = AccessController.getContext(); while (activeTasks.size() < queueSize) { synchronized (submissionLock) { if (submissionStopped || !input.hasNext()) { @@ -326,7 +330,11 @@ private void fillWindow() { } BatchTask task = new BatchTask<>( - processor, input.next(), classLoader, this::stopSubmission); + processor, + input.next(), + classLoader, + accessControlContext, + this::stopSubmission); executor.execute(task); activeTasks.add(task); } @@ -401,6 +409,7 @@ private static class BatchTask implements Runnable { private final Function> processor; private final U input; private final ClassLoader classLoader; + private final AccessControlContext accessControlContext; private final Runnable stopSubmission; private final CountDownLatch completion = new CountDownLatch(1); @@ -414,10 +423,12 @@ private BatchTask( Function> processor, U input, ClassLoader classLoader, + AccessControlContext accessControlContext, Runnable stopSubmission) { this.processor = processor; this.input = input; this.classLoader = classLoader; + this.accessControlContext = accessControlContext; this.stopSubmission = stopSubmission; } @@ -438,7 +449,10 @@ public void run() { ClassLoader originalClassLoader = currentThread.getContextClassLoader(); try { currentThread.setContextClassLoader(classLoader); - result = processor.apply(input); + result = + AccessController.doPrivileged( + (PrivilegedAction>) () -> processor.apply(input), + accessControlContext); } catch (RuntimeException | Error taskFailure) { failure = taskFailure; stopSubmission.run(); diff --git a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java index 899c7842fffc..b61593482f0e 100644 --- a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java +++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java @@ -25,6 +25,10 @@ import org.junit.jupiter.api.Test; +import javax.security.auth.Subject; + +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -267,6 +271,53 @@ public void testDirectExecutorPreservesCallerInterrupt() { } } + @Test + public void testWorkerRunsWithTheSubmittingSubject() throws Exception { + ExecutorService workers = + ThreadPoolUtils.createCachedThreadPool(1, "subject-propagation-test"); + Subject firstSubject = new Subject(); + Subject secondSubject = new Subject(); + List seenSubjects = new ArrayList<>(); + List seenWorkers = new ArrayList<>(); + + try { + for (Subject subject : Arrays.asList(firstSubject, secondSubject)) { + seenSubjects.add( + Subject.doAs( + subject, + (PrivilegedAction) + () -> { + try (CloseableBatchIterator iterator = + ThreadPoolUtils + .sequentialBatchedExecuteCloseable( + workers, + ignored -> { + seenWorkers.add( + Thread + .currentThread()); + return Collections + .singletonList( + Subject + .getSubject( + AccessController + .getContext())); + }, + Collections.singletonList(0), + 1)) { + return iterator.next(); + } + })); + } + + assertThat(seenWorkers.get(1)).isSameAs(seenWorkers.get(0)); + assertThat(seenSubjects.get(0)).isSameAs(firstSubject); + assertThat(seenSubjects.get(1)).isSameAs(secondSubject); + } finally { + workers.shutdownNow(); + assertThat(workers.awaitTermination(3, TimeUnit.SECONDS)).isTrue(); + } + } + @Test public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws Exception { LinkedBlockingQueue taskQueue = new LinkedBlockingQueue<>(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java index 4c8beca9c175..f9d08a5c1bfb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java @@ -82,6 +82,10 @@ public BatchTableCommit newCommit() { table.partitionManager() != null && !table.partitionKeys().isEmpty() ? options.formatTableCommitCleanupThreadNum() : 1; + int publishThreadNum = + table.partitionManager() != null && !table.partitionKeys().isEmpty() + ? options.formatTableCommitPublishThreadNum() + : 1; return new FormatTableCommit( table.location(), table.partitionKeys(), @@ -95,7 +99,8 @@ public BatchTableCommit newCommit() { table.catalogContext(), table.partitionManager(), options.dynamicPartitionOverwrite(), - cleanupThreadNum); + cleanupThreadNum, + publishThreadNum); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index e54f67b557f1..f1442d85b0d6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -64,6 +64,7 @@ import java.util.stream.Collectors; import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition; +import static org.apache.paimon.utils.ExceptionUtils.firstOrSuppressed; import static org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator; import static org.apache.paimon.utils.ThreadPoolUtils.sequentialBatchedExecuteAwaitRunningTasksOnClose; @@ -72,11 +73,11 @@ public class FormatTableCommit implements BatchTableCommit { private static final Logger LOG = LoggerFactory.getLogger(FormatTableCommit.class); - private static final int MAX_CLEANUP_THREAD_NUM = 64; + private static final int MAX_COMMIT_THREAD_NUM = 64; - private static final ExecutorService CLEANUP_EXECUTOR = + private static final ExecutorService COMMIT_EXECUTOR = ThreadPoolUtils.createCachedThreadPool( - MAX_CLEANUP_THREAD_NUM, "FORMAT-TABLE-COMMIT-CLEANUP-THREAD-POOL"); + MAX_COMMIT_THREAD_NUM, "FORMAT-TABLE-COMMIT-THREAD-POOL"); private String location; private final boolean formatTablePartitionOnlyValueInPath; @@ -90,6 +91,7 @@ public class FormatTableCommit implements BatchTableCommit { @Nullable private final FormatTablePartitionManager partitionManager; private final boolean dynamicPartitionOverwrite; private final int cleanupThreadNum; + private final int publishThreadNum; public FormatTableCommit( String location, @@ -105,11 +107,49 @@ public FormatTableCommit( @Nullable FormatTablePartitionManager partitionManager, boolean dynamicPartitionOverwrite, int cleanupThreadNum) { - if (cleanupThreadNum < 1 || cleanupThreadNum > MAX_CLEANUP_THREAD_NUM) { + this( + location, + partitionKeys, + fileIO, + formatTablePartitionOnlyValueInPath, + defaultPartName, + overwrite, + tableIdentifier, + staticPartitions, + syncHiveUri, + catalogContext, + partitionManager, + dynamicPartitionOverwrite, + cleanupThreadNum, + 1); + } + + FormatTableCommit( + String location, + List partitionKeys, + FileIO fileIO, + boolean formatTablePartitionOnlyValueInPath, + String defaultPartName, + boolean overwrite, + Identifier tableIdentifier, + @Nullable Map staticPartitions, + @Nullable String syncHiveUri, + CatalogContext catalogContext, + @Nullable FormatTablePartitionManager partitionManager, + boolean dynamicPartitionOverwrite, + int cleanupThreadNum, + int publishThreadNum) { + if (cleanupThreadNum < 1 || cleanupThreadNum > MAX_COMMIT_THREAD_NUM) { throw new IllegalArgumentException( String.format( "Format Table cleanup thread number must be between 1 and %s, but was %s.", - MAX_CLEANUP_THREAD_NUM, cleanupThreadNum)); + MAX_COMMIT_THREAD_NUM, cleanupThreadNum)); + } + if (publishThreadNum < 1 || publishThreadNum > MAX_COMMIT_THREAD_NUM) { + throw new IllegalArgumentException( + String.format( + "Format Table publish thread number must be between 1 and %s, but was %s.", + MAX_COMMIT_THREAD_NUM, publishThreadNum)); } this.location = location; this.fileIO = fileIO; @@ -123,6 +163,7 @@ public FormatTableCommit( this.partitionManager = partitionManager; this.dynamicPartitionOverwrite = dynamicPartitionOverwrite; this.cleanupThreadNum = cleanupThreadNum; + this.publishThreadNum = publishThreadNum; if (syncHiveUri != null) { try { Options options = new Options(); @@ -209,9 +250,9 @@ public void commit(List commitMessages) { boolean reportsStatistics = registersPartitions && partitionManager != null; Map, PartitionStatistics> statisticsByPartition = new LinkedHashMap<>(); + publishMessages(messages); for (TwoPhaseCommitMessage message : messages) { TwoPhaseOutputStream.Committer committer = message.getCommitter(); - committer.commit(this.fileIO); if (registersPartitions) { // Extracted once: registration and statistics must key on the same spec. Map spec = @@ -289,6 +330,37 @@ public void commit(List commitMessages) { } } + private void publishMessages(List messages) throws IOException { + if (publishThreadNum == 1 || messages.size() <= 1) { + for (TwoPhaseCommitMessage message : messages) { + message.getCommitter().commit(fileIO); + } + return; + } + + try (CloseableBatchIterator published = + sequentialBatchedExecuteAwaitRunningTasksOnClose( + COMMIT_EXECUTOR, + this::publishMessage, + messages.iterator(), + publishThreadNum)) { + while (published.hasNext()) { + published.next(); + } + } catch (UncheckedIOException e) { + throw (IOException) unwrapUncheckedIOException(e); + } + } + + private List publishMessage(TwoPhaseCommitMessage message) { + try { + message.getCommitter().commit(fileIO); + return Collections.emptyList(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + /** * Registers the partitions this commit touched, carrying the statistics of what it wrote. An * overwrite also empties partitions it writes nothing to - those below a static prefix, and @@ -465,20 +537,69 @@ private static Path buildPartitionPath( @Override public void abort(List commitMessages) { - try { - for (CommitMessage commitMessage : commitMessages) { - if (commitMessage instanceof TwoPhaseCommitMessage) { - TwoPhaseCommitMessage twoPhaseCommitMessage = - (TwoPhaseCommitMessage) commitMessage; - twoPhaseCommitMessage.getCommitter().discard(this.fileIO); - } else { - throw new RuntimeException( - "Unsupported commit message type: " - + commitMessage.getClass().getName()); - } + Throwable failure = null; + for (CommitMessage commitMessage : commitMessages) { + if (!(commitMessage instanceof TwoPhaseCommitMessage)) { + failure = + firstOrSuppressed( + new RuntimeException( + "Unsupported commit message type: " + + commitMessage.getClass().getName()), + failure); + continue; + } + + TwoPhaseOutputStream.Committer committer = + ((TwoPhaseCommitMessage) commitMessage).getCommitter(); + try { + committer.discard(fileIO); + } catch (Throwable discardFailure) { + failure = firstOrSuppressed(discardFailure, failure); } - } catch (Exception e) { - throw new RuntimeException(e); + + // FormatTableSingleFileWriter opens every target with overwrite=false. The target is + // therefore owned by this write attempt, so it is safe to remove even when a remote + // multipart completion took effect but its response was lost. Keep this rollback here: + // a generic multipart committer may also be used to overwrite an existing object. + try { + deletePublishedFile(committer.targetPath()); + } catch (Throwable deleteFailure) { + failure = firstOrSuppressed(deleteFailure, failure); + } + } + + if (failure instanceof Error) { + throw (Error) failure; + } + if (failure != null) { + throw new RuntimeException(failure); + } + } + + private void deletePublishedFile(Path targetPath) throws IOException { + String failureMessage = "Failed to delete published Format Table file " + targetPath; + boolean deleted; + try { + deleted = fileIO.delete(targetPath, false); + } catch (FileNotFoundException ignored) { + return; + } catch (IOException e) { + throw new IOException(failureMessage, e); + } + if (deleted) { + return; + } + + boolean stillExists; + try { + stillExists = fileIO.exists(targetPath); + } catch (FileNotFoundException ignored) { + return; + } catch (IOException e) { + throw new IOException(failureMessage, e); + } + if (stillExists) { + throw new IOException(failureMessage); } } @@ -569,7 +690,7 @@ private Set deletePreviousDataFiles( // so a failure cannot leave a worker still deleting after this method returns. try (CloseableBatchIterator cleared = sequentialBatchedExecuteAwaitRunningTasksOnClose( - CLEANUP_EXECUTOR, this::deleteAndReportCleared, dataFiles, threadNum)) { + COMMIT_EXECUTOR, this::deleteAndReportCleared, dataFiles, threadNum)) { while (cleared.hasNext()) { clearedPartitionPaths.add(cleared.next()); } @@ -592,7 +713,7 @@ private static Throwable unwrapUncheckedIOException(Throwable failure) { return unwrapped; } - /** Deletes one file and reports the partition it emptied, for a worker that cannot throw. */ + /** Deletes one listed file and reports its parent when this commit removed the file. */ private List deleteAndReportCleared(FileStatus file) { try { return deleteDataFile(file) diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index f3e81179d733..ea7513208702 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -325,4 +325,29 @@ public void testFormatTableCommitCleanupThreadNumRejectsValuesOutsideSupportedRa .hasMessageContaining("1") .hasMessageContaining("64"); } + + @Test + public void testFormatTableCommitPublishThreadNumDefaultsTo64AndAcceptsBounds() { + Options conf = new Options(); + assertThat(new CoreOptions(conf).formatTableCommitPublishThreadNum()).isEqualTo(64); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM, 1); + assertThat(new CoreOptions(conf).formatTableCommitPublishThreadNum()).isEqualTo(1); + + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM, 64); + assertThat(new CoreOptions(conf).formatTableCommitPublishThreadNum()).isEqualTo(64); + } + + @Test + public void testFormatTableCommitPublishThreadNumRejectsValuesOutsideSupportedRange() { + for (int invalid : new int[] {0, -1, 65}) { + Options conf = new Options(); + conf.set(CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM, invalid); + assertThatThrownBy(() -> new CoreOptions(conf).formatTableCommitPublishThreadNum()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("format-table.commit.publish-thread-num") + .hasMessageContaining("1") + .hasMessageContaining("64"); + } + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java index 77ac0ff03d3a..e892d010dafa 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java @@ -65,6 +65,7 @@ import static org.apache.paimon.shade.guava30.com.google.common.base.Throwables.getRootCause; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.catchThrowable; import static org.assertj.core.api.Assertions.entry; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; @@ -158,6 +159,52 @@ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception { .createPartitions(anyList(), eq(true), any(), anyBoolean()); } + @Test + void testAbortAttemptsEveryRollbackAndReportsDeleteFailure() throws Exception { + Path tablePath = new Path(tempDir.toUri()); + Path refusedPath = new Path(tablePath, "part=p/data-refused.csv"); + Path removablePath = new Path(tablePath, "part=p/data-removable.csv"); + SelectiveRefusingDeleteFileIO fileIO = new SelectiveRefusingDeleteFileIO(refusedPath); + fileIO.writeFile(refusedPath, "published", false); + fileIO.writeFile(removablePath, "published", false); + + TwoPhaseOutputStream.Committer first = mock(TwoPhaseOutputStream.Committer.class); + when(first.targetPath()).thenReturn(refusedPath); + doThrow(new IOException("discard failed")).when(first).discard(fileIO); + TwoPhaseOutputStream.Committer second = mock(TwoPhaseOutputStream.Committer.class); + when(second.targetPath()).thenReturn(removablePath); + List messages = + Arrays.asList(new TwoPhaseCommitMessage(first), new TwoPhaseCommitMessage(second)); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Collections.singletonList("part"), + fileIO, + false, + PARTITION_DEFAULT_NAME.defaultValue(), + false, + Identifier.create("rollback_db", "rollback_table"), + null, + null, + null, + null, + /* dynamicPartitionOverwrite */ true, + /* cleanupThreadNum */ 1); + + Throwable failure = catchThrowable(() -> commit.abort(messages)); + + assertThat(failure).isInstanceOf(RuntimeException.class); + assertThat(failureTree(failure)) + .extracting(Throwable::getMessage) + .contains( + "discard failed", + "Failed to delete published Format Table file " + refusedPath); + verify(first).discard(fileIO); + verify(second).discard(fileIO); + assertThat(fileIO.exists(refusedPath)).isTrue(); + assertThat(fileIO.exists(removablePath)).isFalse(); + } + @Test void testRegistersRawPartitionValuesForEscapedPath() throws Exception { Path tablePath = new Path(tempDir.toUri()); @@ -710,6 +757,243 @@ void testCatalogManagedBuilderUses64WayCleanupByDefault() throws Exception { } } + @Test + void testCatalogManagedBuilderPublishesSamePartitionConcurrentlyAndWaitsForBarrier() + throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + CountDownLatch firstTwoStarted = new CountDownLatch(2); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch releaseSecond = new CountDownLatch(1); + CountDownLatch thirdFinished = new CountDownLatch(1); + AtomicInteger activePublishes = new AtomicInteger(); + AtomicInteger maxConcurrentPublishes = new AtomicInteger(); + List committers = new ArrayList<>(); + List messages = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + int index = i; + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(partitionPath, "data-" + i + ".csv")); + doAnswer( + invocation -> { + int active = activePublishes.incrementAndGet(); + maxConcurrentPublishes.updateAndGet( + previous -> Math.max(previous, active)); + try { + if (index == 0) { + firstTwoStarted.countDown(); + if (!releaseFirst.await(10, TimeUnit.SECONDS)) { + throw new IOException( + "Timed out waiting to release first publication"); + } + } else if (index == 1) { + firstTwoStarted.countDown(); + if (!releaseSecond.await(10, TimeUnit.SECONDS)) { + throw new IOException( + "Timed out waiting to release second publication"); + } + } + return null; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("File publication was interrupted", e); + } finally { + activePublishes.decrementAndGet(); + if (index == 2) { + thirdFinished.countDown(); + } + } + }) + .when(committer) + .commit(fileIO); + committers.add(committer); + messages.add(new TwoPhaseCommitMessage(committer)); + } + FormatTableCommit commit = + (FormatTableCommit) + formatTable( + tablePath, + fileIO, + partitionManager, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM + .key(), + "2")) + .newBatchWriteBuilder() + .newCommit(); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit(() -> commit.commit(messages)); + try { + assertThat(firstTwoStarted.await(3, TimeUnit.SECONDS)).isTrue(); + assertThat(activePublishes).hasValue(2); + + releaseFirst.countDown(); + assertThat(thirdFinished.await(3, TimeUnit.SECONDS)).isTrue(); + assertThat(activePublishes).hasValue(1); + verify(partitionManager, never()) + .createPartitions(anyList(), eq(true), any(), anyBoolean()); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer, never()).clean(fileIO); + } + + releaseSecond.countDown(); + result.get(10, TimeUnit.SECONDS); + + assertThat(maxConcurrentPublishes).hasValue(2); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer).clean(fileIO); + } + verify(partitionManager).createPartitions(anyList(), eq(true), any(), eq(false)); + } finally { + releaseFirst.countDown(); + releaseSecond.countDown(); + if (!result.isDone()) { + result.get(10, TimeUnit.SECONDS); + } + executor.shutdownNow(); + } + } + + @Test + void testPublishFailureDrainsRunningWorkBeforeAbort() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path partitionPath = new Path(tablePath, "part=p"); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + CountDownLatch secondStarted = new CountDownLatch(1); + CountDownLatch releaseSecond = new CountDownLatch(1); + AtomicInteger activePublishes = new AtomicInteger(); + ConcurrentLinkedQueue activePublishesAtDiscard = new ConcurrentLinkedQueue<>(); + List committers = new ArrayList<>(); + List targetPaths = new ArrayList<>(); + List messages = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + Path targetPath = new Path(partitionPath, "data-" + i + ".csv"); + when(committer.targetPath()).thenReturn(targetPath); + doAnswer( + invocation -> { + activePublishesAtDiscard.add(activePublishes.get()); + return null; + }) + .when(committer) + .discard(fileIO); + committers.add(committer); + targetPaths.add(targetPath); + messages.add(new TwoPhaseCommitMessage(committer)); + } + doAnswer( + invocation -> { + activePublishes.incrementAndGet(); + try { + if (!secondStarted.await(10, TimeUnit.SECONDS)) { + throw new IOException("The second publication did not start"); + } + fileIO.writeFile(targetPaths.get(0), "published", false); + throw new IOException("publish failed"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("File publication was interrupted", e); + } finally { + activePublishes.decrementAndGet(); + } + }) + .when(committers.get(0)) + .commit(fileIO); + doAnswer( + invocation -> { + activePublishes.incrementAndGet(); + secondStarted.countDown(); + try { + if (!releaseSecond.await(10, TimeUnit.SECONDS)) { + throw new IOException( + "Timed out waiting to release publication"); + } + fileIO.writeFile(targetPaths.get(1), "published", false); + return null; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("File publication was interrupted", e); + } finally { + activePublishes.decrementAndGet(); + } + }) + .when(committers.get(1)) + .commit(fileIO); + FormatTableCommit commit = + (FormatTableCommit) + formatTable( + tablePath, + fileIO, + partitionManager, + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM + .key(), + "2")) + .newBatchWriteBuilder() + .newCommit(); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit(() -> commit.commit(messages)); + try { + assertThat(secondStarted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThatThrownBy(() -> result.get(300, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer, never()).discard(fileIO); + } + + releaseSecond.countDown(); + assertThat(getRootCause(awaitFailure(result))).hasMessage("publish failed"); + + verify(committers.get(2), never()).commit(fileIO); + assertThat(activePublishesAtDiscard).containsExactly(0, 0, 0); + for (TwoPhaseOutputStream.Committer committer : committers) { + verify(committer).discard(fileIO); + } + assertThat(fileIO.exists(targetPaths.get(0))).isFalse(); + assertThat(fileIO.exists(targetPaths.get(1))).isFalse(); + } finally { + releaseSecond.countDown(); + if (!result.isDone()) { + try { + result.get(10, TimeUnit.SECONDS); + } catch (ExecutionException ignored) { + // The test expects the first publication to fail. + } + } + executor.shutdownNow(); + } + } + + @Test + void testPublishConcurrencyIsGatedToCatalogManagedPartitionedTables() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Map options = + Collections.singletonMap( + CoreOptions.FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM.key(), "64"); + + FormatTableCommit filesystemDiscovered = + (FormatTableCommit) + formatTable(new Path(tablePath, "filesystem"), fileIO, null, options) + .newBatchWriteBuilder() + .newCommit(); + assertPublishesOnCaller( + filesystemDiscovered, fileIO, new Path(tablePath, "filesystem/part=p")); + + FormatTableCommit unpartitioned = + builderUnpartitionedOverwriteCommit( + new Path(tablePath, "unpartitioned"), + fileIO, + mock(FormatTablePartitionManager.class), + options); + assertPublishesOnCaller(unpartitioned, fileIO, new Path(tablePath, "unpartitioned")); + } + @Test void testCatalogManagedBuilderHonorsConfiguredSerialCleanup() throws Exception { SerialProbeFileIO fileIO = new SerialProbeFileIO(); @@ -885,7 +1169,7 @@ void testCleanupFailureStopsNewSubmissionsAndDrainsTheAlreadyRunningDelete() thr assertThat(getRootCause(awaitFailure(result))) .hasMessage("delete failed at input position 0"); assertThat(fileIO.attemptedFiles()) - .containsExactlyInAnyOrder("data-000.csv", "data-001.csv"); + .containsExactlyInAnyOrder("data-000.csv", "data-001.csv", "data-new.csv"); assertThat(fileIO.successfulFiles()).containsExactly("data-001.csv"); verify(committer, never()).commit(fileIO); } finally { @@ -1485,10 +1769,33 @@ private static void writeOldFiles(LocalFileIO fileIO, Path partitionPath, int co } } + private static void assertPublishesOnCaller( + FormatTableCommit commit, FileIO fileIO, Path parent) throws IOException { + Thread caller = Thread.currentThread(); + ConcurrentLinkedQueue publishThreads = new ConcurrentLinkedQueue<>(); + List messages = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + when(committer.targetPath()).thenReturn(new Path(parent, "data-" + i + ".csv")); + doAnswer( + invocation -> { + publishThreads.add(Thread.currentThread()); + return null; + }) + .when(committer) + .commit(fileIO); + messages.add(new TwoPhaseCommitMessage(committer)); + } + + commit.commit(messages); + + assertThat(publishThreads).containsExactly(caller, caller); + } + private static ExecutionException awaitFailure(Future future) throws Exception { try { future.get(10, TimeUnit.SECONDS); - throw new AssertionError("Expected cleanup commit to fail"); + throw new AssertionError("Expected Format Table commit to fail"); } catch (ExecutionException expected) { return expected; } @@ -1511,6 +1818,22 @@ private static void collectFailures(Throwable throwable, List failure collectFailures(throwable.getCause(), failures); } + private static class SelectiveRefusingDeleteFileIO extends LocalFileIO { + + private static final long serialVersionUID = 1L; + + private final Path refusedPath; + + private SelectiveRefusingDeleteFileIO(Path refusedPath) { + this.refusedPath = refusedPath; + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + return path.equals(refusedPath) ? false : super.delete(path, recursive); + } + } + private static class ParallelDeleteFileIO extends LocalFileIO { private final int firstWaveSize;