Cache Parquet Metadata

[alamb]

Is your feature request related to a problem or challenge?

When looking at some Samply profiles of ClickBench queries on my laptop, it appears there are several times where processing stalls due to parsing parquet metadata:

To reproduce, profile using Samply

To reproduce, get the ClickBench dataset

cd benchmarks
./bench.sh data clickbench_1

Then run

datafusion-cli -c "SELECT \"WatchID\", \"ClientIP\", COUNT(*) AS c, SUM(\"IsRefresh\"), AVG(\"ResolutionWidth\") FROM 'data/hits.parquet' WHERE \"SearchPhrase\" <> '' GROUP BY \"WatchID\", \"ClientIP\" ORDER BY c DESC LIMIT 10;"

Profile wiht samply (you must build datafusion-cli with `--profile profiling):

samply record datafusion-cli -c "SELECT \"WatchID\", \"ClientIP\", COUNT(*) AS c, SUM(\"IsRefresh\"), AVG(\"ResolutionWidth\") FROM 'data/hits.parquet' WHERE \"SearchPhrase\" <> '' GROUP BY \"WatchID\", \"ClientIP\" ORDER BY c DESC LIMIT 10;"

Describe the solution you'd like

I think we should look into caching this meta

There is a bunch of prior art like

• Improve parquet ListingTable speed with parquet metadata (short clickbench queries) #11719

Also in theory this API should allow metadata caching:

• https://docs.rs/datafusion/latest/datafusion/execution/cache/cache_manager/struct.CacheManager.html

But I don't think there is a default implementation and it isn't hooked up

Describe alternatives you've considered

What I would suggest doing first is

1. Do profiling / confirm you see the same thing
2. Make a quick and dirty global parquet metadata cache (just put it into some global variable key on filename)

If you see significant performance improvements with 2, then we can figure out how to get it in for real

Additional context

No response

[matthewmturner]

I am working on this for dft right now actually and I plan on integrating it into the observability feature that I have been working on (where different observability metrics are exposed as tables). Specifically, I made a new MapTable that allows querying and updating different types of map data structures (for now just an IndexMap but i plan to try others like BTreeMap as well). The idea is that the same map that is used for caching both file and parquet metadata can be queried to get details such as when the entry was created, last update, # of hits, etc.

I would be happy to share / upstream any work I do on this if there is interest.

[alamb]

I would be happy to share / upstream any work I do on this if there is interest.

Thanks @matthewmturner -- what I think would be really valuable is if you could prove/disprove the theory that caching parsed metadata improves local TPCH performance

I think it is already pretty well understood that avoiding object store requests to remote object stores helps, and many downstream systems already do this (influx, pydantic, etc) .

However, we haven't built in caching for metadata locally because I think we assumed the overhead wasn't very big compared to the complexity of a built in metadata cache -- maybe we should re-evalute that assesment

[adriangb]

I'll mention that we now avoid reading metadata entirely for a lot of queries using an approach along the lines of #15585

[JigaoLuo]

As a DataFusion user, I’m wondering if the Parquet footer could be cached. This blog mentioned footer caching. It is a bit confusing to me.

(I’m also involved in promoting footer caching in cuDF, which is why I mentioned this issue there as well. In cuDF, Thrift decoding is a bottleneck on the CPU due to its single-threaded implementation, often leaving the GPU idle.)

[alamb]

As a DataFusion user, I’m wondering if the Parquet footer could be cached. This blog mentioned footer caching. It is a bit confusing to me.

(I’m also involved in promoting metadata caching in cuDF, which is why I mentioned this issue there as well.)

DataFusion (and datafusion-cli) doesn't do footer caching by default -- you can see this here:

• Cache Parquet Metadata #15582
• Improve performance of datafusion-cli when reading from remote storage #16365

You can do it and most high performance integrations (like InfluxDB and pydantic) absolutely do. An example of doing it is here:
https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/advanced_parquet_index.rs

I am thinking adding the footer caching thing to datafusion-cli might be a nice way to improve performance and make it easier to see how to add these features

[nuno-faria]

@alamb is this still something that would benefit upstream Datafusion?

I've implemented the caching of Parquet metadata after noticing that a large amount of time spent on our workloads comes from (re)reading it, as pointed by metadata_load_time. This time becomes noticeable in large files that have a large number of pages, where most of the metadata reading times comes from retrieving the page indexes.

Here are some results:

• Simple reads

  +----------------------------------------------------+
  |      1k simple queries (select where k = ...)      |
  +-----------+------------+---------+-----------------+
  | # of Rows | Not cached |  Cached |                 |
  +-----------+------------+---------+-----------------+
  |    100k   |   1.9447s  | 1.1773s |  1.6517x faster |
  +-----------+------------+---------+-----------------+
  |    100M   |  21.2953s  | 1.6018s | 13.2943x faster |
  +-----------+------------+---------+-----------------+
  

• Partial explain of a DataSourceExec:

  # not cached
  metrics=[
      bytes_scanned=938381
      metadata_load_time=12.697101ms
      time_elapsed_opening=13.0083ms
      time_elapsed_processing=13.1382ms
      ...
  ]
  
  # cached
  metrics=[
      bytes_scanned=22488
      metadata_load_time=35.801µs
      time_elapsed_opening=243.4µs
      time_elapsed_processing=817.2µs
      ...
  ]
  

• TPC-H

  • As for TPC-H, the improvements won't be as noticeable given the queries are more complex, but they will be there. Here are the metadata_load_times of a TPC-H query (sf=10):

    # not cached
    [nation]    metadata_load_time=338.501µs
    [orders]    metadata_load_time=95.710812ms
    [supplier]  metadata_load_time=1.478501ms
    [lineitem]  metadata_load_time=424.488412ms
    [lineitem]  metadata_load_time=383.579612ms
    [lineitem]  metadata_load_time=461.110012ms
    
    # cached
    [nation]    metadata_load_time=76.001µs
    [orders]    metadata_load_time=15.823212ms
    [supplier]  metadata_load_time=49.201µs
    [lineitem]  metadata_load_time=906.312µs
    [lineitem]  metadata_load_time=6.372212ms
    [lineitem]  metadata_load_time=1.617712ms
    

As for the implementation, it relies on the session's CacheManager and a custom CachedParquetFileReaderFactory. The metadata caching can be set independently for each Parquet file, using ParquetReadOptions, and is invalidated when the underlying file is modified. I would be happy to open a PR, but I still have some questions:

• For the CacheManager to remain generic, I created a pub type FileMetadata = dyn Any + Send + Sync; to represent metadata, which essentially can end up storing anything. Unlike the other information stored (Statistics and ObjectMeta), there isn't a common type for embedded metadata. Is this the right approach, or should the CacheManager be aware of ParquetMetaData?
• Unlike the other CacheManager parameters, which I believe are exclusively user-provided, I think it would make sense for the metadata cache be populated with a DefaultFilesMetadataCache, so its easier to enable caching just with ParquetReadOptions or set .... Does this make sense?

[alamb]

| 100M | 21.2953s | 1.6018s | 13.2943x faster |

13x faster 👏

I would love to help review / get this feature merged in

For the CacheManager to remain generic, I created a pub type FileMetadata = dyn Any + Send + Sync; to represent metadata, which essentially can end up storing anything. Unlike the other information stored (Statistics and ObjectMeta), there isn't a common type for embedded metadata. Is this the right approach, or should the CacheManager be aware of ParquetMetaData?

I think keeping it generic makes the most sense because as you say there isn't something clearly that maps to all file formats

Unlike the other CacheManager parameters, which I believe are exclusively user-provided, I think it would make sense for the metadata cache be populated with a DefaultFilesMetadataCache, so its easier to enable caching just with ParquetReadOptions or set .... Does this make sense?

It sounds so but I don't fully understand the proposal so I may me missing someting