[EPIC] Improve the performance of ListingTable

[alamb]

Is your feature request related to a problem or challenge?

The ListingTable works quite well in practice, but like all software could be made better. I am writing up this ticket to enumerate some areas for improvement in the hopes people who are interested can collaborate / coordinate their efforts

Background

DataFusion has a ListingTable that effectively reading tables stored in one or more files in a "hive partitioned" directory structure:

So for example, give files like this:

/path/to/my_table/file1.parquet
/path/to/my_table/file2.parquet
/path/to/my_table/file3.parquet

You can create a table with a command like

CREATE EXTERNAL TABLE my_table
LOCATION '/path/to/my_table'

And the ListingTable will handle figuring out schema, and running queries against those files as though they were a single table.

Describe the solution you'd like

Here are some things I suspect could be improved:

All Formats

Object store list caching

For large tables (many files) on remote stores, the actual object store call to LIST may be non trivially expensive and thus doing over and over is expensive

@henrifroese points out a similar thing for pruning partitions #9654

Parquet Specific

MetaData caching

ListingTable (code link) prunes files based on statistics, and then inside the ParquetExec itself (link) where it again prunes row groups and data pages based on metadata. Fetching and parsing this metatadata twice (once to prune files and once to prune row groups) could be improved

IO granularity

I have heard it said that the DataFusion ParquetExec reader reads a page at a time -- this is fine if the parquet file is a local file on disk, but it is likely quite inefficient if each page must be fetched with an individual remote object store request. This assertion needs to be researched, but if true we could make queries on remote parquet files much faster by making fewer larger requests

Describe alternatives you've considered

@Ted-Jiang added some APIs in #7570 https://github.com/apache/arrow-datafusion/blob/2b0a7db0ce64950864e07edaddfa80756fe0ffd5/datafusion/execution/src/cache/mod.rs but there aren't any default implementations in DataFusion so the metadata is read multiple times

Maybe we can add a default implementation of the caches in SessionContext with a simple policy (like LRU / some max size)

Another potential way to improve performance is to cache the decoded metadata from the Parquet footer rather than checking it once to prune files and then again to prune row groups / pages. This could be taken even farther with pruning files and row groups and pages in one go using and API like #9929

Additional context

@matthewmturner mentioned interest in improving listing table performance: #9899 (comment)

Note we don't use ListingTable in InfluxDB for some of the reasons described above

Related tickets:

• Partitioned object store lists all files on every query when using hive-partitioned parquet files #9654
• API in ParquetExec to pass in RowSelections to ParquetExec (enable custom indexes, finer grained pushdown) #9929
• Cache Parquet Metadata #15582
• Filter cache based on the paper "Predicate Caching: Query-Driven Secondary Indexing for Cloud Data" #15585

[Lordworms]

Interested in this one

[alamb]

FYI I think this is more like an Epic that can be used to coordinate individual tasks / changes rather than a specific change itself.

Interested in this one

Thanks @Lordworms -- one thing that would probably help to start this project along would be to gather some data.

Specifically, put the LIstingTable against data on a remote object store (eg. figure out how to write a query against 100 parquet files on an S3 bucket).

And then measure how much time is spent:

1. object store listing
2. fetching metadata
3. pruning / fetching IO
4. How many object store requests are made

Does anyone know a good public data set on S3 that we could use to test / benchmark with?

[Lordworms]

FYI I think this is more like an Epic that can be used to coordinate individual tasks / changes rather than a specific change itself.

Interested in this one

Thanks @Lordworms -- one thing that would probably help to start this project along would be to gather some data.

Specifically, put the LIstingTable against data on a remote object store (eg. figure out how to write a query against 100 parquet files on an S3 bucket).

And then measure how much time is spent:

1. object store listing
2. fetching metadata
3. pruning / fetching IO
4. How many object store requests are made

Does anyone know a good public data set on S3 that we could use to test / benchmark with?

I got it, I'll search some data first

[Lordworms]

I'll start with a random dataset here https://registry.opendata.aws/

[MohamedAbdeen21]

