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

• 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.

[alamb]

@nuno-faria and @kosiew -- shall we merge this PR? Or is it waiting on anything else?

[mkleen]

@nuno-faria and @kosiew -- shall we merge this PR? Or is it waiting on anything else?

@alamb No, there is a blocker. While merging in the main branch i noticed one failing test:

at /Users/mkleen/Test/datafusion/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:700


2. query result mismatch:
[SQL] EXPLAIN SELECT b FROM (
  SELECT a, b FROM t_union_mem
  UNION ALL
  SELECT a, b FROM t_union_parquet
) WHERE b > 2;
[Diff] (-expected|+actual)
    logical_plan
    01)Projection: b
    02)--Filter: b > Int64(2)
    03)----Union
    04)------Projection: t_union_mem.a, t_union_mem.b
    05)--------TableScan: t_union_mem
    06)------Projection: t_union_parquet.a, t_union_parquet.b
-   07)--------TableScan: t_union_parquet
-   physical_plan
-   01)UnionExec
-   02)--FilterExec: b@0 > 2
-   03)----DataSourceExec: partitions=1, partition_sizes=[1]
-   04)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet]]}, projection=[b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+   07)--------TableScan: t_union_parquet
at /Users/mkleen/Test/datafusion/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt:713

This is related to this commit d09a919 If I revert this commit everything is passing. How should i proceed here?

[alamb]

This is related to the this commit d09a919 If I revert this commit everything is passing. How should i proceed here?

I recommend merging up from main and hopefully it passes; I'll click the button and see what happens

[alamb]

We'll probably have to figure out what is causing that failure / if we should update the expected output or if there is a bug to fix

[mkleen]

We'll probably have to figure out what is causing that failure / if we should update the expected output or if there is a bug to fix

The query in the test is executed twice, once for the output and once for the plan (EXPLAIN) and now the caching kicks in, which wasn't the case before. I will look into the details.

[mkleen]

I would say it's a bug which wasn't exposed before.

[mkleen]

The filter pushdown does not work as expected anymore when the stats are cached. The commit d09a919 changes the stats handling for this case. I will look into it.