Add a memory bound FileStatisticsCache for the Listing Table

[mkleen]

Which issue does this PR close?

This change introduces a default FileStatisticsCache implementation for the Listing-Table with a size limit, implementing the following steps following #19052 (comment) :

• Add heap size estimation for file statistics and the relevant data types used in caching (This is temporary until Add a crate for HeapSize trait arrow-rs#9138 is resolved)

• Redesign DefaultFileStatisticsCache to use a LruQueue to make it memory-bound following Adds memory-bound DefaultListFilesCache #18855

• Introduce a size limit and use it together with the heap-size to limit the memory usage of the cache

• Move FileStatisticsCache creation into CacheManager, making it session-scoped and shared across statements and tables.

• Disable caching in some of the SQL-logic tests where the change altered the output result, because the cache is now session-scoped and not query-scoped anymore.

• Closes Add a default FileStatisticsCache implementation for the ListingTable #19217

• Closes Add limit to DefaultFileStatisticsCache #19052

Rationale for this change

See above.

What changes are included in this PR?

See above.

Are these changes tested?

Yes.

Are there any user-facing changes?

A new runtime setting datafusion.runtime.file_statistics.cache_limit

[kosiew]

@mkleen

Thanks for working on this.

[mkleen]

@kosiew Thank you for the feedback!

[kosiew]

LGTM

[mkleen]

@kosiew Anything else needed to get this merged? Another approval maybe?

[martin-g]

This won't be accurate.

let a1 = Arc::new(vec![1, 2, 3]);
let a2 = a1.clone();
let a3 = a1.clone();
let a4 = a3.clone();

// this should be true because all `a`s point to the same object in memory 
// but the current implementation does not detect this and counts them separately
assert_eq!(a4.heap_size(), a1.heap_size() + a2.heap_size() + a3.heap_size() + a4.heap_size());

The only solution I imagine is the caller to keep track of the pointer addresses which have been "sized" and ignore any Arc's which point to an address which has been "sized" earlier.

[mkleen]

Good catch! I took this implementation from https://github.com/apache/arrow-rs/blob/main/parquet/src/file/metadata/memory.rs#L97-L102 . I would suggest to also do a follow-up here. We are planing anyway to restructure the whole heap size estimation.

[mkleen]

@martin-g Thanks for this great review! I am on it.

[mkleen]

I’ve incorporated all reviewer sfeedback and addressed the issues raised.

• Fixed the regression affecting cache hits by ensuring the listing table cache is correctly referenced via the CacheManager.
• Scoped the cache by including the table name in the cache key, allowing entries to be removed when a table is dropped.

With these changes, all tests are now passing. However, I’m not fully confident that the overall design is ideal.

I’d appreciate another round of review.

[mkleen]

@martin-g Could you also do another review round please?

[mkleen]

Error message changed from:

DataFusion error: Parquet error: Parquet error: Parquet file has an encrypted footer but decryption properties were not provided

to:

DataFusion error: Parquet error: Parquet error: Failed to fetch metadata for file Users/mkleen/datafusion/datafusion/sqllogictest/test_files/scratch/encrypted_parquet/YkbUUuKrhTO6FhwX_3.parquet: Parquet error: Parquet error: Parquet file has an encrypted footer but decryption properties were not provided

[mkleen]

And sorry to everyone that this took so long. I was very busy and could not find the time.

[kosiew]

@mkleen

Thanks for the update here. I like the direction overall, but I found two issues that should be fixed before this lands. I also left one non-blocking suggestion around cache introspection.

[kosiew]

Nice change overall, but I think there is a bug in the replacement path for memory_used.

When an existing cache entry is overwritten, this path adds the new key size and value size, but if old_value is present it only subtracts old_entry.heap_size(). It does not subtract the old key's heap usage.

That means repeatedly refreshing the same file will slowly inflate memory_used, which can trigger eviction earlier than expected and make the memory bound inaccurate.

Could you update the overwrite path to subtract the previous key contribution as well? Please also add a regression test that does repeated put calls on the same TableScopedPath and verifies memory_used stays stable.

[kosiew]

I think this invalidation logic now only runs when both caches are enabled, because get_list_files_cache() and get_file_statistic_cache() are chained in the same if let.

That creates a cleanup gap. For example, if a session disables the list-files cache but keeps the file-statistics cache enabled, deregistering a table will leave the statistics entries behind. That leaves stale session-scoped state around and can leak memory if the table is re-registered.

Can we make these invalidations independent so each cache is cleaned up whenever it is present?

[kosiew]

Non-blocking thought: now that the file statistics cache key is TableScopedPath, list_entries() -> HashMap<Path, FileStatisticsCacheEntry> seems to flatten away the table dimension again.

That can make observability a bit confusing for helpers like statistics_cache(), especially if two tables point at the same object-store path.

It may be worth exposing TableScopedPath here, or otherwise carrying the table reference through the reported entry, so the debug and introspection surface matches the actual cache semantics.