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.

[kosiew]

@mkleen

Thanks for addressing the earlier feedback.

Most of the issues look resolved and the changes are in a much better place now. There are just a couple of remaining concerns that would be good to fix before merging.

[kosiew]

It looks like DFHeapSize for Arc<T> is still double counting shared allocations. Each clone of the same Arc ends up reporting the full size of the underlying data.

This was mentioned earlier, but it is still an issue. Any structure holding multiple clones of the same Arc will overestimate memory usage.

Since this trait is now used for cache accounting, this could lead to premature evictions and confusing metrics. It would be great to adjust this so shared allocations are only counted once.

[kosiew]

This change subtly alters the behavior of with_file_statistics_cache(None). It no longer disables the cache if file_statistics_cache_limit > 0, because try_new will create a default cache in that case.

Previously, passing None was enough to disable the cache. Now callers also need to know to set the limit to 0, which is not obvious.

This feels like a silent behavior change and it also contradicts the current builder documentation. It would be good to either restore the original behavior or make the contract explicit in both code and docs.