Repository navigation
Enable the ListFilesCache to be available for partitioned tables #17211
Description
Is your feature request related to a problem or challenge?
When using "high latency" storage (e.g. remote object stores, such as AWS S3) listing objects and collecting object metadata can end up taking a non-trivial amount of time. This has already been noted in #7618 where a simple cache was implemented and partially integrated. Unfortunately, partitioned tables cannot hit the cache path in the current ListingTable implementation. A discrepancy between the performance of flat vs partitioned tables has been previously noted in #9654. Unfortunately, that issue didn't provide much detail on the timing discrepancies, nor did it provide an easy way to reproduce the behavior.
In order to better illustrate this issue, I have added some basic timing statements in the ListingTable which can be reproduced using the diff below:
diff --git a/datafusion/core/src/datasource/listing/table.rs b/datafusion/core/src/datasource/listing/table.rs
index a93654cdf..8d1ddc909 100644
--- a/datafusion/core/src/datasource/listing/table.rs
+++ b/datafusion/core/src/datasource/listing/table.rs
@@ -1342,6 +1342,7 @@ impl ListingTable {
return Ok((vec![], Statistics::new_unknown(&self.file_schema)));
};
// list files (with partitions)
+ let start = datafusion_common::instant::Instant::now();
let file_list = future::try_join_all(self.table_paths.iter().map(|table_path| {
pruned_partition_list(
ctx,
@@ -1353,6 +1354,10 @@ impl ListingTable {
)
}))
.await?;
+ println!(
+ "file_list duration: {}ms",
+ 1000. * start.elapsed().as_secs_f32()
+ );
let meta_fetch_concurrency =
ctx.config_options().execution.meta_fetch_concurrency;
let file_list = stream::iter(file_list).flatten_unordered(meta_fetch_concurrency);
@@ -1370,8 +1375,17 @@ impl ListingTable {
.boxed()
.buffer_unordered(ctx.config_options().execution.meta_fetch_concurrency);
+ let group_start = datafusion_common::instant::Instant::now();
let (file_group, inexact_stats) =
get_files_with_limit(files, limit, self.options.collect_stat).await?;
+ println!(
+ "file_group duration: {}ms",
+ 1000. * group_start.elapsed().as_secs_f32()
+ );
+ println!(
+ "Total duration: {}ms",
+ 1000. * start.elapsed().as_secs_f32()
+ );
let file_groups = file_group.split_files(self.options.target_partitions);
let (mut file_groups, mut stats) = compute_all_files_statistics((note that reproducing these results currently relies on the resolution to /pull/17050)
The discrepancies between the timing for these operations can be seen below. In spite of the fact the queries ultimately end up operating on the same exact subset of the files, the queries issued to the partitioned dataset continue to take longer to complete. In all cases shown here the cumulative duration required to build the file_groups dominates the query times. Each query is issued twice to allow the default caching mechanisms in DataFusion to build cached data, and then operate on that cached data in the 2nd query.
DataFusion CLI v49.0.0
> set datafusion.execution.parquet.cache_metadata = true;
0 row(s) fetched.
Elapsed 0.000 seconds.
> CREATE EXTERNAL TABLE overture_maps
STORED AS PARQUET LOCATION 's3://overturemaps-us-west-2/release/2025-07-23.0/';
0 row(s) fetched.
Elapsed 7.724 seconds.
> select count(*) from overture_maps where type='address';
file_list duration: 317.549ms
file_group duration: 240.83353ms
Total duration: 558.39996ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.563 seconds.
> select count(*) from overture_maps where type='address';
file_list duration: 371.57742ms
file_group duration: 0.034617003ms
Total duration: 371.62094ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.376 seconds.
> CREATE EXTERNAL TABLE overture_maps_flat
STORED AS PARQUET LOCATION 's3://overturemaps-us-west-2/release/2025-07-23.0/theme=addresses/type=address/';
0 row(s) fetched.
Elapsed 1.478 seconds.
> select count(*) from overture_maps_flat;
file_list duration: 0.004915ms
file_group duration: 421.62354ms
Total duration: 421.64505ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.423 seconds.
> select count(*) from overture_maps_flat;
file_list duration: 0.004658ms
file_group duration: 42.788586ms
Total duration: 42.806194ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.044 seconds.The basic layout of the underlying dataset can be seen here. Note that the subset of files accessed for the queries against the partitioned table will be exactly the same as the flat dataset.
$ aws s3 ls s3://overturemaps-us-west-2/release/2025-07-23.0/ --recursive
2025-07-22 14:42:50 1073537247 release/2025-07-23.0/theme=addresses/type=address/part-00000-57f746d8-98a1-4faa-94c2-4f084343ecac-c000.zstd.parquet
2025-07-22 14:43:06 1013189887 release/2025-07-23.0/theme=addresses/type=address/part-00001-57f746d8-98a1-4faa-94c2-4f084343ecac-c000.zstd.parquet
. . .
$ aws s3 ls s3://overturemaps-us-west-2/release/2025-07-23.0/theme=addresses/
PRE type=address/The DefaultListFilesCache can be forced into the mix using the following diff:
diff --git a/datafusion/execution/src/cache/cache_manager.rs b/datafusion/execution/src/cache/cache_manager.rs
index 37f1baa17..1e3faad15 100644
--- a/datafusion/execution/src/cache/cache_manager.rs
+++ b/datafusion/execution/src/cache/cache_manager.rs
@@ -80,6 +80,9 @@ impl CacheManager {
}
if let Some(lc) = &config.list_files_cache {
manager.list_files_cache = Some(Arc::clone(lc))
+ } else {
+ manager.list_files_cache =
+ Some(Arc::new(super::cache_unit::DefaultListFilesCache::default()));
}
if let Some(mc) = &config.file_metadata_cache {
manager.file_metadata_cache = Some(Arc::clone(mc));When the cache is inserted, the performance for the flat table shows significant improvements on repeat queries, however the partitioned table continues to suffer from latencies dominated by listing files.
DataFusion CLI v49.0.0
> set datafusion.execution.parquet.cache_metadata = true;
0 row(s) fetched.
Elapsed 0.000 seconds.
> CREATE EXTERNAL TABLE overture_maps
STORED AS PARQUET LOCATION 's3://overturemaps-us-west-2/release/2025-07-23.0/';
0 row(s) fetched.
Elapsed 9.056 seconds.
> select count(*) from overture_maps where type='address';
file_list duration: 412.9362ms
file_group duration: 272.7048ms
Total duration: 685.6636ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.690 seconds.
> select count(*) from overture_maps where type='address';
file_list duration: 136.47644ms
file_group duration: 0.034272ms
Total duration: 136.51955ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.141 seconds.
> CREATE EXTERNAL TABLE overture_maps_flat
STORED AS PARQUET LOCATION 's3://overturemaps-us-west-2/release/2025-07-23.0/theme=addresses/type=address/';
0 row(s) fetched.
Elapsed 0.727 seconds.
> select count(*) from overture_maps_flat;
file_list duration: 0.0063920002ms
file_group duration: 357.8303ms
Total duration: 357.86032ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.359 seconds.
> select count(*) from overture_maps_flat;
file_list duration: 0.0066520004ms
file_group duration: 0.030025ms
Total duration: 0.047092ms
+-----------+
| count(*) |
+-----------+
| 446544475 |
+-----------+
1 row(s) fetched.
Elapsed 0.001 seconds.Describe the solution you'd like
Ultimately I'd like partitioned datasets to operate with similar performance to flat datasets, and have caching mechanisms available to both. Based on the structure of the existing code, I believe that the code path the ListingTable executes when discovering files and collecting initial metadata should be normalized between partitioned datasets and flat datasets.
During a table scan, both table types first call list_files_for_scan followed by pruned_partition_list
In pruned_partition_list flat tables end up executing list_all_files which is where the ListFilesCache is currently leveraged:
datafusion/datafusion/catalog-listing/src/helpers.rs
Lines 417 to 432 in f3941b2
Partitioned datasets, however, end up making calls that at least partially, and may fully rediscover, the partition and file structure by calling list_partitions
datafusion/datafusion/catalog-listing/src/helpers.rs
Lines 435 to 436 in f3941b2
Note that list_partitions executes a breadth first search of the (potentially prefixed, depending on query filter evaluation) partition structure, and as such the per-query latency penalty it applies scales linearly with the depth of the partition tree.
My Opinion
Executing a breadth first search on the full, or a prefixed subset of, the dataset is likely less performant than simply recursively listing the full, or a prefixed subset of, dataset. The partitions of any given object can be directly derived from the object's path, allowing pruning of objects to continue working as it works right now. Additionally, any caching for listing all of the objects can then apply to both partitioned and flat tables. I suspect there are table structures where the current approach is better, such as a table with many many objects and a shallow partition structure. However, most remote object storage solutions will return many hundreds to thousands of object paths in a single list request, so the tables would need to be quite large in terms of number of objects.
Describe alternatives you've considered
It would probably be possible to apply a ListFilesCache prior to the divergence between the flat and partitioned tables. In this scenario I think care would need to be taken when loading objects into the cache, because partitioned datasets don't necessarily return all the potential objects due to pruning. This would also mean that repeat queries to partitioned datasets could continue to have a high percentage of cache misses depending on user supplied filters.
Additional context
Existing issues mentioned:
- [Cache] Support cache ListFiles result cache in session level #7618
- Partitioned object store lists all files on every query when using hive-partitioned parquet files #9654
Since this relates to the ListingTable and remote dataset:
Activity
alamb commentedon Sep 5, 2025
Interestingly enough I was reviewing #17364 from @xiedeyantu and it seems like there is already code to look for the listing files cache
BlakeOrth commentedon Sep 5, 2025
@alamb Yes, that is the case when the
list_all_filesmethod is called!https://github.com/apache/datafusion/blob/main/datafusion/datasource/src/url.rs#L237
Unfortunately, this method is called conditionally on a table scan, which means the underlying table structure effectively determines whether or not it's possible to utilize the existing cache infrastructure. My proposed solution to this issue is to use
list_all_filesin all cases, so the existing cache infrastructure can be leveraged regardless of the underlying table structure.Here's the code that shows the divergence in logic for when
list_all_filesis used, versus the other methods that are called:datafusion/datafusion/catalog-listing/src/helpers.rs
Lines 417 to 438 in f3941b2
alamb commentedon Sep 6, 2025
This makes a lot of sense to me
alamb commentedon Sep 8, 2025
What I am hoping to do is complete the work on the instrumented object store
Which will then help us verify all the additional optimizations for listing table
BlakeOrth commentedon Sep 8, 2025
Yes, I agree with this approach. Being able to effectively demonstrate a reduction in IO operations is important here, moreso than simply looking at query duration I think. Since remote object storage IO is heavily dependent on an individual user's situation (internet connection quality/speed, physical distance from the datacenter etc.) an increase in performance based strictly on query duration is potentially ambiguous.
alamb commentedon Sep 8, 2025
yes 100%
BlakeOrth commentedon Sep 8, 2025
@alamb Would you feel comfortable gathering some initial data for this effort using the existing code in
even though it's still in draft? If so, I think we'd need this PR merged
CREATE EXTERNAL TABLE#17232so we can effectively interact with partitioned and flat tables, and then I'd be happy to update /pull/17266 and collect some data for us to improve our understanding of the existing request patterns.
alamb commentedon Sep 9, 2025
Yes indeed
Yes sorry I forgot to merge that yesterday and I was trying to avoid causing issues with the datafusion 50 release
BlakeOrth commentedon Sep 9, 2025
Ok, I've done some data collection using the in progress instrumented object store that illustrates some differences in request patterns. I'll share and go over the entire output, including the setup so it can be reproduced. Note that all of the following sections were run in a single
datafusion-clisession, I'm just breaking it up a bit to make it easier to understand and discuss. One item to note is I'm running adebugbuild instead ofrelease-nonltobecause I think we're mostly interested in the IO operations, not necessarily whether or not they're running at optimized speeds.Analysis
Table Setup
This is the test case setup. In order to keep things as fair as possible, we can set up two tables using the same underlying data. One of the tables is a flat table with no partitions, the other lets the ListingTable operate on the partitions defined by the paths. Initially I'm using the summary mode for profiling due to the size of the underlying data, but even here we can already see some discrepancies between the flat and partitioned table emerge. In spite of the same number of
GETrequests, the partitioned table executes 2 additionalLISToperations.First queries
These first queries really begin to show discrepancies. In spite of the
GETrequests being identical between the two queries, the partitioned table executes 22LISToperations compared to the flat table's 1LISToperation. Curiously, theGETrequests appear to be identical between the table creation above and the first queries. It merits additional investigation, but the fact the minimum size for theGETis8, I suspect this is fetching parquet footers and subsequently the metadata. This may point to an opportunity to seed the metadata cache earlier, which could improve performance for the first query to theListingTablein general.Subsequent Queries (cached metadata)
These final queries have been switched over to
tracefor the profiling now that metadata has been cached and the output won't be dominated by over 1500GETrequests. The difference in operation between the two tables is pretty clear here. Curiously, theLISToperations for the partitioned table are being emitted with aduration, which implies it's actually runninglist_with_delimiter()which has allowed the instrumented object store to collect accurate duration data!Paying close attention to the timestamps here we can see the breadth first search listing strategy used on partitioned tables in action. First the root path is listed, blocking until the result is returned:
Then, the first level of the table is listed (this is the partition of
theme). This set of requests are executed in parallel, but block their respective next level in the search. To keep things brief I'll just show the math fortheme=addresses.Finally the last set of directories get listed as their respective upstream list requests complete
Additionally, comparing the first
$(17:53:18.265574453 + 0.179032s) - 17:53:17.991269778 = 0.453336675$
LISTtimestamp to the tailLISTtimestamp and how long it took to complete, we can see that we spent 0.453s executing just listing operations of our total 0.943s query execution time.@alamb Given the above analysis, in the spirit of talking about this in terms of IO reduction, normalizing the IO pattern between partitioned and flat tables in this specific case would eliminate 21
LISTcalls. Unfortunately, it's difficult to talk about this in absolutes because the number ofLISTcalls for partitioned tables depends on the table structure (and actually on the query run as well, but I think this initial analysis is already long enough). This data also suggests that if aListFilesCachewere to be implemented and enabled it could eliminate all the IO for simple queries that can be computed entirely from metadata.alamb commentedon Sep 9, 2025
Yes this seems like a quite compelling reason to:
This sounds great and a worthy goal for sure
BlakeOrth commentedon Sep 9, 2025
If we'd like to move forward with some initial plans or implementations I'm happy to contribute in whatever way I can to help move the effort forward, just let me know what would be the most helpful.
My personal approach would be to start with normalizing the code paths between flat vs partitioned tables so any existing/future caching work can apply to both without additional work or furthering the discrepancies between the two. However, given some list caching infrastructure already exists, I could also see justification for building out a default listing cache first. Based on my current code exploration I don't think the two are very entwined, so it's possible the efforts could proceed as parallel efforts with minimal conflicts.
BlakeOrth commentedon Sep 12, 2025
@alamb I've found the in-review documentation on process updates in:
In an effort to follow the project's processes, considering our existing discussions about this issue, would you be willing to be the "<in_review_adjective> Committer" for this effort?
alamb commentedon Sep 19, 2025
Yes, @BlakeOrth I will plan the the supervising maintainer for this project (specifically #17214)
I am sorry about the delay. I have been focusing on getting DataFusion 50 out, but now that that is done I have more bandwidth to focus on this issue
What I would like to do first is complete
datafusion-cli] Add a way to see what object store requests are made #17207 / POC: datafusion-cli instrumented object store #17266The reason to do this is so we can more easily test the solution and ensure we have completed the issue
Please let me know what you think
BlakeOrth commentedon Sep 19, 2025
@alamb No worries on the delay, I have also been juggling additional items for the past week or so and haven't been able to devote as much time to this effort as it is. I did finally get some time to focus on this and was able to get a functional POC for this issue around the end of the day yesterday. I think the approach is looking quite promising. If you have the time and would be interested in looking at some in-progress code (complete with in-progress Rust things like unwrapping errors and the like) I can open a draft PR whenever.
I agree completing #17266 first is important since it's our validation mechanism for this effort! I know there's still some work to do on the PR before it's actually ready (or maybe needs to be split up, I'm not sure we've really decided on a specific path forward), but let me know how we'd like to proceed there and that can be my primary focus with regards to this effort.
BlakeOrth commentedon Oct 17, 2025
@alamb Based on the POC work we're only dealing with Get and List requests, and specifically focused on List requests, for this issue. I believe the merge of this PR:
completes enough of the implementation of #17207 that we can accurately evaluate any changes for this issue. I have a PR that is surprisingly small (I'm as surprised as anyone else) that implements my proposed solution of normalizing the code between partitioned and flat tables, and as such reduces the number of object store requests for partitioned tables. Are we ready for a PR against this effort?
alamb commentedon Oct 17, 2025
Yes I think so! Thank you
Adds Partitioned CSV test to object store access tests (#18370)
Adds Partitioned CSV test to object store access tests (apache#18370)
Normalize partitioned and flat object listing (#18146)
Adds Partitioned CSV test to object store access tests (apache#18370)
Normalize partitioned and flat object listing (apache#18146)
BlakeOrth commentedon Nov 17, 2025
@alamb I believe with the acceptance/merge of:
This issue is (as written) technically complete because users can now implement and leverage a
ListFilesCachefor partitioned tables. (Pedantically I guess this was also true with the previous PR that was merged, but now it's more back to parity with the previous implementation). I think you and I are both excited about getting a default implementation for aListFilesCacheas the next step. Would you prefer we close this issue and open a new one for the cache implementation, or should we just piggy-back off this issue and the associated epic? (I'm happy to facilitate either direction)Adds prefix filtering for table URLs (#18780)
alamb commentedon Nov 18, 2025
I am always a fan of claiming success / closing tickets and filing new ones for additional effort
Thus I would suggest we file a new ticket for adding
select * from metadata_cache();)BlakeOrth commentedon Nov 19, 2025
@alamb perfect!
Closing this as it's been solved by:
With the next follow-on issue here:
ListFilesCacheimplementation for theListingTable#18827alamb commentedon Nov 19, 2025
🎉
Adds prefix filtering for table URLs (apache#18780)