Improve performance of `datafusion-cli` when reading from remote storage

[alamb]

Is your feature request related to a problem or challenge?

• Part of https://github.com/apache/datafusion/pull/16300/files

While testing #16300, I (re-noticed) that datafusion-cli is really slow when reading remote files

For example,

The initial table creation takes 7.5 seconds on my pretty crappy connection:

DataFusion CLI v48.0.0
> CREATE EXTERNAL TABLE nyc_taxi_rides
STORED AS PARQUET LOCATION 's3://altinity-clickhouse-data/nyc_taxi_rides/data/tripdata_parquet/';

0 row(s) fetched.
Elapsed 7.492 seconds.

However then simple queries just to get the count take 8-10 seconds 😱

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched.
Elapsed 8.945 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched.
Elapsed 10.456 seconds.

I am almost certain this delay is due to having to read the footers of the parquet files for each query.

Note the speed is much faster when collect_statistics is on

> set datafusion.execution.collect_statistics = true;
0 row(s) fetched.
Elapsed 0.012 seconds.

> CREATE EXTERNAL TABLE nyc_taxi_rides
STORED AS PARQUET LOCATION 's3://altinity-clickhouse-data/nyc_taxi_rides/data/tripdata_parquet/';
0 row(s) fetched.
Elapsed 7.770 seconds.

The first query is still slow for some reason: (5 seconds)

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched.
Elapsed 5.114 seconds.

But subsequent queries are quite fast:

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched.
Elapsed 0.297 seconds.

Describe the solution you'd like

I would like:

1. datafusion-cli to be faster for such queries
2. datafusion-cli to be an easy to follow model for how to cache metadata when working with ListingTabe that others who build with DataFusion could follow

Describe alternatives you've considered

I think the obvious thing that is needed is a cache for the ParquetMetadata

I think the actual cache should be in datafusion-cli but NOT in the datafusion core crate as I think what and how to cache will be different across systems.

What I envision is:

1. APIs in the ListingTable / RuntimeEnv / etc for adding caching of ParquetMetadata
2. An implementation of those APIs in datafusion-cli
3. Bonus Points: documentation / examples that show how to use those APIs in other system

There is some vestigal code in the cache_manager crate that I think could provide a home for such caching APIs: * https://docs.rs/datafusion/latest/datafusion/execution/cache/cache_manager/index.html

Additional context

Related issues

• Filter cache based on the paper "Predicate Caching: Query-Driven Secondary Indexing for Cloud Data" #15585
• Cache Parquet Metadata #15582

[swaingotnochill]

Hi @alamb, I would like to take up on this. I am new to the project, but this seems like a good way to start.

[alamb]

Hi @alamb, I would like to take up on this. I am new to the project, but this seems like a good way to start.

It would certainly be nice, though it might be tricky for someone without experience with query processing from remote object storage.

[swaingotnochill]

I will try to study and dig more on this. It would be great if you can give starting pointers when i am stuck :)

[comphead]

Thanks @swaingotnochill I would probably start with profiling. The profiling techniques can be found
https://github.com/apache/datafusion/blob/main/docs/source/library-user-guide/profiling.md#profiling-using-samply-cross-platform-profiler

[swaingotnochill]

I started profiling and I can definitely see reading of parquet footers and metadata taking a chunk of time initially. Thanks @comphead

[JigaoLuo]

Hi @swaingotnochill
I’m curious about how we can isolate the footer parsing time from the end-to-end Parquet reading time. Could you explain the best approach to measure this for a runtime-breakdown?

[alamb]

Hi @swaingotnochill I’m curious about how we can isolate the footer parsing time from the end-to-end Parquet reading time. Could you explain the best approach to measure this for a runtime-breakdown?

I personally recommend looking at using samply to profile the run: https://datafusion.apache.org/library-user-guide/profiling.html#profile-the-benchmark

You could potentially use some of the example datasets in https://datafusion.apache.org/user-guide/cli/datasources.html#remote-files-directories

Something like
samply record datafusion-cli -c "select count(*) from 'https://datasets.clickhouse.com/hits_compatible/athena_partitioned/hits_1.parquet'"

Note you'll have to build datafusion-cli with --profile profiling

You should see something like this

[swaingotnochill]

Hi @swaingotnochill
I’m curious about how we can isolate the footer parsing time from the end-to-end Parquet reading time. Could you explain the best approach to measure this for a runtime-breakdown?

let metadata = Arc::new(
        ParquetMetaDataReader::new()
            .with_prefetch_hint(size_hint)
            .load_and_finish(fetch, file_size)
            .await?,
    );

