[core][common][oss] Parallelize format table overwrite commit and batch delete old files - #9396
Closed
sundapeng wants to merge 4 commits into
Closed
Conversation
Add bounded cleanup concurrency for catalog-managed partitioned Format Tables while preserving serial behavior for other table paths. Drain accepted deletes before publishing or aborting and keep failure ordering deterministic.
Publish the files of a commit with bounded concurrency for catalog-managed partitioned Format Tables, grouped by target partition so a partition keeps its input order while different partitions overlap. Compute statistics, clean staging and update the catalog on the calling thread once every publication has completed.
Let a FileIO expose a batch deleter for the provider serving a path, forwarded by the plugin, resolving, caching and REST token wrappers. A provider that has started a batch request never falls back to individual deletes, so a partial success cannot be hidden by deleting the remaining files one by one. OSS deletes at most 1000 keys of one bucket per request and verifies the per-object result.
Use the batch delete capability, when the provider offers one, to clean the old files of the partitions a dynamic overwrite writes. Whole-table and static-prefix overwrite keep individual deletes, because they report which files this commit removed and a deleted-or-not-found batch result cannot answer that.
Member
Author
|
Closing this in favour of one PR per change, which is easier to review:
The fourth change, which makes a dynamic overwrite consume that capability, depends on #9397 and #9398 and I will send it once they land. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
INSERT OVERWRITEon a catalog-managed Format Table finishes its Spark stages and then spends therest of the statement in the driver.
FormatTableCommitdeletes the old data files of the targetpartitions one at a time, and then publishes the newly written files one at a time, so a commit that
replaces N files pays N sequential round trips to object storage, twice.
A profile of one production statement: the statement took 857.9s, the last Spark task finished 605.8s
before it returned, and no Spark stage ran in that window. In the same window object storage recorded
one driver client issuing 12,700 metadata, 9,384 PUT and 9,384 DELETE requests over roughly 363s, and
then 9,384
CompleteUploadPartrequests over roughly 242s. The two spans match the cleanup loop andthe publish loop in
FormatTableCommit. The end-to-end effect of this change on that workload has notbeen measured yet, so the numbers above are the profile that motivated the work, not a claimed speedup.
Reading these tables is already parallel:
format-table.scan.list-parallelism(default 64) listspartition files concurrently during split planning. This gives the write side the same treatment,
within the same scope. Only an internal Format Table whose partitions the catalog manages is affected,
that is
partitionManager != null && !partitionKeys.isEmpty(). Filesystem-discovered format tables,unpartitioned format tables, ordinary Paimon tables and the existing public
FormatTableCommitconstructor all keep the serial path they have today, and
TRUNCATE TABLEandTRUNCATE PARTITIONstay serial as well.
Four commits:
format-table.commit.cleanup-thread-num(1 to 64, default 64) bounds how many old data files acommit deletes at once. One task per file and a sliding window of at most 64 in flight per commit, on
one FIFO executor shared by concurrent commits, so a large commit cannot lock a small one out. The
first failure stops handing out work, every accepted task is drained before the commit fails, and the
primary exception is picked by input order with the rest attached as suppressed. Cleanup completes
before anything is published.
format-table.commit.publish-thread-num(1 to 64, default 64) does the same for publication,grouped by target partition: one file at a time within a partition so its input order is preserved,
different partitions in parallel. Workers only publish and return their result; partition statistics,
staging cleanup and the catalog update stay on the calling thread after the barrier.
FileIO.batchFileDeleter(Path)returns an optional batch-delete capability for the provider servingthat path. The default is empty and performs no storage access, and
PluginFileIO,ResolvingFileIO,CachingFileIOandRESTTokenFileIOforward it.OSSFileIOimplements it overdeleteObjectswith atmost 1000 keys of one bucket per request and verifies the per-object result. The contract is strict:
an absent capability is the only signal that a caller may use individual deletes, and once a batch
request has started, a failure or an incomplete response fails the caller rather than deleting the
remaining files one by one, which would hide a partial success. A caller that retries must resend the
same complete batch.
Dynamic overwrite consumes the capability: when the partitions to replace are the ones the commit
wrote and the provider offers a batch deleter, cleanup deletes the old files in full batches instead
of 64-way single deletes. Whole-table and static-prefix overwrite keep individual deletes, because
those paths report which files this commit removed and a deleted-or-not-found batch result cannot
answer that.
Jindo is the other provider that matters for this workload and it is deliberately not here. 6.9.1 has
no batch API, and while 6.10.7 has one it is disabled by default with single failover on, so it needs
its own change backed by runtime evidence.
Tests
CoreOptionsTestfor the two options, their bounds and their defaults.FormatTableCommitTestfor cleanup: the concurrency a catalog-managed builder actually reaches and the64 ceiling, the serial path for explicit
1, filesystem-discovered, unpartitioned and old-constructortables, the cleanup to publish barrier, first failure stopping submission and draining accepted work,
the primary exception chosen by input position with the others suppressed, interrupt flag restoration,
abort failure not masking the original one, fairness between a large and a small commit on the shared
executor, one concurrency window across partition roots, and no partition root listed before the first
deletes complete.
FormatTableCommitPublishTestfor publication: order within a partition and overlap acrosspartitions, the full barrier before statistics, staging cleanup and the catalog update, first failure
draining in-flight publications before abort, caller interrupt, TCCL of a reused worker, executor
rejection, and the single-partition fast path.
FormatTableCommitBatchDeleteTestfor the consumer: the gates that select the batch path, batching tothe provider maximum, exact-set and order verification of the result, no fallback to single deletes
once a request has started, and the paths that stay on single deletes.
FileIOBatchDeleteContractTestandFileIOBatchDeleteForwardingTestfor the capability and the fourwrappers, including discovery that performs no I/O, a plugin invocation restoring the context
classloader, mixed authorities under
ResolvingFileIO, and rediscovery after a REST token refresh.OSSFileIOBatchDeleteTestfor the provider: batch size, bucket and authority validation before anyclient is obtained, and verbose response checking.
API and Format
FileIOgains one default method that returns empty and touches no storage, so existingimplementations keep working unchanged.
BatchFileDeleterandBatchDeleteResultare new publictypes. No file format change. The public
FormatTableCommitconstructor is unchanged and staysserial; the two thread counts are passed through a package-private constructor that
FormatBatchWriteBuilderuses.Documentation
docs/generated/core_configuration.htmlis regenerated for the two new options.