[core] Parallelize per-partition file listing for catalog-managed format tables#8845
Open
XiaoHongbo-Hope wants to merge 15 commits into
Open
[core] Parallelize per-partition file listing for catalog-managed format tables#8845XiaoHongbo-Hope wants to merge 15 commits into
XiaoHongbo-Hope wants to merge 15 commits into
Conversation
…mat tables Format table split planning lists every partition's files serially, which is slow for internal (catalog-managed) tables with many partitions (thousands of object store LIST calls on the coordinator). Extract the per-partition listing into a PartitionSplitPlanner strategy with two implementations selected by the existing internal/external discriminator (partitionManager != null): - SequentialSplitPlanner (external): unchanged serial path; a missing partition directory is rethrown. - ParallelSplitPlanner (internal): bounded, order-preserving fan-out via ManifestReadThreadPool; a registered partition without a directory reads as empty (matching Hive), any other listing failure fails the whole scan. Split order and per-partition sort/bin-packing are unchanged, so output is identical to the serial path. Parallelism is bounded by a new option format-table.scan.list-parallelism (default #CPU); external tables are untouched.
sundapeng
reviewed
Jul 24, 2026
| } | ||
|
|
||
| /** Lists partition files and builds splits; per-partition sort and bin-packing are shared. */ | ||
| private interface PartitionSplitPlanner { |
Member
There was a problem hiding this comment.
better to remove this interface
Contributor
Author
There was a problem hiding this comment.
better to remove this interface
Removed
Inline the internal/external branch directly in splits(): a private listPartitionFilesInParallel(...) for internal (catalog-managed) tables and the serial loop for external tables. Behavior is unchanged.
XiaoHongbo-Hope
marked this pull request as ready for review
July 25, 2026 01:34
A non-positive format-table.scan.list-parallelism would create a SemaphoredDelegatingExecutor with zero/negative permits, making the listing acquire() block forever. Clamp to a minimum of 1 so a misconfigured value degrades to serial listing instead of hanging.
JingsongLi
reviewed
Jul 25, 2026
| BinaryRow partitionRow = toPartitionRow(partitionSpec); | ||
| if (partitionFilter == null || partitionFilter.test(partitionRow)) { | ||
| try { | ||
| if (partitionManager != null) { |
Contributor
There was a problem hiding this comment.
I think we could refactor the code here; the partitionManager should be renamed to SplitEnumerator so that it directly generates the corresponding Split.
Contributor
Author
There was a problem hiding this comment.
I think we could refactor the code here; the
partitionManagershould be renamed toSplitEnumeratorso that it directly generates the correspondingSplit.
Done
Move format-table partition discovery and split generation into a package-private FormatTableSplitEnumerator. Keep FormatTablePartitionManager focused on catalog metadata, parallelize catalog partition file listing, and preserve serial filesystem discovery.
JingsongLi
reviewed
Jul 25, 2026
| import static org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths; | ||
|
|
||
| /** Enumerates {@link FormatDataSplit}s for a {@link FormatTable}. */ | ||
| final class FormatTableSplitEnumerator { |
Contributor
There was a problem hiding this comment.
I mean to create a SplitEnumerator, and to two SplitEnumerators for catalog and filesystem. This can reduce many if else.
Per review, replace the partitionManager-based branching inside the single FormatTableSplitEnumerator with two implementations chosen once at construction by FormatTableSplitEnumerator.create: CatalogFormatTableSplitEnumerator (catalog-managed partitions, parallel listing) and FileSystemFormatTableSplitEnumerator (filesystem discovery, serial). enumerate/findPartitions/listPartitionEntries are now polymorphic, so the per-call 'partitionManager != null' if/else is gone.
The dedicated createCachedThreadPool(1000) sets corePoolSize == maximumPoolSize == 1000, so it spawns a new thread for every submitted task until the pool holds 1000 threads rather than reusing idle workers, even though per-scan concurrency is bounded by the semaphore. Listing ~1000 partitions on the coordinator could thus create ~1000 threads and risk 'unable to create native thread'/OOM. Delegate to ManifestReadThreadPool.randomlyExecuteSequentialReturn, the shared IO pool Paimon already uses for scans (sized to availableProcessors, grown only when a larger parallelism is explicitly requested and capped by the format-table.scan.list-parallelism permit).
…filesystem context Address review of the parallel listing: - Size the dedicated pool at 128 (was 1000) and clamp format-table.scan.list-parallelism to [1, 128], so the pool reuses workers and can no longer grow to ~1000 threads or be pushed there by a large config value. Reverts the previous fix that had mutated the shared ManifestReadThreadPool. - Establish the filesystem on the caller thread (fileIO.exists on the table location) before dispatching, so the listing workers reuse a filesystem created under the caller's security context instead of lazily creating and caching it under a shared worker's context. This matches the FileStore scan path, which reads the snapshot on the caller thread before its parallel manifest reads.
…the caller thread
…ng workers 128 was an arbitrary cap; revert to the intended model - per-scan concurrency 64 (format-table.scan.list-parallelism) over a shared pool capped at 1000 (Trino's hive.split-loader-concurrency / hive.max-split-iterator-threads). Build the pool as a bounded cached pool (core 0, SynchronousQueue) so it reuses idle workers and grows only on demand, instead of createCachedThreadPool whose corePoolSize == max spawns a new thread per task up to the cap. CallerRunsPolicy applies back pressure once the cap is hit.
…pls per review Rename FormatTableSplitEnumerator -> SplitEnumerator, CatalogFormatTableSplitEnumerator -> CatalogSplitEnumerator, and FileSystemFormatTableSplitEnumerator -> FileSystemSplitEnumerator.
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
3 participants
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.
No description provided.