Skip to content

[core][common][oss] Parallelize format table overwrite commit and batch delete old files - #9396

Closed
sundapeng wants to merge 4 commits into
apache:masterfrom
sundapeng:upstream/format-table-overwrite-commit-concurrency
Closed

[core][common][oss] Parallelize format table overwrite commit and batch delete old files#9396
sundapeng wants to merge 4 commits into
apache:masterfrom
sundapeng:upstream/format-table-overwrite-commit-concurrency

Conversation

@sundapeng

Copy link
Copy Markdown
Member

Purpose

INSERT OVERWRITE on a catalog-managed Format Table finishes its Spark stages and then spends the
rest of the statement in the driver. FormatTableCommit deletes the old data files of the target
partitions 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 CompleteUploadPart requests over roughly 242s. The two spans match the cleanup loop and
the publish loop in FormatTableCommit. The end-to-end effect of this change on that workload has not
been 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) lists
partition 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 FormatTableCommit
constructor all keep the serial path they have today, and TRUNCATE TABLE and TRUNCATE PARTITION
stay serial as well.

Four commits:

  1. format-table.commit.cleanup-thread-num (1 to 64, default 64) bounds how many old data files a
    commit 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.

  2. 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.

  3. FileIO.batchFileDeleter(Path) returns an optional batch-delete capability for the provider serving
    that path. The default is empty and performs no storage access, and PluginFileIO, ResolvingFileIO,
    CachingFileIO and RESTTokenFileIO forward it. OSSFileIO implements it over deleteObjects with at
    most 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.

  4. 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

CoreOptionsTest for the two options, their bounds and their defaults.

FormatTableCommitTest for cleanup: the concurrency a catalog-managed builder actually reaches and the
64 ceiling, the serial path for explicit 1, filesystem-discovered, unpartitioned and old-constructor
tables, 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.

FormatTableCommitPublishTest for publication: order within a partition and overlap across
partitions, 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.

FormatTableCommitBatchDeleteTest for the consumer: the gates that select the batch path, batching to
the 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.

FileIOBatchDeleteContractTest and FileIOBatchDeleteForwardingTest for the capability and the four
wrappers, 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.

OSSFileIOBatchDeleteTest for the provider: batch size, bucket and authority validation before any
client is obtained, and verbose response checking.

API and Format

FileIO gains one default method that returns empty and touches no storage, so existing
implementations keep working unchanged. BatchFileDeleter and BatchDeleteResult are new public
types. No file format change. The public FormatTableCommit constructor is unchanged and stays
serial; the two thread counts are passed through a package-private constructor that
FormatBatchWriteBuilder uses.

Documentation

docs/generated/core_configuration.html is regenerated for the two new options.

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.
@sundapeng

Copy link
Copy Markdown
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.

@sundapeng sundapeng closed this Aug 26, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant