Parallelize `list_files_for_scan`

[Dandandan]

Is your feature request related to a problem or challenge?

Cold runs are slow: due to reading / processing everything on one thread.

Currently list_files_for_scan runs in a single thread.

Describe the solution you'd like

We should be able to use all threads.

Describe alternatives you've considered

No response

Additional context

No response

[Tushar7012]

Hi @Dandandan

I'd like to take on this issue! I've been actively contributing to DataFusion and am excited to work on this performance optimization that's part of the ClickBench EPIC (#18489).

I've analyzed the current implementation and here's my understanding and proposed approach:

Current Bottleneck:
The list_files_for_scan method in catalog-listing/src/table.rs currently uses try_join_all on pruned_partition_list calls sequentially, and while there's buffer_unordered for statistics collection (controlled by meta_fetch_concurrency), the core file listing and partition evaluation can be parallelized further.

Proposed Solution:

1. Parallelize pruned_partition_list calls - When multiple table_paths exist, process them concurrently using FuturesUnordered or similar
2. Leverage existing meta_fetch_concurrency config - Reuse this setting (default=32) to control parallelism, maintaining consistency with existing patterns
3. Stream-based parallelism - Use flatten_unordered more aggressively in the file discovery phase, not just for metadata fetching
4. Consider partition filter evaluation - The filter_partitions step in helpers.rs could benefit from parallel evaluation when dealing with many partitioned files

Testing Plan:

• Benchmark with ClickBench queries to measure cold-start improvement
• Test with partitioned datasets (like TPC-H, TPC-DS) where list_files_for_scan is called frequently
• Ensure no regression in single-threaded scenarios

I'm familiar with the async patterns used in DataFusion and the object_store crate. Would be happy to discuss the approach before implementation if you'd prefer.

Could you please assign this to me?

[Dandandan]

I think the proposal is a bit too complex / unneeded, I think just spawning within the function should give the optimization.

Also see:

#19969

[Tushar7012]

Thanks for the assignment and the pointer to PR #19969!

I've reviewed your infer_schema parallelization approach and understand the pattern now - using tokio spawning directly within the function rather than a multi-layered solution.

For list_files_for_scan, I'll follow the same approach:

1. Spawn parallel tasks within the function to list files concurrently
2. Use JoinSet or similar to collect results
3. Keep the existing API surface unchanged

I'll also note that benchmarks may not capture the improvement well since the gain is primarily on cold start / first query (before caching kicks in).

I'll start working on a draft PR following this pattern. Let me know if there's anything specific I should be aware of for the list_files_for_scan case!

[alamb]

FYI I think @BlakeOrth has worked on this before and may have some thoughts

[Dandandan]

Related: #20005

[Tushar7012]

Thanks for the guidance. I’ll implement the simpler approach: spawn tasks directly inside list_files_for_scan (following #19969), collect with JoinSet, and keep the API unchanged. I’ll also respect the existing meta_fetch_concurrency for throttling. I’ll open a draft PR shortly and tag @BlakeOrth for any thoughts

[BlakeOrth]

I'm sure there are performance gains to be had during the file listing phase of a cold query. I'm skeptical (read: actual evidence of performance improvement should be required here) that there's much performance to be had in terms of parallelizing calls for actually listing of objects backing a table. The issue with parallelizing calls for listing itself is the underlying object_store machinery is inherently sequential. Even if you invoke it in a parallel manner, the underlying implementations have to make sequential calls (how do you parallelize the discovery of a set that doesn't guarantee deterministic ordering of results that's also of unknown size?).

@Dandandan I'm not familiar with the output of the benchmarking tool you're using here. Is the single thread that's operating during the listing operations actually exhibiting CPU bottlenecking, or is DataFusion spending time waiting on IO here?

[Dandandan]

I am using samply as tool which does sample-based CPU profiling https://github.com/apache/datafusion/blob/6524d91938d2ea6c764edd1a2bc3fd4c98cfcc9c/docs/source/library-user-guide/profiling.md#profiling-using-samply-cross-platform-profiler

The example of the screenshot is just one query (5) of clickbench_partitioned (which has 100 files).

I agree there is probably not much to be added re: listing of objects, but the heavy part (when running it locally agains a number of files) is actually the CPU part: deserializing/converting/merging/... Parquet metadata + statistics, which is also done in list_files_for_scan.
The file IO itself is pretty quick of course when running it locally.

Moving this to use a number of threads should at least spread the work.

[Tushar7012]

Thanks for the insight @Dandandan!

I've implemented the parallelization in list_files_for_scan using tokio::task::JoinSet.

Each pruned_partition_list call (which handles the listing and the CPU-intensive metadata/statistics processing you mentioned) is now spawned as a separate Tokio task. This allows us to spread the deserialization and processing work across multiple threads, which should directly address the bottleneck observed in the profile significantly improving performance for tables with many partitions or files.

I've also refactored ListingTableUrl to pass ConfigOptions and Arc<RuntimeEnv> explicitly, ensuring Send compliance for the spawned tasks.

[BlakeOrth]

deserializing/converting/merging/... Parquet metadata + statistics

This portion, which is operating on the results of the actual IO operations, in parallel also makes sense to me as a first step towards additional performance in this area.