apache / apache/datafusion

[EPIC] Avoid sort for already sorted Parquet files that do not overlap values on condition

Open
#6,672 9 comments 0 reactions 0 assignees View on GitHub
enhancement performance PROPOSAL EPIC question
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Describe the bug

I'm testing performance of querying a number of Parquet files, where I can make some assumptions about the Parquet files.
- Each Parquet file is already sorted on the column "timestamp".
- Each Parquet file does not overlap values on the column "timestamp". For instance, file A has values for timestamps for 2022, and file B has values for timestamps 2023.

The schema of the files are:
- "timestamp": TimestampMillisecond
- "value": Float64

Consider the following query and it's query plan:

```sql
SELECT timestamp, value
FROM samples
ORDER BY timestamp ASC
```

```
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortPreservingMergeExec: [timestamp@0 ASC], metrics=[output_rows=1000000, elapsed_compute=572.526968ms] |
| | ParquetExec: file_groups={20 groups: [[0.parquet], [1.parquet], [2.parquet], [3.parquet], [4.parquet], ...]}, projection=[timestamp, value], output_ordering=[timestamp@0 ASC], metrics=[output_rows=1000000, elapsed_compute=20ns, num_predicate_creation_errors=0, predicate_evaluation_errors=0, bytes_scanned=57972, page_index_rows_filtered=0, row_groups_pruned=0, pushdown_rows_filtered=0, time_elapsed_processing=51.918935ms, page_index_eval_time=40ns, time_elapsed_scanning_total=48.94925ms, time_elapsed_opening=2.996325ms, time_elapsed_scanning_until_data=48.311008ms, pushdown_eval_time=40ns] |
| | |
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
```

The 572 milliseconds on the `SortPreservingMergeExec` seems to be the bottleneck in the query, so I would like to optimize it.

Given the assumptions I can make about the Parquet files, I think that the `SortPreservingMergeExec` can be replaced by what is essentially a concatenation of each of the Parquet files.

What would be the best approach to remove the `SortPreservingMergeExec`?
My ideas:
- Manually re-partition the Parquet files into a single Parquet file using this new API: https://docs.rs/parquet/latest/parquet/file/writer/struct.SerializedRowGroupWriter.html#method.append_column
- I have an idea of implementing a custom `PhysicalOptimizerRule` that looks for the `SortPreservingMergeExec ParquetExec` pattern, and replaces it with a concatenation instead.

But I would like to hear if there are any better ways.

### Related
* Blog post about this optimization: https://www.influxdata.com/blog/making-recent-value-queries-hundreds-times-faster/

### Infrastructure Tasks 🚧
- [x] https://github.com/apache/datafusion/issues/15290
- [x] https://github.com/apache/datafusion/pull/15432
- [x] https://github.com/apache/datafusion/pull/15379
- [x] https://github.com/apache/datafusion/issues/15689
- [ ] https://github.com/apache/datafusion/issues/15191
- [ ] https://github.com/apache/datafusion/pull/15539

### Major Tasks
- [x] https://github.com/apache/datafusion/issues/15495
- [ ] https://github.com/apache/datafusion/issues/10316

### Related (though not necessarily required)
- [ ] https://github.com/apache/datafusion/issues/8078
- [ ] https://github.com/apache/datafusion/issues/10336
- [x] https://github.com/apache/datafusion/issues/10488

Contributor guide

Open the contributing guide

Research direction

Start by reading the SortPreservingMergeExec and ParquetExec plan entries, then compare the proposed PhysicalOptimizerRule approach with the SerializedRowGroupWriter API mentioned in the issue. Done means establishing a safe way to concatenate non-overlapping, timestamp-sorted Parquet files without the merge bottleneck, with validation for the stated ordering assumptions.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, sql
Domain
data-engineering, databases, performance
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.