apache / apache/datafusion

Optimize `ORDER BY time DESC LIMIT 1` queries ( TopK or aggr rewrite)

Open
#17,098 8 comments 3 reactions 0 assignees View on GitHub
enhancement performance
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Is your feature request related to a problem or challenge?

This came up in the context of an internal investigation I did at InfluxData. I am not sure how common this use case is, but since I spent time analyzing it, I figured I would share it here.

We have a query like
```
SELECT .. ORDER BY time DESC LIMIT 1;
```

The plan looks like this:
```sql
SortExec: TopK(fetch=1), expr=[time@0 DESC], ...]
DataSourceExec: file_groups={1 groups: [...]]
```

This uses the TopK operator to find the single largest value in the `time` column. Clearly a better
approach would be a query like `SELECT MIN(time)`. However the metrics from our query show that the TopK operator is taking a significant time (there are 200M input rows):
```
SortExec: TopK(fetch=1), expr=[time@0 DESC], preserve_partitioning=[true], metrics=[output_rows=2, elapsed_compute=818.779943ms, row_replacements=2]
```

### Describe the solution you'd like

Make queries like the above go faster

Here is a test query you could use to see the difference:

```sql
> select value from generate_series(1, 10000000000) ORDER BY value LIMIT 1;
+-------+
| value |
+-------+
| 1 |
+-------+
1 row(s) fetched.
Elapsed 5.649 seconds.
```

Here is the plan
```sql
> explain format indent select value from generate_series(1, 10000000000) ORDER BY value LIMIT 1;
+---------------+---------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+---------------------------------------------------------------------------------------------------------------+
| logical_plan | Sort: generate_series().value ASC NULLS LAST, fetch=1 |
| | TableScan: generate_series() projection=[value] |
| physical_plan | SortExec: TopK(fetch=1), expr=[value@0 ASC NULLS LAST], preserve_partitioning=[false] |
| | LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=10000000000, batch_size=8192] |
| | |
+---------------+---------------------------------------------------------------------------------------------------------------+
```

Explain Analyze showing that the TopK operator takes 1.3s of compute time
```sql
> explain analyze select value from generate_series(1, 10000000000) ORDER BY value LIMIT 1;
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortExec: TopK(fetch=1), expr=[value@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[value@0 < 1], metrics=[output_rows=1, elapsed_compute=1.313177866s, row_replacements=1] |
| | LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=10000000000, batch_size=8192], metrics=[output_rows=10000000000, elapsed_compute=4.835229677s] |
| | |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 6.272 seconds.
```

### Describe alternatives you've considered

## Option 1: Query rewrite
One possibility is to rewrite queries automatically to use `min` / `max` rather than ORDER BY / LIMIT.

## Option 2: Improve the TopK operators performance

Another possibility is to improve the TopK operator itself

Ehen I looked at the TopK operator,
code https://github.com/apache/datafusion/blob/9bb309c88ae94d495b487a97cb635c52225689e8/datafusion/physical-plan/src/topk/mod.rs#L48-L101

I found it uses the Row format to compare entries. We could potentially optimize
this code with a specialized implementation for single columns when we know the type, similarly
to how we have specialized implementations for single column group by columns:

https://github.com/apache/datafusion/blob/1d1f3534b4ef790d376ed6fde69a3b404c8c988d/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs#L18-L22

### Additional context

_No response_

Contributor guide

Open the contributing guide

Research direction

Start with the TopK implementation in datafusion/physical-plan/src/topk/mod.rs and compare it with the single-column group-by implementation in datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs. Run the provided generate_series ORDER BY value LIMIT 1 query and EXPLAIN ANALYZE to establish the baseline. Done means the query is measurably faster through either a rewrite or a more efficient TopK path.

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
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.