sundapeng opened a new pull request, #9396:
URL: https://github.com/apache/paimon/pull/9396

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


-- 
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]

Reply via email to