[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.
|
[P1] Preserve replacement files after overwrite cleanup has committed In Please transition overwrite messages to a state where their targets are not reclaimed after old-data cleanup succeeds and before publishing begins, while still cleaning any staging resources. Regression coverage should include partial parallel publish failure, multipart completion response loss, and post-publish |
|
Fixed in 3cbd4b9:
Local RED-to-GREEN verification: all three new regressions failed before the fix; 76 FormatTable commit/statistics tests, 15 focused common tests, and Spotless now pass. |
|
[P2] Make preserved multipart cleanup safe to repeat
Please either make absent/finished multipart staging cleanup idempotent, or catch the Flink-side abort failure and suppress it onto the original exception, and add a double-abort regression that verifies the original cause is retained. |
|
Suggestion: keep the side-effect draining change focused on FormatTableCommit The shared Please prefer keeping the bounded submission and failure/drain logic local to |
|
Fixed in 4c0e933 for the P2 double-abort report (#9397 (comment)). The Flink sink now keeps the original commit failure as the cause and attaches any repeated abort failure as suppressed, including a self-suppression guard. The regression reproduces the internal abort followed by the Flink-side abort, and verifies the exact primary/suppressed failures, two abort attempts, and writer close. |
|
Addressed in 4c0e933 for the scope suggestion (#9397 (comment)). The shared ThreadPoolUtils and ManifestReadThreadPool changes are restored byte-for-byte to the merge base, and the new shared executor test is removed. The bounded lazy stop/cancel/drain behavior is now private to FormatTableCommit, with focused concurrency, interruption, deterministic failure-order, and TCCL/ACC regression coverage. |
|
+1 |
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.