(It's not a "performance" issue, but rather for better user experience. Also, it requires upstream changes to arrow-rs/object_store.)

The recent change in #9912 uses 10 random files to infer the partition columns, this means that we may fail to catch corrupted/manually-changed partitions on table creation (shouldn't be a common case). This is because ObjectStore only provides list function to retrieve objects.

If we can provide a BFS approach to traverse the Object Store and use that in partition inference, that will be a nice QoL change.

Curious to know how we feel about upstream changes for such non-critical changes in DF?

[alamb]

The recent change in #9912 uses 10 random files to infer the partition columns, this means that we may fail to catch corrupted/manually-changed partitions on table creation (shouldn't be a common case). This is because ObjectStore only provides list function to retrieve objects.

I think ObjectStore only provides a LIST api as that is the functionality offered by S3, GCP, etc.

The recent change in #9912 uses 10 random files to infer the partition columns, this means that we may fail to catch corrupted/manually-changed partitions on table creation (shouldn't be a common case).

In this particular case, I think there are different needs for different users (some might want to pay the cost for a full validation, but some might be happy with just a sanity check) One approach would be to add a config option to DataFusion that controls the maxumum number of paths to check when creating a listing table 🤔

Curious to know how we feel about upstream changes for such non-critical changes in DF?

In general I think upstream changes are great to propose if they help more than just DataFusion

[Lordworms]

I have done some basic play with the bitcoin dataset

and also did some profiling with instrument

FYI I think this is more like an Epic that can be used to coordinate individual tasks / changes rather than a specific change itself.

Interested in this one

Thanks @Lordworms -- one thing that would probably help to start this project along would be to gather some data.

Specifically, put the LIstingTable against data on a remote object store (eg. figure out how to write a query against 100 parquet files on an S3 bucket).

And then measure how much time is spent:

1. object store listing

> 2. fetching metadata

3. pruning / fetching IO
4. How many object store requests are made

Does anyone know a good public data set on S3 that we could use to test / benchmark with?

just want to know what is a good start to solving this issue, should I implement the cache https://github.com/apache/arrow-datafusion/blob/2b0a7db0ce64950864e07edaddfa80756fe0ffd5/datafusion/execution/src/cache/mod.rs here first?

[alamb]

And then measure how much time is spent:

that is very interesting

just want to know what is a good start to solving this issue, should I implement the cache

just want to know what is a good start to solving this issue, should I implement the cache https://github.com/apache/arrow-datafusion/blob/2b0a7db0ce64950864e07edaddfa80756fe0ffd5/datafusion/execution/src/cache/mod.rs here first?

If indeed most of the exection time is spent parsing (or fetching) parquet metadata, implementing a basic cache would likely help.

Also, @tustvold brought https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/trait.ParquetFileReaderFactory.html to my attention which might be able to help avoid the overhead

So what I suggest is:

1. Do a proof of concept (POC - hack it in, don't worry about tests, etc) with your approach and see if you can show performance improvements (WIP: Avoid copying LogicalPlans / Exprs during OptimizerPasses #9708 is an example of such a PR)
2. If you can show it improves performance significantly, then we can work on a final design / tests / etc

The reason to do the POC first is that performance analysis is notoriously tricky at the system lavel so you want to have evidence your work will actually improve performance before you spend a bunch of time polishing up the PR (it is very demotivating, at least to me, to make a beautiful PR only to find out it doesn't really help performance)

[Lordworms]

And then measure how much time is spent:

that is very interesting

just want to know what is a good start to solving this issue, should I implement the cache

just want to know what is a good start to solving this issue, should I implement the cache https://github.com/apache/arrow-datafusion/blob/2b0a7db0ce64950864e07edaddfa80756fe0ffd5/datafusion/execution/src/cache/mod.rs here first?

If indeed most of the exection time is spent parsing (or fetching) parquet metadata, implementing a basic cache would likely help.

Also, @tustvold brought https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/trait.ParquetFileReaderFactory.html to my attention which might be able to help avoid the overhead

So what I suggest is:

1. Do a proof of concept (POC - hack it in, don't worry about tests, etc) with your approach and see if you can show performance improvements (WIP: Avoid copying LogicalPlans / Exprs during OptimizerPasses #9708 is an example of such a PR)
2. If you can show it improves performance significantly, then we can work on a final design / tests / etc

The reason to do the POC first is that performance analysis is notoriously tricky at the system lavel so you want to have evidence your work will actually improve performance before you spend a bunch of time polishing up the PR (it is very demotivating, at least to me, to make a beautiful PR only to find out it doesn't really help performance)

Got it