[api][core] Parallelize format table overwrite commit - #9397
Conversation
72e9018 to
5f10ffe
Compare
sundapeng
changed the title
sundapeng
changed the title
A Format Table INSERT OVERWRITE deletes the old data files it replaces one at a time on the driver. On a table holding thousands of files that is thousands of synchronous round trips after the last task has finished, with nothing else running. Hand those deletes to a bounded runner. Only a catalog-managed partitioned Format Table uses it; filesystem-discovered, unpartitioned, truncate and ordinary Paimon table paths stay serial, and format-table.commit.cleanup-thread-num = 1 opts out. The runner is the bounded batch execution in ThreadPoolUtils, extended here for callers that change stored state: Let it take an iterator, so a caller that discovers its work by listing storage does not have to list all of it before the first task can start, and refill the window as a slot frees rather than a batch at a time. An overwrite that replaces the whole table then holds one partition rather than every file the table has. Add a variant whose close waits for a task that has already started instead of interrupting it. Interrupting a delete halfway leaves the caller unable to say whether it took effect, so a caller that changes stored state cannot let close cancel what it has already handed out. Give a worker its thread's classloader back, and clear the interrupt a cancelled task may leave behind, so that neither reaches whatever the shared pool runs next. Everything accepted is waited for before the commit fails, and failures keep their input order.
Publishing the files a Format Table commit wrote is the other half of the same problem: one synchronous multipart completion per file, on the driver, after cleanup has finished. Cleanup concurrency alone leaves that untouched. Publish through the same bounded runner, under its own format-table.commit.publish-thread-num with the same scope and the same opt out. The caller still publishes directly when there is one file or one thread, so a small commit gains nothing and risks nothing. Statistics, staging clean up and the catalog update stay on the caller, after every publish has been waited for. Nothing reads a partition's files while another thread may still be adding to them. Roll a failed commit back file by file. A publication that a concurrent commit can fail part way through leaves files behind that no partition should hold, and discarding the staging output does not remove one that was already published. Every target belongs to this write attempt, so removing it is safe even when a completion took effect but its response was lost. Carry the caller's access control context into the workers, so publication runs with the same permissions whether or not it is handed to the pool.
6d3ef33 to
29d8558
Compare
sundapeng
changed the title
…lsTest testCloseCancelsQueuedTasksAndWaitsUninterruptibly reads runQueuedTaskOnInterrupt as soon as workerInterrupted is released. The hook that clears the flag does so after super.interrupt(), and it is super.interrupt() that wakes the worker and releases that latch. The read and the clear are therefore unordered: let the closing thread lose the CPU between the two and the assertion sees a flag the hook has not consumed yet. That is a fixture defect, not a product one. close still cancels every unstarted task before it interrupts a running one, and the invariants the test exists for, executions == 2 and thirdExecuted == false, are asserted separately. Have the hook count down a latch once it is done, and await that latch before reading the flag. The test still kills the bug it was written for: reversing the cancel and interrupt loops in close fails it on executions. It reproduced on every run pinned to a single CPU and on none of the unpinned ones, which is why a low core count runner saw it and a development machine did not.
| } | ||
| // 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. | ||
| markPublishedTargetsToPreserveOnAbort(messages); |
There was a problem hiding this comment.
[P1] Do not preserve published append files for every metadata failure
This flag is set before createPartitions, so even a definite pre-mutation failure (for example, loading the catalog or an authorization rejection) makes abort skip deletion for every published target. For an append to an already registered partition, scans list files directly from the partition directory, so the failed write remains visible and a retry can duplicate the rows. The new partial-batch test also preserves targets from the batch that throws before applying anything. Please track outcomes per request or partition: roll back definitely unapplied targets, and reconcile or retry idempotently only when the metadata outcome is genuinely indeterminate.
| classLoader, | ||
| accessControlContext, | ||
| this::stopSubmission); | ||
| executor.execute(task); |
There was a problem hiding this comment.
[P2] Avoid calling a potentially blocking executor while holding submissionLock
SemaphoredDelegatingExecutor.execute blocks while acquiring a permit. With permitCount = 1 and queueSize = 2, the second submission holds submissionLock while waiting for the only permit; if the first task fails, stopSubmission waits for the same lock, so that task cannot return and release its permit. The iterator and close path then deadlock. Please make failure signaling lock-free or move execute outside any lock required by workers, and add a timed permit-1/window-2 regression test.
| failure = taskFailure; | ||
| stopSubmission.run(); | ||
| } finally { | ||
| currentThread.setContextClassLoader(originalClassLoader); |
There was a problem hiding this comment.
[P2] Always publish task completion if TCCL handling fails
Under a JDK 8 SecurityManager that denies RuntimePermission("setContextClassLoader"), line 476 records the first SecurityException, but this restore throws again before state = FINISHED and completion.countDown(), leaving result() and close() blocked forever. getContextClassLoader() at line 474 is also outside the protected region. Please put state publication and latch countdown in an outermost finally, handle TCCL restore failures separately, and add a timed denying-SecurityManager regression test.
2 participants
Purpose
A Format Table
INSERT OVERWRITEdeletes each old data file and completes each new one in its ownround trip on the driver. On a table holding thousands of files that dominates the statement: in one
profile the commit ran more than twice as long as every Spark task put together, leaving the statement
around three times slower than the writer the table had been migrated from. Deleting the old files was
the larger half of it, publishing the new ones the rest.
Both loops now run with bounded concurrency under
format-table.commit.cleanup-thread-numandformat-table.commit.publish-thread-num, both defaulting to 64.Cleanup still finishes before publication starts, and a commit that fails now removes the files it had
already published.
Tests
FormatTableCommitTestcovers the bound, lazy listing, stop-and-drain on the first failure, failureordering, per-file rollback and every path that stays serial.
ThreadPoolUtilsTest,CoreOptionsTestand
FormatTableCommitStatisticsTestcover the runner, the options and what a commit reports.API and Format
ThreadPoolUtilsgains one method,sequentialBatchedExecuteAwaitRunningTasksOnClose; its twoexisting signatures are unchanged. Its shared iterator now refills as a slot frees rather than a batch
at a time, which the manifest read path gets as well. No file format change.
Documentation
The two options in
docs/generated/core_configuration.html.