[Rust] [DataFusion] Improve parallelism when reading Parquet files

[asfimport]

Currently the unit of parallelism is the number of parquet files being read.

For example, if we run a query against a Parquet table that consists of 8 partitions then we will attempt to run 8 async tasks in parallel and if there is a single Parquet file then we will only try and run 1 async task so this does not scale well. Also, if there are hundreds or thousands of Parquet files then we will try and process them all concurrently which also doesn't scale well.

These are the options for improving this situation:

 

1. Use Parquet row groups as the unit of partitioning and divide the number of row groups by the desired level of concurrency (defaulting to number of cores)

2. Keep file as the unit of partitions and add a RepartitionExec into the plan if there are fewer partitions (files) than cores and in the case where there are more files than cores, split the files up into lists so that each partition is a list of files rather than a single file. Each partition task will process one file at a time.

    

    

Reporter: Andy Grove / @andygrove
Assignee: Andy Grove / @andygrove

Related issues:

• [Rust] Parquet ArrayReader should allow reading a subset of row groups (is blocked by)

PRs and other links:

• GitHub Pull Request #9029

_{Note: This issue was originally created as ARROW-10995. Please see the migration documentation for further details.}

[asfimport]

Andy Grove / @andygrove:
Issue resolved by pull request 9029
#9029

[alamb]

This came up on discord recently so moving some info

DataFusion supports reading the same parquet file using multiple threads

Here is the low level arrow API to do it from @tustvold

You can read row groups in parallel by creating separate readers with different row group selections

https://docs.rs/parquet/latest/parquet/arrow/arrow_reader/type.ParquetRecordBatchReaderBuilder.html#method.new_with_metadata

This is used by by datafusion automatically. The function that does the parallelization is

1. https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#method.repartitioned
2. implemented for https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/struct.ParquetExec.html