This seems to be the lowest level that is doing both the I/O bound(fetching footer data) and CPU bound(parsing data). However, the fetches are asynchronous. I would likely start with this here and add metrics to measure this exactly in runtime. Other is the profiling which andrew suggested which seems simpler once we setup.

[alamb]

@XiangpengHao perhaps has implemented code that could be useful here: XiangpengHao/liquid-cache#227 (comment)

Basically it shows how to implement paruqet metadata caching with just the external APIs

It would be great to make this easier

@JigaoLuo has a great write up of why we need to cache metadata: #227 (comment)

Metadata cache update

I have a metadata cache implementation a while ago, but have procrastinated forever to actually polish it.

Maybe perfect is the enemy of good, I'll just show whatever I already have: https://github.com/XiangpengHao/parquet-study. Copy some of the readme here:

So you want to cache Parquet metadata in DataFusion? (so that one Parquet metadata is read/decoded once and only once).

It's not easy (a blog post is coming soon), but not impossible.

Usage

Copy the src/metedata_cache.rs to your project, and use it like below.

Option 1

use datafusion::prelude::*;
use crate::metadata_cache::RegisterParquetWithMetaCache;

let ctx = SessionContext::new();

// Instead of:
// ctx.register_parquet("table", "file.parquet", ParquetReadOptions::default()).await?;
ctx.register_parquet_with_meta_cache(
"table",
"path/to/file.parquet",
ParquetReadOptions::default()
).await?;

Option 2

If you're low-level listing table users:

use crate::metadata_cache::{ParquetFormatMetadataCacheFactory, ToListingOptionsWithMetaCache};

let parquet_options = ParquetReadOptions::default();
let listing_options = parquet_options.to_listing_options_with_meta_cache(&ctx.copied_config(), ctx.copied_table_options());

ctx.register_listing_table(
"table",
"path/to/file.parquet",
listing_options,
parquet_options.schema.map(|s| Arc::new(s.to_owned())),
None,
).await?;

More writings

Basically there are three places we read Parquet metadata:

1. Infer schema
2. Infer stats
3. Open Parquet files

Reading metadata has two costs:

1. IO cost to read the data. Each of the metadata read can cost up to 2 network request: first to load the Parquet footer, which decides the metadata size; second to load the actual metadata. But most of the case we only need 1, e.g., we optimistically fetch the last 4MB of the parquet file, which likely already covers the all metadata.
2. CPU cost to decode metadata. Metadata is not the fastest thing to decode, it's usually not a big problem, but some people complain about it.

(Since I have written so many things here, maybe just to finish that blog post?)

[swaingotnochill]

This seems nice. I am already working on an implementation, I can maybe take some inspiration from this. Let me have a basic POC working. I will try to create a draft PR by the end of this week.

[swaingotnochill]

I observed a significant performance improvement just testing on the master branch without any metadata caching as compared to before. I don't know what change caused this? Any idea.

➜  datafusion git:(main) ✗ ./target/debug/datafusion-cli 
DataFusion CLI v48.0.0
> CREATE EXTERNAL TABLE nyc_taxi_rides
STORED AS PARQUET LOCATION 's3://altinity-clickhouse-data/nyc_taxi_rides/data/tripdata_parquet/';
0 row(s) fetched. 
Elapsed 4.889 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 3.508 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 0.252 seconds.

[alamb]

I observed a significant performance improvement just testing on the master branch without any metadata caching as compared to before. I don't know what change caused this? Any idea.

It is almost certainly due to https://datafusion.apache.org/library-user-guide/upgrading.html#datafusion-execution-collect-statistics-now-defaults-to-true

[swaingotnochill]

@alamb You are right, the improvement is due to collect_statistics.

POC is able to reduce the original time, however it still doesn't seem to work very well with collect_statistics which is I am still working on.

This has most of the code from @XiangpengHao and some changes to incorporate it, but not very promising when compared to collect_statistics=true.

Here are the results:

➜  datafusion git:(rswain/parquet_metadata_caching) ✗ ./target/debug/datafusion-cli 
DataFusion CLI v48.0.0
> set datafusion.execution.collect_statistics = false;
0 row(s) fetched. 
Elapsed 0.008 seconds.

> CREATE EXTERNAL TABLE nyc_taxi_rides
STORED AS PARQUET LOCATION 's3://altinity-clickhouse-data/nyc_taxi_rides/data/tripdata_parquet/';

0 row(s) fetched. 
Elapsed 8.987 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 1.422 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 1.343 seconds.

> set datafusion.execution.collect_statistics = true;
0 row(s) fetched. 
Elapsed 0.002 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 1.818 seconds.

> select count(*) from nyc_taxi_rides;
+------------+
| count(*)   |
+------------+
| 1310903963 |
+------------+
1 row(s) fetched. 
Elapsed 1.157 seconds.