multiPartUploadStore(FileIO fileIO) throws IOException {
if (fileIO instanceof RESTTokenFileIO) {
RESTTokenFileIO restTokenFileIO = (RESTTokenFileIO) fileIO;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java
index 931969ec68cb..1fa0d7cfd371 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java
@@ -53,6 +53,19 @@ public interface Committer extends Serializable {
*/
void discard(FileIO fileIO) throws IOException;
+ /**
+ * Discards staged resources without deleting {@link #targetPath()}.
+ *
+ * This is used when a commit may have taken effect and its target must therefore be
+ * preserved. The default delegates to {@link #clean}. Override this method if a failed or
+ * uncertain commit can leave staged resources that {@code clean} does not release.
+ *
+ * @throws IOException if an I/O error occurs during cleanup
+ */
+ default void discardStaging(FileIO fileIO) throws IOException {
+ clean(fileIO);
+ }
+
Path targetPath();
/**
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..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
@@ -78,6 +78,14 @@ 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;
+ int publishThreadNum =
+ table.partitionManager() != null && !table.partitionKeys().isEmpty()
+ ? options.formatTableCommitPublishThreadNum()
+ : 1;
return new FormatTableCommit(
table.location(),
table.partitionKeys(),
@@ -90,7 +98,9 @@ public BatchTableCommit newCommit() {
syncHiveUri,
table.catalogContext(),
table.partitionManager(),
- options.dynamicPartitionOverwrite());
+ options.dynamicPartitionOverwrite(),
+ 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 be2105930a57..76de97dcfb39 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,9 @@
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.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,24 +49,42 @@
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.lang.reflect.Method;
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.ArrayDeque;
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.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
+import static org.apache.paimon.utils.ExceptionUtils.firstOrSuppressed;
/** Commit for Format Table. */
public class FormatTableCommit implements BatchTableCommit {
private static final Logger LOG = LoggerFactory.getLogger(FormatTableCommit.class);
+ private static final int MAX_COMMIT_THREAD_NUM = 64;
+
+ private static final ExecutorService COMMIT_EXECUTOR =
+ ThreadPoolUtils.createCachedThreadPool(
+ MAX_COMMIT_THREAD_NUM, "FORMAT-TABLE-COMMIT-THREAD-POOL");
+
private String location;
private final boolean formatTablePartitionOnlyValueInPath;
private final String defaultPartName;
@@ -75,6 +96,8 @@ public class FormatTableCommit implements BatchTableCommit {
private Identifier tableIdentifier;
@Nullable private final FormatTablePartitionManager partitionManager;
private final boolean dynamicPartitionOverwrite;
+ private final int cleanupThreadNum;
+ private final int publishThreadNum;
public FormatTableCommit(
String location,
@@ -89,6 +112,50 @@ public FormatTableCommit(
CatalogContext catalogContext,
@Nullable FormatTablePartitionManager partitionManager,
boolean dynamicPartitionOverwrite) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ defaultPartName,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ partitionManager,
+ dynamicPartitionOverwrite,
+ 1,
+ 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_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;
this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath;
@@ -100,6 +167,8 @@ public FormatTableCommit(
this.tableIdentifier = tableIdentifier;
this.partitionManager = partitionManager;
this.dynamicPartitionOverwrite = dynamicPartitionOverwrite;
+ this.cleanupThreadNum = cleanupThreadNum;
+ this.publishThreadNum = publishThreadNum;
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -136,6 +205,7 @@ public void commit(List commitMessages) {
Set> partitionSpecs = new HashSet<>();
Set clearedPartitionPaths = new HashSet<>();
+ Path staticPartitionPath = null;
if (staticPartitions != null && !staticPartitions.isEmpty()) {
Path partitionPath =
@@ -144,6 +214,7 @@ public void commit(List commitMessages) {
staticPartitions,
formatTablePartitionOnlyValueInPath,
partitionKeys);
+ staticPartitionPath = partitionPath;
if (staticPartitions.size() == partitionKeys.size()) {
partitionSpecs.add(staticPartitions);
}
@@ -151,34 +222,38 @@ 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()));
- }
- if (!fileIO.exists(partitionPath)) {
- fileIO.mkdirs(partitionPath);
+ deletePreviousDataFiles(
+ Collections.singletonList(partitionPath),
+ partitionKeys.size() - staticPartitions.size(),
+ cleanupThreadNum));
}
} 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));
}
}
+ if (overwrite) {
+ // Old data is now permanently gone. Preserve any replacement that may become
+ // visible, while abort still cleans its staging resources.
+ markPublishedTargetsToPreserveOnAbort(messages);
+ }
+ if (staticPartitionPath != null && !fileIO.exists(staticPartitionPath)) {
+ fileIO.mkdirs(staticPartitionPath);
+ }
boolean registersPartitions =
partitionKeys != null
@@ -187,9 +262,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 =
@@ -213,7 +288,7 @@ public void commit(List commitMessages) {
for (TwoPhaseCommitMessage message : messages) {
message.getCommitter().clean(this.fileIO);
}
- if (reportsStatistics) {
+ if (reportsStatistics && overwrite) {
reportPartitions(
partitionSpecs,
statisticsByPartition,
@@ -221,8 +296,9 @@ public void commit(List commitMessages) {
commitTime,
overwrite);
} else if (partitionManager != null && !partitionSpecs.isEmpty()) {
- // Concurrent writers may touch the same partition, so registration ignores the
- // ones that already exist rather than failing the commit.
+ // Register an append before reporting its additive statistics. Registration is
+ // idempotent, so a failed multi-batch call can roll back every file from this
+ // attempt and leave any completed batches as harmless empty partition entries.
partitionManager.createPartitions(new ArrayList<>(partitionSpecs), true);
}
for (Map partitionSpec : partitionSpecs) {
@@ -243,10 +319,88 @@ public void commit(List commitMessages) {
}
}
}
+ if (!overwrite && registersPartitions) {
+ // Every partition registration is now complete. A later abort must not remove
+ // these visible files, while a failed additive report must not make the engine
+ // retry the data write and add the same rows again.
+ markPublishedTargetsToPreserveOnAbort(messages);
+ if (reportsStatistics && !statisticsByPartition.isEmpty()) {
+ try {
+ reportPartitions(
+ partitionSpecs,
+ statisticsByPartition,
+ clearedPartitionPaths,
+ commitTime,
+ false);
+ } catch (RuntimeException statisticsFailure) {
+ LOG.warn(
+ "Committed data for format table {}, but failed to report append "
+ + "statistics for {} partitions. Run ANALYZE TABLE {} "
+ + "COMPUTE STATISTICS to refresh the partition statistics.",
+ tableIdentifier.getFullName(),
+ partitionSpecs.size(),
+ tableIdentifier.getFullName(),
+ statisticsFailure);
+ }
+ }
+ }
- } 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);
+ }
+ }
+
+ private void publishMessages(List messages) throws IOException {
+ if (publishThreadNum == 1 || messages.size() <= 1) {
+ for (TwoPhaseCommitMessage message : messages) {
+ message.getCommitter().commit(fileIO);
+ }
+ return;
+ }
+
+ try {
+ executeSideEffects(
+ COMMIT_EXECUTOR,
+ this::publishMessage,
+ messages.iterator(),
+ publishThreadNum,
+ ignored -> {});
+ } 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);
+ }
+ }
+
+ private static void markPublishedTargetsToPreserveOnAbort(
+ List messages) {
+ for (TwoPhaseCommitMessage message : messages) {
+ message.markPublishedTargetToPreserveOnAbort();
}
}
@@ -278,8 +432,6 @@ private void reportPartitions(
if (specs.isEmpty()) {
return;
}
- // A commit that replaced what the partitions held reports a total; an appending one saw
- // only its own files, so its numbers are an increment.
partitionManager.createPartitions(
new ArrayList<>(specs),
true,
@@ -426,20 +578,79 @@ 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);
+ Throwable failure = null;
+ for (CommitMessage commitMessage : commitMessages) {
+ if (!(commitMessage instanceof TwoPhaseCommitMessage)) {
+ failure =
+ firstOrSuppressed(
+ new RuntimeException(
+ "Unsupported commit message type: "
+ + commitMessage.getClass().getName()),
+ failure);
+ continue;
+ }
+
+ TwoPhaseCommitMessage twoPhaseCommitMessage = (TwoPhaseCommitMessage) commitMessage;
+ TwoPhaseOutputStream.Committer committer = twoPhaseCommitMessage.getCommitter();
+ boolean preservePublishedTarget =
+ twoPhaseCommitMessage.shouldPreservePublishedTargetOnAbort();
+ try {
+ if (preservePublishedTarget) {
+ committer.discardStaging(fileIO);
} else {
- throw new RuntimeException(
- "Unsupported commit message type: "
- + commitMessage.getClass().getName());
+ committer.discard(fileIO);
}
+ } catch (Throwable discardFailure) {
+ failure = firstOrSuppressed(discardFailure, failure);
+ }
+
+ if (preservePublishedTarget) {
+ continue;
}
- } 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);
}
}
@@ -506,40 +717,339 @@ 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 {
+ Iterator dataFiles = 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 {
+ if (threadNum == 1) {
+ while (dataFiles.hasNext()) {
+ FileStatus file = dataFiles.next();
+ if (deleteDataFile(file)) {
+ clearedPartitionPaths.add(file.getPath().getParent());
+ }
+ }
+ 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. The local
+ // runner stops filling its window and waits for the deletes already handed out, so a
+ // failure cannot leave a worker still deleting after this method returns.
+ executeSideEffects(
+ COMMIT_EXECUTOR,
+ this::deleteAndReportCleared,
+ dataFiles,
+ threadNum,
+ clearedPartitionPaths::add);
+ } catch (UncheckedIOException e) {
+ throw (IOException) unwrapUncheckedIOException(e);
+ }
+ return clearedPartitionPaths;
+ }
+
+ /**
+ * Runs a bounded sliding window of side effects and consumes their results in input order.
+ *
+ * Once a failure is observed, the runner stops filling the window. Tasks which have not
+ * started are cancelled, while running tasks are allowed to finish and are drained before the
+ * failure is returned, so rollback cannot race a side effect already handed to the executor.
+ */
+ private static void executeSideEffects(
+ ExecutorService executor,
+ Function> processor,
+ Iterator input,
+ int maxConcurrency,
+ Consumer resultConsumer) {
+ AtomicBoolean submissionStopped = new AtomicBoolean();
+ ArrayDeque> activeTasks = new ArrayDeque<>(maxConcurrency);
+ long nextInputPosition = 0;
+ Throwable failure = null;
+
+ try {
+ while (true) {
+ while (activeTasks.size() < maxConcurrency && !submissionStopped.get()) {
+ if (!input.hasNext() || submissionStopped.get()) {
+ break;
+ }
+ I nextInput = input.next();
+ if (submissionStopped.get()) {
+ break;
+ }
+ SideEffectTask task =
+ new SideEffectTask<>(
+ processor,
+ nextInput,
+ nextInputPosition++,
+ Thread.currentThread().getContextClassLoader(),
+ AccessController.getContext(),
+ submissionStopped);
+ if (submissionStopped.get()) {
+ break;
+ }
+ // Add before execute so a rejected submission is covered by the drain below.
+ activeTasks.addLast(task);
+ executor.execute(task);
+ }
+
+ if (activeTasks.isEmpty()) {
+ return;
+ }
+
+ SideEffectTask first = activeTasks.getFirst();
+ for (O result : first.result()) {
+ resultConsumer.accept(result);
+ }
+ activeTasks.removeFirst();
+ }
+ } catch (Throwable sideEffectFailure) {
+ failure = sideEffectFailure;
+ submissionStopped.set(true);
+ }
+
+ boolean interrupted = Thread.interrupted();
+ for (SideEffectTask task : activeTasks) {
+ try {
+ task.cancelIfUnstarted();
+ } catch (Throwable cancellationFailure) {
+ failure = firstOrSuppressed(cancellationFailure, failure);
+ }
+ }
+ // ArrayDeque iteration is input order, so the earliest worker failure is primary unless a
+ // caller-side listing, submission, consumption, or interruption failure initiated drain.
+ for (SideEffectTask task : activeTasks) {
+ while (true) {
try {
- deleted = fileIO.delete(file.getPath(), false);
- } catch (FileNotFoundException ignore) {
- continue;
- } catch (IOException e) {
- throw new RuntimeException(e);
+ task.awaitCompletion();
+ break;
+ } catch (InterruptedException ignored) {
+ interrupted = true;
}
- 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()));
+ }
+ Throwable taskFailure = task.unreportedFailure();
+ if (taskFailure != null) {
+ failure = firstOrSuppressed(taskFailure, failure);
+ }
+ }
+
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ throw rethrowSideEffectFailure(failure);
+ }
+
+ private static RuntimeException rethrowSideEffectFailure(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure instanceof RuntimeException) {
+ return (RuntimeException) failure;
+ }
+ return new RuntimeException(failure);
+ }
+
+ private static class SideEffectTask implements Runnable {
+
+ private static final int CREATED = 0;
+ private static final int RUNNING = 1;
+ private static final int CANCELLED = 2;
+ private static final int FINISHED = 3;
+
+ private final Function> processor;
+ private final I input;
+ private final long inputPosition;
+ private final ClassLoader callerClassLoader;
+ private final AccessControlContext callerAccessControlContext;
+ private final AtomicBoolean submissionStopped;
+ private final CountDownLatch completion = new CountDownLatch(1);
+
+ private int state = CREATED;
+ private List result;
+ private Throwable failure;
+ private volatile boolean failureReported;
+
+ private SideEffectTask(
+ Function> processor,
+ I input,
+ long inputPosition,
+ ClassLoader callerClassLoader,
+ AccessControlContext callerAccessControlContext,
+ AtomicBoolean submissionStopped) {
+ this.processor = processor;
+ this.input = input;
+ this.inputPosition = inputPosition;
+ this.callerClassLoader = callerClassLoader;
+ this.callerAccessControlContext = callerAccessControlContext;
+ this.submissionStopped = submissionStopped;
+ }
+
+ @Override
+ public void run() {
+ synchronized (this) {
+ // A queued task which reaches a worker after another task failed has not started
+ // its side effect and is safe to skip. The volatile stop check is the
+ // linearization point between a task which was already running and one which can
+ // still be cancelled.
+ if (state == CANCELLED || submissionStopped.get()) {
+ result = Collections.emptyList();
+ state = FINISHED;
+ completion.countDown();
+ return;
+ }
+ state = RUNNING;
+ }
+
+ Thread currentThread = Thread.currentThread();
+ boolean interruptedOnEntry = currentThread.isInterrupted();
+ ClassLoader workerClassLoader = null;
+ boolean workerClassLoaderCaptured = false;
+ try {
+ try {
+ workerClassLoader = currentThread.getContextClassLoader();
+ workerClassLoaderCaptured = true;
+ currentThread.setContextClassLoader(callerClassLoader);
+ result =
+ AccessController.doPrivileged(
+ (PrivilegedAction>) () -> processor.apply(input),
+ callerAccessControlContext);
+ } catch (RuntimeException | Error taskFailure) {
+ failure = taskFailure;
+ } finally {
+ if (workerClassLoaderCaptured) {
+ try {
+ currentThread.setContextClassLoader(workerClassLoader);
+ } catch (RuntimeException | Error restoreFailure) {
+ failure = firstOrSuppressed(restoreFailure, failure);
+ }
+ }
+ }
+ if (failure != null) {
+ submissionStopped.set(true);
+ }
+ } finally {
+ try {
+ synchronized (this) {
+ state = FINISHED;
+ }
+ // Do not leak an interrupt into a reused worker, while preserving the entry
+ // state for an executor which runs tasks directly on the caller thread.
+ Thread.interrupted();
+ if (interruptedOnEntry) {
+ currentThread.interrupt();
+ }
+ } finally {
+ completion.countDown();
}
}
}
- return clearedPartitionPaths;
+
+ private synchronized void cancelIfUnstarted() {
+ if (state == CREATED) {
+ state = CANCELLED;
+ completion.countDown();
+ }
+ }
+
+ private List result() {
+ if (completion.getCount() != 0) {
+ try {
+ completion.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ if (failure != null) {
+ failureReported = true;
+ throw rethrowSideEffectFailure(failure);
+ }
+ return result;
+ }
+
+ private void awaitCompletion() throws InterruptedException {
+ completion.await();
+ }
+
+ private Throwable unreportedFailure() {
+ return failureReported ? null : failure;
+ }
+
+ @Override
+ public String toString() {
+ return "FormatTableSideEffectTask{inputPosition=" + inputPosition + '}';
+ }
+ }
+
+ /** 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(unwrapUncheckedIOException(suppressed));
+ }
+ return unwrapped;
+ }
+
+ /** Deletes one listed file and reports its parent when this commit removed the file. */
+ 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 {
+ try {
+ if (fileIO.delete(file.getPath(), false)) {
+ return true;
+ }
+ } catch (FileNotFoundException ignore) {
+ return false;
+ }
+ 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 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/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java b/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java
index f44c5e9b8904..c936e7ccbb0a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/TwoPhaseCommitMessage.java
@@ -40,6 +40,10 @@ public class TwoPhaseCommitMessage implements CommitMessage {
private final long recordCount;
private final long fileSizeInBytes;
+ // Set once an abort must preserve the published target. Serializing the flag lets a later abort
+ // instance preserve it too.
+ private boolean preservePublishedTargetOnAbort;
+
public TwoPhaseCommitMessage(TwoPhaseOutputStream.Committer committer) {
this(committer, PartitionStatistics.UNKNOWN, PartitionStatistics.UNKNOWN);
}
@@ -70,6 +74,14 @@ public TwoPhaseOutputStream.Committer getCommitter() {
return committer;
}
+ void markPublishedTargetToPreserveOnAbort() {
+ preservePublishedTargetOnAbort = true;
+ }
+
+ boolean shouldPreservePublishedTargetOnAbort() {
+ return preservePublishedTargetOnAbort;
+ }
+
/** Rows in this file, or {@link PartitionStatistics#UNKNOWN} when nobody counted them. */
public long recordCount() {
return recordCount;
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..93bde01a2f1b 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,54 @@ 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() {
+ for (int invalid : new int[] {0, -1, 65}) {
+ Options conf = new Options();
+ conf.set(CoreOptions.FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM, invalid);
+ assertThatThrownBy(() -> new CoreOptions(conf).formatTableCommitCleanupThreadNum())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("format-table.commit.cleanup-thread-num")
+ .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/FormatTableCommitCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
new file mode 100644
index 000000000000..632ea63c46d2
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCommitCompatibilityTest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.format.FormatTableCommit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+
+/** Compatibility tests for the public Format Table commit API. */
+class FormatTableCommitCompatibilityTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testLegacyPublicConstructorCanCommitOutsideFormatPackage() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ try (FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.emptyList(),
+ LocalFileIO.create(),
+ false,
+ "__DEFAULT_PARTITION__",
+ false,
+ Identifier.create("compatibility_db", "compatibility_table"),
+ null,
+ null,
+ null,
+ null,
+ true)) {
+ commit.commit(Collections.emptyList());
+ }
+ }
+}
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..5ad2c306e180 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
@@ -19,6 +19,8 @@
package org.apache.paimon.table.format;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
@@ -39,6 +41,7 @@
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InstantiationUtil;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -55,6 +58,7 @@
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
import static org.assertj.core.api.Assertions.assertThat;
@@ -63,6 +67,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;
@@ -971,31 +976,130 @@ void testTheNumbersReachTheCatalogThroughTheWriteBuilder() throws Exception {
}
@Test
- void testAFailedReportFailsTheCommitAndDiscardsWhatItWrote() throws Exception {
+ void testAppendRegistrationLoaderFailureDeletesPublishedTarget() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
- FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
- RuntimeException failure = new RuntimeException("catalog says 429");
- doThrow(failure)
- .when(partitionManager)
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean());
+ RuntimeException failure = new RuntimeException("catalog loader failed");
+ AtomicInteger loads = new AtomicInteger();
+ CatalogLoader loader =
+ () -> {
+ loads.incrementAndGet();
+ throw failure;
+ };
+ FormatTablePartitionManager partitionManager =
+ FormatTablePartitionManager.create(TABLE, PARTITION_KEYS, loader);
CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128);
Path written = ((TwoPhaseCommitMessage) message).getCommitter().targetPath();
- // Registration and statistics ride in one request, so a failure says nothing about
- // whether the partition was registered. Committing anyway would leave data behind that
- // nothing points at.
assertThatThrownBy(
() ->
commit(tablePath, fileIO, partitionManager, false, null)
.commit(Collections.singletonList(message)))
.hasRootCause(failure);
+ // No catalog request ran, so abort can safely delete this attempt's published file.
+ assertThat(loads).hasValue(1);
assertThat(fileIO.exists(written)).isFalse();
}
@Test
- void testAFailedReportOfAnOverwriteLeavesThePartitionEmpty() throws Exception {
+ void testEmptyStaticAppendOnlyRegistersPartition() {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ Map staticPartition = spec("2025", "10");
+
+ commit(tablePath, fileIO, partitionManager, false, staticPartition)
+ .commit(Collections.emptyList());
+
+ verify(partitionManager).createPartitions(Collections.singletonList(staticPartition), true);
+ verify(partitionManager, never())
+ .createPartitions(anyList(), eq(true), any(), anyBoolean());
+ }
+
+ @Test
+ void testAppendRegistrationBatchFailureDeletesAllTargetsWithoutReportingStatistics()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Catalog catalog = mock(Catalog.class);
+ Catalog.TableNoPermissionException failure = new Catalog.TableNoPermissionException(TABLE);
+ AtomicInteger requests = new AtomicInteger();
+ List> registered = new ArrayList<>();
+ List appliedStatistics = new ArrayList<>();
+ doAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ List> batch = invocation.getArgument(1);
+ @SuppressWarnings("unchecked")
+ List statistics = invocation.getArgument(3);
+ if (requests.incrementAndGet() == 2) {
+ // Permission is checked before the catalog mutation, so the second
+ // batch was not applied.
+ throw failure;
+ }
+ registered.addAll(batch);
+ if (statistics != null) {
+ appliedStatistics.addAll(statistics);
+ }
+ return null;
+ })
+ .when(catalog)
+ .createPartitions(any(), anyList(), anyBoolean(), any(), anyBoolean());
+ FormatTablePartitionManager partitionManager =
+ FormatTablePartitionManager.create(TABLE, PARTITION_KEYS, () -> catalog);
+ List messages = new ArrayList<>();
+ List targets = new ArrayList<>();
+ for (int i = 0; i < 1001; i++) {
+ CommitMessage message =
+ writtenFile(fileIO, tablePath, String.format("year=2025/month=%04d", i), 1, 1);
+ messages.add(message);
+ targets.add(((TwoPhaseCommitMessage) message).getCommitter().targetPath());
+ }
+
+ assertThatThrownBy(
+ () ->
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .commit(messages))
+ .hasRootCause(failure);
+
+ assertThat(requests).hasValue(2);
+ assertThat(registered).hasSize(1000);
+ // Partition rows left by a successful registration batch are harmlessly empty. Applying
+ // additive statistics before every registration succeeds would instead leave them
+ // describing files this failed attempt rolls back.
+ assertThat(appliedStatistics).isEmpty();
+ assertThat(targets).allSatisfy(target -> assertThat(fileIO.exists(target)).isFalse());
+ }
+
+ @Test
+ void testAppendStatisticsResponseLossIsNotRetriedAndDoesNotFailCommit() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ ApplyingThenFailingStatisticsManager partitionManager =
+ new ApplyingThenFailingStatisticsManager();
+ CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128);
+ Path written = ((TwoPhaseCommitMessage) message).getCommitter().targetPath();
+
+ // The catalog may have applied an additive report before its response was lost. Retrying
+ // that report would double count it, so the data commit succeeds after one best-effort try.
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .commit(Collections.singletonList(message));
+
+ assertThat(partitionManager.calls).containsExactly("registration", "statistics");
+ assertThat(partitionManager.statisticsAttempts).isOne();
+ assertThat(partitionManager.appliedRecordCount).isEqualTo(3);
+ assertThat(fileIO.exists(written)).isTrue();
+
+ TwoPhaseCommitMessage roundTripped =
+ InstantiationUtil.clone((TwoPhaseCommitMessage) message);
+ commit(tablePath, fileIO, partitionManager, false, null)
+ .abort(Collections.singletonList(roundTripped));
+ assertThat(fileIO.exists(written)).isTrue();
+ }
+
+ @Test
+ void testFailedOverwriteReportPreservesReplacementAfterDeletingOldData() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
@@ -1017,13 +1121,58 @@ void testAFailedReportOfAnOverwriteLeavesThePartitionEmpty() throws Exception {
.commit(Collections.singletonList(message)))
.hasRootCauseMessage("catalog says 429");
- // The state this leaves is worth stating rather than discovering: the overwrite already
- // deleted what the partition held, and the abort takes back what it wrote, so the
- // partition is empty on disk while the catalog still describes what used to be there.
- assertThat(fileIO.exists(written)).isFalse();
+ // The old file cannot be restored. Keep the replacement because the failed catalog call
+ // may already have made its metadata durable.
+ assertThat(fileIO.exists(written)).isTrue();
assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=10/old-data.csv"))).isFalse();
}
+ /** A catalog whose additive report takes effect before its response is lost. */
+ private static class ApplyingThenFailingStatisticsManager
+ implements FormatTablePartitionManager {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List calls = new ArrayList<>();
+ private int statisticsAttempts;
+ private long appliedRecordCount;
+
+ @Override
+ public void createPartitions(
+ List> partitions,
+ boolean ignoreIfExists,
+ @Nullable List statistics,
+ boolean replaceStatistics) {
+ if (statistics == null) {
+ calls.add("registration");
+ return;
+ }
+
+ calls.add("statistics");
+ statisticsAttempts++;
+ for (PartitionStatistics statistic : statistics) {
+ appliedRecordCount += statistic.recordCount();
+ }
+ throw new RuntimeException("statistics response lost");
+ }
+
+ @Override
+ public List listPartitions(
+ Map prefix, @Nullable Predicate filter) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitionsByNames(List> partitions) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropPartitions(List> partitions) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
/** What one call reported to the catalog. */
private static class Reported {
private final List> specs;
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..737fd65adbfa 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,34 +18,78 @@
package org.apache.paimon.table.format;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.BaseMultiPartUploadCommitter;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.MultiPartUploadStore;
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.InstantiationUtil;
import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.ReflectionUtils;
+
+import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.mockito.ArgumentCaptor;
+import javax.security.auth.Subject;
+
+import java.io.FileNotFoundException;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.Callable;
+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 java.util.function.Consumer;
+import java.util.function.Function;
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.catchThrowable;
import static org.assertj.core.api.Assertions.entry;
import static org.mockito.ArgumentMatchers.any;
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;
@@ -58,7 +102,7 @@ class FormatTableCommitTest {
@TempDir java.nio.file.Path tempDir;
@Test
- void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception {
+ void testPartitionRegistrationFailureDeletesPublishedTarget() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
Path targetPath = new Path(tablePath, "year=2025/month=10/data-1.csv");
@@ -69,34 +113,220 @@ void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
RuntimeException registrationFailure =
new RuntimeException("Catalog partition registration unavailable");
- doThrow(registrationFailure)
- .when(partitionManager)
+ doThrow(registrationFailure).when(partitionManager).createPartitions(anyList(), eq(true));
+ Identifier identifier =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ identifier,
+ null,
+ null,
+ null,
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
+ TwoPhaseCommitMessage message = new TwoPhaseCommitMessage(committer);
+ List messages = Collections.singletonList(message);
+
+ assertThatThrownBy(() -> commit.commit(messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("Catalog partition registration unavailable");
+ assertThat(fileIO.exists(targetPath)).isFalse();
+ verify(partitionManager).createPartitions(anyList(), eq(true));
+ verify(partitionManager, never())
.createPartitions(anyList(), eq(true), any(), anyBoolean());
+ }
+ @Test
+ void testRegistrationResponseLossStillDeletesPublishedTarget() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "part=p/data.csv");
+ Identifier identifier =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+ Catalog catalog = mock(Catalog.class);
+ List> registeredPartitions = new ArrayList<>();
+ RuntimeException registrationFailure = new RuntimeException("registration response lost");
+ doAnswer(
+ invocation -> {
+ List> batch = invocation.getArgument(1);
+ registeredPartitions.addAll(batch);
+ throw registrationFailure;
+ })
+ .when(catalog)
+ .createPartitions(eq(identifier), anyList(), eq(true), eq(null), eq(false));
+ FormatTablePartitionManager partitionManager =
+ FormatTablePartitionManager.create(
+ identifier, Collections.singletonList("part"), () -> catalog);
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ CommitMessage message = new TwoPhaseCommitMessage(outputStream.closeForCommit());
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
- Arrays.asList("year", "month"),
+ Collections.singletonList("part"),
fileIO,
false,
PARTITION_DEFAULT_NAME.defaultValue(),
false,
- Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ identifier,
null,
null,
null,
partitionManager,
/* dynamicPartitionOverwrite */ true);
- CommitMessage message = new TwoPhaseCommitMessage(committer);
assertThatThrownBy(() -> commit.commit(Collections.singletonList(message)))
.isInstanceOf(RuntimeException.class)
- .hasRootCauseMessage("Catalog partition registration unavailable");
+ .hasRootCauseMessage("registration response lost");
+
+ // Registration without statistics is idempotent. Even if it took effect before the
+ // response was lost, deleting this attempt's unique file leaves a safe empty partition.
+ assertThat(registeredPartitions).containsExactly(Collections.singletonMap("part", "p"));
+ assertThat(fileIO.exists(targetPath)).isFalse();
+ verify(catalog, never())
+ .createPartitions(eq(identifier), anyList(), eq(true), anyList(), eq(false));
+ }
+
+ @Test
+ void testHivePostRegistrationFailurePreservesOverwriteTarget() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path partitionPath = new Path(tablePath, "part=p");
+ Path oldPath = new Path(partitionPath, "data-old.csv");
+ Path targetPath = new Path(partitionPath, "data-new.csv");
+ fileIO.writeFile(oldPath, "old", false);
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit();
+ Map staticPartition = Collections.singletonMap("part", "p");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("hive_db", "hive_table"),
+ staticPartition,
+ null,
+ null,
+ null,
+ /* dynamicPartitionOverwrite */ true);
+ PostRegistrationFailingHiveCatalog hiveCatalog =
+ new PostRegistrationFailingHiveCatalog(fileIO, tablePath);
+ ReflectionUtils.setPrivateFieldValue(commit, "hiveCatalog", hiveCatalog);
+ List messages =
+ Collections.singletonList(new TwoPhaseCommitMessage(committer));
+
+ assertThatThrownBy(() -> commit.commit(messages))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("Hive failed after partition registration");
+
+ // The replacement is the only remaining copy after Hive mutates and then fails.
+ assertThat(hiveCatalog.registeredPartitions).containsExactly(staticPartition);
+ assertThat(fileIO.exists(oldPath)).isFalse();
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ }
+
+ @Test
+ void testHivePostRegistrationFailureDeletesAppendTarget() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "part=p/data-new.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ Map staticPartition = Collections.singletonMap("part", "p");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ Identifier.create("hive_db", "hive_table"),
+ staticPartition,
+ null,
+ null,
+ null,
+ /* dynamicPartitionOverwrite */ true);
+ PostRegistrationFailingHiveCatalog hiveCatalog =
+ new PostRegistrationFailingHiveCatalog(fileIO, tablePath);
+ ReflectionUtils.setPrivateFieldValue(commit, "hiveCatalog", hiveCatalog);
+
+ assertThatThrownBy(
+ () ->
+ commit.commit(
+ Collections.singletonList(
+ new TwoPhaseCommitMessage(
+ outputStream.closeForCommit()))))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("Hive failed after partition registration");
- // A failed write leaves nothing behind, whichever step failed: rerunning it converges,
- // and an idempotent registration makes a partition that was registered anyway harmless.
+ // Hive registration is idempotent and carries no file statistics. Its empty partition may
+ // remain, while removing this attempt's file makes a retry safe.
+ assertThat(hiveCatalog.registeredPartitions).containsExactly(staticPartition);
assertThat(fileIO.exists(targetPath)).isFalse();
- verify(partitionManager).createPartitions(anyList(), eq(true), any(), anyBoolean());
+ }
+
+ @Test
+ void testSuccessfulHiveAppendSurvivesAbortAfterMessageRoundTrip() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "part=p/data-new.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseCommitMessage message = new TwoPhaseCommitMessage(outputStream.closeForCommit());
+ Map staticPartition = Collections.singletonMap("part", "p");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ Identifier.create("hive_db", "hive_table"),
+ staticPartition,
+ null,
+ null,
+ null,
+ /* dynamicPartitionOverwrite */ true);
+ RecordingHiveCatalog hiveCatalog = new RecordingHiveCatalog(fileIO, tablePath);
+ ReflectionUtils.setPrivateFieldValue(commit, "hiveCatalog", hiveCatalog);
+
+ commit.commit(Collections.singletonList(message));
+
+ TwoPhaseCommitMessage roundTripped = InstantiationUtil.clone(message);
+ FormatTableCommit abortCommit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ Identifier.create("hive_db", "hive_table"),
+ staticPartition,
+ null,
+ null,
+ null,
+ /* dynamicPartitionOverwrite */ true);
+ abortCommit.abort(Collections.singletonList(roundTripped));
+
+ assertThat(hiveCatalog.registeredPartitions).containsExactly(staticPartition);
+ assertThat(fileIO.exists(targetPath)).isTrue();
}
@Test
@@ -131,6 +361,202 @@ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception {
.createPartitions(anyList(), eq(true), any(), anyBoolean());
}
+ @Test
+ void testStagingCleanupFailureDeletesPublishedFileBeforeMetadataMutation() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "part=p/data.csv");
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ when(committer.targetPath()).thenReturn(targetPath);
+ doAnswer(
+ ignored -> {
+ fileIO.writeFile(targetPath, "published", false);
+ return null;
+ })
+ .when(committer)
+ .commit(fileIO);
+ doThrow(new IOException("staging cleanup failed")).when(committer).clean(fileIO);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ false,
+ Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ null,
+ null,
+ null,
+ partitionManager,
+ /* dynamicPartitionOverwrite */ true);
+
+ assertThatThrownBy(
+ () ->
+ commit.commit(
+ Collections.singletonList(
+ new TwoPhaseCommitMessage(committer))))
+ .hasRootCauseMessage("staging cleanup failed");
+
+ verify(committer).discard(fileIO);
+ verify(partitionManager, never())
+ .createPartitions(anyList(), eq(true), any(), anyBoolean());
+ assertThat(fileIO.exists(targetPath)).isFalse();
+ }
+
+ @Test
+ void testOverwriteMultipartCompletionResponseLossPreservesReplacementAndReportsAbortFailure()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "multipart-response-loss");
+ Path partitionPath = new Path(tablePath, "part=p");
+ Path oldPath = new Path(partitionPath, "data-old.csv");
+ Path targetPath = new Path(partitionPath, "data-new.csv");
+ Path stagingPath =
+ new Path(new Path(tempDir.toUri()), "multipart-staging/upload-in-progress");
+ fileIO.writeFile(oldPath, "old", false);
+ fileIO.writeFile(stagingPath, "staged", false);
+
+ @SuppressWarnings("unchecked")
+ MultiPartUploadStore uploadStore = mock(MultiPartUploadStore.class);
+ doAnswer(
+ invocation -> {
+ fileIO.writeFile(targetPath, "replacement", false);
+ throw new IOException("multipart completion response lost");
+ })
+ .when(uploadStore)
+ .completeMultipartUpload(
+ eq("part=p/data-new.csv"),
+ eq("upload-id"),
+ eq(Collections.singletonList("etag")),
+ eq(1L));
+ doAnswer(
+ invocation -> {
+ fileIO.delete(stagingPath, false);
+ throw new IOException("multipart abort response lost");
+ })
+ .when(uploadStore)
+ .abortMultipartUpload(eq("part=p/data-new.csv"), eq("upload-id"));
+ TwoPhaseOutputStream.Committer committer =
+ new BaseMultiPartUploadCommitter(
+ "upload-id",
+ Collections.singletonList("etag"),
+ "part=p/data-new.csv",
+ 1L,
+ targetPath) {
+ @Override
+ protected MultiPartUploadStore multiPartUploadStore(
+ FileIO ignored, Path ignoredTarget) {
+ return uploadStore;
+ }
+ };
+
+ Throwable failure =
+ catchThrowable(
+ () ->
+ staticPartitionOverwriteCommit(tablePath, fileIO, 1)
+ .commit(
+ Collections.singletonList(
+ new TwoPhaseCommitMessage(committer))));
+
+ assertThat(getRootCause(failure)).hasMessage("multipart completion response lost");
+ assertThat(failureTree(failure))
+ .extracting(Throwable::getMessage)
+ .contains(
+ "Failed to discard multipart upload with ID: upload-id",
+ "multipart abort response lost");
+ assertThat(fileIO.exists(oldPath)).isFalse();
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ assertThat(fileIO.exists(stagingPath)).isFalse();
+ }
+
+ @Test
+ void testOverwritePostPublishCleanFailurePreservesReplacementAndRetriesStagingCleanup()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "post-publish-clean-failure");
+ Path partitionPath = new Path(tablePath, "part=p");
+ Path oldPath = new Path(partitionPath, "data-old.csv");
+ Path targetPath = new Path(partitionPath, "data-new.csv");
+ Path stagingPath = new Path(new Path(tempDir.toUri()), "clean-failure-staging/data.tmp");
+ fileIO.writeFile(oldPath, "old", false);
+ fileIO.writeFile(stagingPath, "staged", false);
+ AtomicBoolean failFirstClean = new AtomicBoolean(true);
+ StagedFileCommitter committer =
+ new StagedFileCommitter(targetPath, stagingPath) {
+ @Override
+ public void commit(FileIO committingFileIO) throws IOException {
+ publish(committingFileIO);
+ }
+
+ @Override
+ public void clean(FileIO cleaningFileIO) throws IOException {
+ if (failFirstClean.compareAndSet(true, false)) {
+ throw new IOException("staging cleanup failed");
+ }
+ super.clean(cleaningFileIO);
+ }
+ };
+
+ assertThatThrownBy(
+ () ->
+ staticPartitionOverwriteCommit(tablePath, fileIO, 1)
+ .commit(
+ Collections.singletonList(
+ new TwoPhaseCommitMessage(committer))))
+ .hasRootCauseMessage("staging cleanup failed");
+
+ assertThat(fileIO.exists(oldPath)).isFalse();
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ assertThat(fileIO.exists(stagingPath)).isFalse();
+ }
+
+ @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);
+
+ 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());
@@ -185,8 +611,6 @@ void testOverwriteKeepsFilesOfConcurrentWritersStagingTrees() throws Exception {
fileIO.writeFile(previousDataFile, "1", false);
// A concurrent job is mid-write in this partition, under a magic committer's tree. Its
// file carries an ordinary data file name; only the directories above it say otherwise.
- // ('_temporary' is the other such tree, but this writer's own clean() still empties it -
- // covered once that is fixed separately.)
Path stagingFile =
new Path(
partitionPath,
@@ -320,8 +744,9 @@ void testOverwritingAPrefixClearsTheDefaultPartitionDirectory() throws Exception
}
@Test
- void testPathNotMatchingThePartitionKeysFails() {
+ void testPathNotMatchingThePartitionKeysFails() throws Exception {
Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "year=2025/day=10/data-1.csv");
// The message names the path and the declared keys, which is what tells a reader that
// 'day' is not where 'month' was expected.
@@ -332,6 +757,7 @@ void testPathNotMatchingThePartitionKeysFails() {
.hasMessageContaining("year=2025/day=10")
.hasMessageContaining("catalog_partition_db.catalog_partition_table")
.hasMessageContaining("[year, month]");
+ assertThat(LocalFileIO.create().exists(targetPath)).isFalse();
}
@Test
@@ -632,27 +1058,1969 @@ void testOverwritingTheWholeTableLeavesADirectoryThatIsNoPartitionOfIt() throws
assertThat(fileIO.exists(new Path(tablePath, "loose.csv"))).isTrue();
}
- /**
- * An overwrite that names no partition: what INSERT OVERWRITE without a PARTITION clause is.
- */
- private FormatTableCommit overwritingCommit(
- Path tableLocation,
- LocalFileIO fileIO,
- boolean dynamicPartitionOverwrite,
- String... partitionKeys) {
- return new FormatTableCommit(
- tableLocation.toString(),
- Arrays.asList(partitionKeys),
- fileIO,
- false,
- PARTITION_DEFAULT_NAME.defaultValue(),
- true,
- Identifier.create("overwrite_db", "overwrite_table"),
- null,
- null,
- null,
- null,
- dynamicPartitionOverwrite);
+ @Test
+ void testLocalSideEffectRunnerPropagatesContextAndRestoresWorkerClassLoader() throws Exception {
+ ExecutorService workers = Executors.newSingleThreadExecutor();
+ try {
+ ClassLoader workerClassLoader =
+ workers.submit(() -> Thread.currentThread().getContextClassLoader())
+ .get(10, TimeUnit.SECONDS);
+ ClassLoader callerClassLoader = new ClassLoader(getClass().getClassLoader()) {};
+ Subject callerSubject = new Subject();
+ AtomicReference seenClassLoader = new AtomicReference<>();
+ AtomicReference seenSubject = new AtomicReference<>();
+
+ ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(callerClassLoader);
+ Subject.doAs(
+ callerSubject,
+ (PrivilegedAction)
+ () -> {
+ try {
+ invokeSideEffectRunner(
+ workers,
+ value -> {
+ seenClassLoader.set(
+ Thread.currentThread()
+ .getContextClassLoader());
+ seenSubject.set(
+ Subject.getSubject(
+ AccessController.getContext()));
+ return Collections.singletonList(value);
+ },
+ Collections.singletonList(1).iterator(),
+ 1,
+ result -> assertThat(result).isOne());
+ return null;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+
+ assertThat(seenClassLoader.get()).isSameAs(callerClassLoader);
+ assertThat(seenSubject.get()).isSameAs(callerSubject);
+ assertThat(
+ workers.submit(() -> Thread.currentThread().getContextClassLoader())
+ .get(10, TimeUnit.SECONDS))
+ .isSameAs(workerClassLoader);
+ } finally {
+ workers.shutdownNow();
+ assertThat(workers.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testLocalSideEffectRunnerCancelsAcceptedTaskWhichHasNotStarted() throws Exception {
+ ExecutorService workers = Executors.newSingleThreadExecutor();
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ ExecutorService withholdingExecutor = mock(ExecutorService.class);
+ CountDownLatch secondTaskAccepted = new CountDownLatch(1);
+ CountDownLatch releaseFirstTask = new CountDownLatch(1);
+ AtomicInteger submissions = new AtomicInteger();
+ AtomicReference withheldTask = new AtomicReference<>();
+ ConcurrentLinkedQueue attemptedInputs = new ConcurrentLinkedQueue<>();
+ RuntimeException workerFailure = new RuntimeException("first side effect failed");
+ doAnswer(
+ invocation -> {
+ Runnable task = invocation.getArgument(0);
+ if (submissions.getAndIncrement() == 0) {
+ workers.execute(task);
+ } else {
+ withheldTask.set(task);
+ secondTaskAccepted.countDown();
+ }
+ return null;
+ })
+ .when(withholdingExecutor)
+ .execute(any(Runnable.class));
+
+ Future result =
+ caller.submit(
+ (Callable)
+ () -> {
+ invokeSideEffectRunner(
+ withholdingExecutor,
+ input -> {
+ attemptedInputs.add(input);
+ if (input == 0) {
+ try {
+ if (!releaseFirstTask.await(
+ 10, TimeUnit.SECONDS)) {
+ throw new RuntimeException(
+ "Timed out waiting to release "
+ + "the first side effect");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ throw workerFailure;
+ }
+ return Collections.emptyList();
+ },
+ Arrays.asList(0, 1).iterator(),
+ 2,
+ ignored -> {});
+ return null;
+ });
+ try {
+ assertThat(secondTaskAccepted.await(10, TimeUnit.SECONDS)).isTrue();
+ releaseFirstTask.countDown();
+ assertThat(getRootCause(awaitFailure(result))).isSameAs(workerFailure);
+ Runnable cancelledTask = withheldTask.getAndSet(null);
+ assertThat(cancelledTask).isNotNull();
+ cancelledTask.run();
+ assertThat(attemptedInputs).containsExactly(0);
+ } finally {
+ releaseFirstTask.countDown();
+ Runnable pendingTask = withheldTask.getAndSet(null);
+ if (pendingTask != null) {
+ pendingTask.run();
+ }
+ caller.shutdownNow();
+ workers.shutdownNow();
+ assertThat(caller.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testLocalSideEffectRunnerSkipsEarlierInputAfterLaterFailure() throws Exception {
+ ExecutorService workers = Executors.newSingleThreadExecutor();
+ ExecutorService reverseOrderExecutor = mock(ExecutorService.class);
+ AtomicReference firstTask = new AtomicReference<>();
+ ConcurrentLinkedQueue attemptedInputs = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue consumedResults = new ConcurrentLinkedQueue<>();
+ RuntimeException workerFailure = new RuntimeException("later side effect failed");
+ doAnswer(
+ invocation -> {
+ Runnable task = invocation.getArgument(0);
+ if (firstTask.compareAndSet(null, task)) {
+ return null;
+ }
+ workers.execute(
+ () -> {
+ task.run();
+ firstTask.get().run();
+ });
+ return null;
+ })
+ .when(reverseOrderExecutor)
+ .execute(any(Runnable.class));
+
+ try {
+ Throwable failure =
+ catchThrowable(
+ () ->
+ invokeSideEffectRunner(
+ reverseOrderExecutor,
+ input -> {
+ attemptedInputs.add(input);
+ if (input == 1) {
+ throw workerFailure;
+ }
+ return Collections.singletonList(input);
+ },
+ Arrays.asList(0, 1).iterator(),
+ 2,
+ consumedResults::add));
+
+ assertThat(failure).isSameAs(workerFailure);
+ assertThat(attemptedInputs).containsExactly(1);
+ assertThat(consumedResults).isEmpty();
+ } finally {
+ workers.shutdownNow();
+ assertThat(workers.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testLocalSideEffectRunnerCompletesWhenContextClassLoaderChangesFail() throws Exception {
+ SecurityException setFailure = new SecurityException("set TCCL denied");
+ SecurityException restoreFailure = new SecurityException("restore TCCL denied");
+ AtomicBoolean denyContextClassLoaderChanges = new AtomicBoolean();
+ AtomicInteger setAttempts = new AtomicInteger();
+ AtomicBoolean processorCalled = new AtomicBoolean();
+ ExecutorService workers =
+ Executors.newSingleThreadExecutor(
+ runnable ->
+ new Thread(runnable, "format-table-denied-tccl-worker") {
+ @Override
+ public void setContextClassLoader(ClassLoader classLoader) {
+ if (denyContextClassLoaderChanges.get()) {
+ throw setAttempts.incrementAndGet() == 1
+ ? setFailure
+ : restoreFailure;
+ }
+ super.setContextClassLoader(classLoader);
+ }
+ });
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ workers.submit(() -> {}).get(10, TimeUnit.SECONDS);
+ denyContextClassLoaderChanges.set(true);
+
+ Future result =
+ caller.submit(
+ () ->
+ catchThrowable(
+ () ->
+ invokeSideEffectRunner(
+ workers,
+ input -> {
+ processorCalled.set(true);
+ return Collections.singletonList(
+ input);
+ },
+ Collections.singletonList(1).iterator(),
+ 1,
+ ignored -> {})));
+ Throwable failure = result.get(10, TimeUnit.SECONDS);
+
+ assertThat(failure).isSameAs(setFailure).hasSuppressedException(restoreFailure);
+ assertThat(setAttempts).hasValue(2);
+ assertThat(processorCalled).isFalse();
+ } finally {
+ denyContextClassLoaderChanges.set(false);
+ caller.shutdownNow();
+ workers.shutdownNow();
+ assertThat(caller.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testLocalSideEffectRunnerPreservesDirectExecutorCallerInterrupt() throws Exception {
+ ExecutorService directExecutor = MoreExecutors.newDirectExecutorService();
+ List consumedResults = new ArrayList<>();
+ Thread.currentThread().interrupt();
+ try {
+ invokeSideEffectRunner(
+ directExecutor,
+ Collections::singletonList,
+ Collections.singletonList(1).iterator(),
+ 1,
+ consumedResults::add);
+
+ assertThat(consumedResults).containsExactly(1);
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ directExecutor.shutdownNow();
+ }
+ }
+
+ @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();
+ 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 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 testOverwritePartialParallelPublishFailurePreservesReplacementAndCleansStaging()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "partial-parallel-publish");
+ Path partitionPath = new Path(tablePath, "part=p");
+ Path oldPath = new Path(partitionPath, "data-old.csv");
+ Path successfulTarget = new Path(partitionPath, "data-success.csv");
+ Path failedTarget = new Path(partitionPath, "data-failed.csv");
+ Path successfulStaging = new Path(new Path(tempDir.toUri()), "partial-staging/success.tmp");
+ Path failedStaging = new Path(new Path(tempDir.toUri()), "partial-staging/failed.tmp");
+ fileIO.writeFile(oldPath, "old", false);
+ fileIO.writeFile(successfulStaging, "staged", false);
+ fileIO.writeFile(failedStaging, "staged", false);
+
+ CountDownLatch bothPublishesStarted = new CountDownLatch(2);
+ CountDownLatch replacementPublished = new CountDownLatch(1);
+ StagedFileCommitter successfulCommitter =
+ new StagedFileCommitter(successfulTarget, successfulStaging) {
+ @Override
+ public void commit(FileIO committingFileIO) throws IOException {
+ bothPublishesStarted.countDown();
+ awaitLatch(bothPublishesStarted, "both overwrite publications to start");
+ publish(committingFileIO);
+ replacementPublished.countDown();
+ }
+ };
+ StagedFileCommitter failingCommitter =
+ new StagedFileCommitter(failedTarget, failedStaging) {
+ @Override
+ public void commit(FileIO committingFileIO) throws IOException {
+ bothPublishesStarted.countDown();
+ awaitLatch(bothPublishesStarted, "both overwrite publications to start");
+ awaitLatch(replacementPublished, "the parallel replacement publication");
+ publish(committingFileIO);
+ throw new IOException("parallel publish failed");
+ }
+ };
+ List messages =
+ Arrays.asList(
+ new TwoPhaseCommitMessage(successfulCommitter),
+ new TwoPhaseCommitMessage(failingCommitter));
+
+ assertThatThrownBy(
+ () -> staticPartitionOverwriteCommit(tablePath, fileIO, 2).commit(messages))
+ .hasRootCauseMessage("parallel publish failed");
+
+ assertThat(fileIO.exists(oldPath)).isFalse();
+ assertThat(fileIO.exists(successfulTarget)).isTrue();
+ assertThat(fileIO.exists(failedTarget)).isTrue();
+ assertThat(fileIO.exists(successfulStaging)).isFalse();
+ assertThat(fileIO.exists(failedStaging)).isFalse();
+ }
+
+ @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();
+ 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.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(getRootCause(awaitFailure(result)))
+ .hasMessage("delete failed at input position 0");
+ assertThat(fileIO.attemptedFiles())
+ .containsExactlyInAnyOrder("data-000.csv", "data-001.csv", "data-new.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(getRootCause(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 = getRootCause(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(getCausalChain(failure.get()))
+ .anyMatch(InterruptedException.class::isInstance);
+ assertThat(interruptRestored).isTrue();
+ assertThat(fileIO.interruptedDeletes()).isZero();
+ 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,
+ /* cleanupThreadNum */ 2,
+ /* publishThreadNum */ 1);
+
+ 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,
+ /* cleanupThreadNum */ 4,
+ /* publishThreadNum */ 1);
+
+ 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(getRootCause(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.
+ */
+ private FormatTableCommit overwritingCommit(
+ Path tableLocation,
+ LocalFileIO fileIO,
+ boolean dynamicPartitionOverwrite,
+ String... partitionKeys) {
+ return new FormatTableCommit(
+ tableLocation.toString(),
+ Arrays.asList(partitionKeys),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ null,
+ null,
+ null,
+ null,
+ dynamicPartitionOverwrite);
+ }
+
+ private FormatTableCommit staticPartitionOverwriteCommit(
+ Path tableLocation, FileIO fileIO, int publishThreadNum) {
+ return new FormatTableCommit(
+ tableLocation.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ true,
+ Identifier.create("overwrite_db", "overwrite_table"),
+ Collections.singletonMap("part", "p"),
+ null,
+ null,
+ null,
+ /* dynamicPartitionOverwrite */ true,
+ /* cleanupThreadNum */ 1,
+ publishThreadNum);
+ }
+
+ 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,
+ /* publishThreadNum */ 1);
+ }
+
+ 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 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 Format Table commit to fail");
+ } catch (ExecutionException expected) {
+ return expected;
+ }
+ }
+
+ private static void invokeSideEffectRunner(
+ ExecutorService executor,
+ Function> processor,
+ Iterator input,
+ int maxConcurrency,
+ Consumer resultConsumer)
+ throws Exception {
+ Method method =
+ FormatTableCommit.class.getDeclaredMethod(
+ "executeSideEffects",
+ ExecutorService.class,
+ Function.class,
+ Iterator.class,
+ int.class,
+ Consumer.class);
+ method.setAccessible(true);
+ try {
+ method.invoke(null, executor, processor, input, maxConcurrency, resultConsumer);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new RuntimeException(cause);
+ }
+ }
+
+ private static void awaitLatch(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 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 abstract static class StagedFileCommitter implements TwoPhaseOutputStream.Committer {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Path targetPath;
+ private final Path stagingPath;
+
+ private StagedFileCommitter(Path targetPath, Path stagingPath) {
+ this.targetPath = targetPath;
+ this.stagingPath = stagingPath;
+ }
+
+ protected void publish(FileIO fileIO) throws IOException {
+ fileIO.writeFile(targetPath, "replacement", false);
+ }
+
+ @Override
+ public void discard(FileIO fileIO) {
+ fileIO.deleteQuietly(targetPath);
+ fileIO.deleteQuietly(stagingPath);
+ }
+
+ @Override
+ public Path targetPath() {
+ return targetPath;
+ }
+
+ @Override
+ public void clean(FileIO fileIO) throws IOException {
+ if (!fileIO.delete(stagingPath, false) && fileIO.exists(stagingPath)) {
+ throw new IOException("Failed to clean staging file " + stagingPath);
+ }
+ }
+ }
+
+ private static class PostRegistrationFailingHiveCatalog extends FileSystemCatalog {
+
+ private final List> registeredPartitions = new ArrayList<>();
+
+ private PostRegistrationFailingHiveCatalog(FileIO fileIO, Path warehouse) {
+ super(fileIO, warehouse);
+ }
+
+ public void createPartitionsUtil(
+ Identifier identifier,
+ List> partitions,
+ boolean partitionOnlyValueInPath) {
+ registeredPartitions.addAll(partitions);
+ throw new RuntimeException("Hive failed after partition registration");
+ }
+ }
+
+ private static class RecordingHiveCatalog extends FileSystemCatalog {
+
+ private final List> registeredPartitions = new ArrayList<>();
+
+ private RecordingHiveCatalog(FileIO fileIO, Path warehouse) {
+ super(fileIO, warehouse);
+ }
+
+ public void createPartitionsUtil(
+ Identifier identifier,
+ List> partitions,
+ boolean partitionOnlyValueInPath) {
+ registeredPartitions.addAll(partitions);
+ }
+ }
+
+ 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;
+ 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 awaitUnexpectedExtraDelete() throws InterruptedException {
+ return unexpectedExtraDelete.await(300, TimeUnit.MILLISECONDS);
+ }
+
+ protected void releaseFirstWave() {
+ releaseFirstWave.countDown();
+ }
+ }
+
+ private static class LazyRootListingFileIO extends ParallelDeleteFileIO {
+
+ 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 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 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 final CountDownLatch deletesStarted;
+ private final CountDownLatch releaseDeletes = new CountDownLatch(1);
+ private final AtomicInteger interruptedDeletes = new AtomicInteger();
+
+ private BlockingDeleteFileIO(int deleteCount) {
+ this.deletesStarted = new CountDownLatch(deleteCount);
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) throws IOException {
+ 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) {
+ interruptedDeletes.incrementAndGet();
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while blocking cleanup delete", e);
+ }
+ }
+
+ private boolean awaitDeletesStarted() throws InterruptedException {
+ return deletesStarted.await(10, TimeUnit.SECONDS);
+ }
+
+ private void releaseDeletes() {
+ releaseDeletes.countDown();
+ }
+
+ private int interruptedDeletes() {
+ return interruptedDeletes.get();
+ }
+ }
+
+ private abstract static class SortedLocalFileIO extends LocalFileIO {
+
+ @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 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 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 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 {
+
+ @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 {
+
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ return false;
+ }
}
private static Map partitionSpec(String year, String month) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
index 029ffe5fac45..07c76b3aba58 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSink.java
@@ -154,9 +154,11 @@ public void close() throws Exception {
if (commitMessages != null && shouldCommit) {
try {
tableCommit.abort(commitMessages);
- } catch (Exception abortFailure) {
+ } catch (Throwable abortFailure) {
// Report the commit failure, not the cleanup that followed it.
- e.addSuppressed(abortFailure);
+ if (abortFailure != e) {
+ e.addSuppressed(abortFailure);
+ }
}
}
throw new RuntimeException(e);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
index 15fb04115930..3c303b986ec7 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/FlinkFormatTableDataStreamSinkTest.java
@@ -27,6 +27,7 @@
import org.apache.paimon.table.format.FormatTableWrite;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
@@ -43,14 +44,19 @@
import org.junit.jupiter.api.io.TempDir;
import java.util.Collections;
+import java.util.List;
import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
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.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -178,6 +184,35 @@ void testFormatTableSinkLineageVertex() throws Exception {
assertThat(vertex.datasets().get(0).name()).isEqualTo("paimon." + table.fullName());
}
+ @Test
+ void testClosePreservesCommitFailureWhenSecondAbortFails() throws Exception {
+ FormatTableWrite tableWrite = mock(FormatTableWrite.class);
+ BatchTableCommit tableCommit = mock(BatchTableCommit.class);
+ CommitMessage message = mock(CommitMessage.class);
+ List messages = Collections.singletonList(message);
+ RuntimeException commitFailure = new RuntimeException("publish response lost");
+ RuntimeException abortFailure = new RuntimeException("multipart upload no longer exists");
+ when(tableWrite.prepareCommit()).thenReturn(messages);
+ doNothing().doThrow(abortFailure).when(tableCommit).abort(messages);
+ doAnswer(
+ invocation -> {
+ tableCommit.abort(messages);
+ throw commitFailure;
+ })
+ .when(tableCommit)
+ .commit(messages);
+
+ SinkWriter writer = createWriter(false, tableWrite, tableCommit);
+
+ Throwable failure = catchThrowable(writer::close);
+
+ assertThat(failure).isInstanceOf(RuntimeException.class);
+ assertThat(failure.getCause()).isSameAs(commitFailure);
+ assertThat(commitFailure.getSuppressed()).containsExactly(abortFailure);
+ verify(tableCommit, times(2)).abort(messages);
+ verify(tableWrite).close();
+ }
+
private SinkWriter createWriter(
boolean overwrite, FormatTableWrite tableWrite, BatchTableCommit tableCommit)
throws Exception {