JingsongLi commented on code in PR #9397:
URL: https://github.com/apache/paimon/pull/9397#discussion_r3879944765
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -506,40 +717,339 @@ private List<Pair<LinkedHashMap<String, String>, Path>>
partitionsInTheFileSyste
*/
private Set<Path> deletePreviousDataFile(Path partitionPath, int
partitionLevels)
throws IOException {
+ return deletePreviousDataFiles(
+ Collections.singletonList(partitionPath), partitionLevels, 1);
+ }
+
+ private Set<Path> deletePreviousDataFiles(
+ List<Path> partitionPaths, int partitionLevels, int threadNum)
throws IOException {
+ Iterator<FileStatus> dataFiles = previousDataFiles(partitionPaths,
partitionLevels);
Set<Path> 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.
+ *
+ * <p>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 <I, O> void executeSideEffects(
+ ExecutorService executor,
+ Function<I, List<O>> processor,
+ Iterator<I> input,
+ int maxConcurrency,
+ Consumer<O> resultConsumer) {
+ AtomicBoolean submissionStopped = new AtomicBoolean();
+ ArrayDeque<SideEffectTask<I, O>> 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<I, O> 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<I, O> 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<I, O> 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<I, O> 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<I, O> 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<I, List<O>> 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<O> result;
+ private Throwable failure;
+ private volatile boolean failureReported;
+
+ private SideEffectTask(
+ Function<I, List<O>> 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<List<O>>) () ->
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<O> 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));
Review Comment:
[P2] Avoid self-suppression while unwrapping shared I/O failures
If two concurrently running publish/delete tasks wrap the same IOException
instance (for example, a shared failed future or a FileIO/committer that
rethrows a cached failure), the runner makes the second UncheckedIOException
suppressed on the first. Both recursive calls here then return the same cause,
so this executes cause.addSuppressed(cause), which throws
IllegalArgumentException: Self-suppression not permitted and replaces the real
storage error. This is reachable because already-running tasks are deliberately
drained after the first failure. Please identity-check the unwrapped suppressed
failure (or aggregate via firstOrSuppressed) and add a regression with two
wrappers sharing one cause.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
