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

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:

  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:

> 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

> 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

> 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

datafusion/datafusion/physical-plan/src/topk/mod.rs

Lines 48 to 101 in 9bb309c

    
           /// Global TopK 
        
           /// 
        
           /// # Background 
        
           /// 
        
           /// "Top K" is a common query optimization used for queries such as 
        
           /// "find the top 3 customers by revenue". The (simplified) SQL for 
        
           /// such a query might be: 
        
           /// 
        
           /// ```sql 
        
           /// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3; 
        
           /// ``` 
        
           /// 
        
           /// The simple plan would be: 
        
           /// 
        
           /// ```sql 
        
           /// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3; 
        
           /// +--------------+----------------------------------------+ 
        
           /// | plan_type    | plan                                   | 
        
           /// +--------------+----------------------------------------+ 
        
           /// | logical_plan | Limit: 3                               | 
        
           /// |              |   Sort: revenue DESC NULLS FIRST       | 
        
           /// |              |     Projection: customer_id, revenue   | 
        
           /// |              |       TableScan: sales                 | 
        
           /// +--------------+----------------------------------------+ 
        
           /// ``` 
        
           /// 
        
           /// While this plan produces the correct answer, it will fully sorts the 
        
           /// input before discarding everything other than the top 3 elements. 
        
           /// 
        
           /// The same answer can be produced by simply keeping track of the top 
        
           /// K=3 elements, reducing the total amount of required buffer memory. 
        
           /// 
        
           /// # Partial Sort Optimization 
        
           /// 
        
           /// This implementation additionally optimizes queries where the input is already 
        
           /// partially sorted by a common prefix of the requested ordering. Once the top K 
        
           /// heap is full, if subsequent rows are guaranteed to be strictly greater (in sort 
        
           /// order) on this prefix than the largest row currently stored, the operator 
        
           /// safely terminates early. 
        
           /// 
        
           /// ## Example 
        
           /// 
        
           /// For input sorted by `(day DESC)`, but not by `timestamp`, a query such as: 
        
           /// 
        
           /// ```sql 
        
           /// SELECT day, timestamp FROM sensor ORDER BY day DESC, timestamp DESC LIMIT 10; 
        
           /// ``` 
        
           /// 
        
           /// can terminate scanning early once sufficient rows from the latest days have been 
        
           /// collected, skipping older data. 
        
           /// 
        
           /// # Structure 
        
           /// 
        
           /// This operator tracks the top K items using a `TopKHeap`.

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:

datafusion/datafusion/physical-plan/src/aggregates/group_values/single_group_by/mod.rs

Lines 18 to 22 in 1d1f353

    
           //! `GroupValues` implementations for single group by cases 
        
           pub(crate) mod bytes; 
        
           pub(crate) mod bytes_view; 
        
           pub(crate) mod primitive;

Additional context

No response

@alamb

Hi @alamb , I’d like to take this issue if no one else is currently working on it.

@Thearas

sure -- feel free @Thearas

However, FYI I am not sure how easy/hard this will be

take

@alamb

Hi @alamb, which benchmark should I run to verify the improvement?

@Thearas

Hi @alamb, which benchmark should I run to verify the improvement?

Hi @Thearas -- I am not sure we have a specific benchmark yet -- that might be a good first PR

Perhaps you could add a "extended" clickbench query here https://github.com/apache/datafusion/tree/main/benchmarks/queries/clickbench#extended-queries

@alamb

@alamb In Option 2: Improve the TopK operators performance, do you mean we should use ScalarValue but not Array(or the encoded bytes) in TopKHeap for caching and comparing?

I find single column group by impl multiple kind of GroupValues, so I'm not sure.

I think the idea there is to specialize the code for concrete types (e.g. ArrowNativeType) so not ScalarValue, but f64, u32, etc.

Another option that we can try is to use https://docs.rs/arrow/latest/arrow/compute/kernels/sort/fn.sort_limit.html for small Ns (e.g. N <= 50) before converting to rows.
This does still a bit of unnecessary work (sort instead of partition) but for small N it should be OK. If it works out we could also implement / include a top kernel that only does the partitioning, rather than sorting.

untake

perf: skip wasted per-batch work in TopK insert_batch (apache#25657)

Navigation Menu

Repository navigation

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

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Option 1: Query rewrite

Option 2: Improve the TopK operators performance

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	/// Global TopK
	///
	/// # Background
	///
	/// "Top K" is a common query optimization used for queries such as
	/// "find the top 3 customers by revenue". The (simplified) SQL for
	/// such a query might be:
	///
	/// ```sql
	/// SELECT customer_id, revenue FROM 'sales.csv' ORDER BY revenue DESC limit 3;
	/// ```
	///
	/// The simple plan would be:
	///
	/// ```sql
	/// > explain SELECT customer_id, revenue FROM sales ORDER BY revenue DESC limit 3;
	/// +--------------+----------------------------------------+
	/// \| plan_type \| plan \|
	/// +--------------+----------------------------------------+
	/// \| logical_plan \| Limit: 3 \|
	/// \| \| Sort: revenue DESC NULLS FIRST \|
	/// \| \| Projection: customer_id, revenue \|
	/// \| \| TableScan: sales \|
	/// +--------------+----------------------------------------+
	/// ```
	///
	/// While this plan produces the correct answer, it will fully sorts the
	/// input before discarding everything other than the top 3 elements.
	///
	/// The same answer can be produced by simply keeping track of the top
	/// K=3 elements, reducing the total amount of required buffer memory.
	///
	/// # Partial Sort Optimization
	///
	/// This implementation additionally optimizes queries where the input is already
	/// partially sorted by a common prefix of the requested ordering. Once the top K
	/// heap is full, if subsequent rows are guaranteed to be strictly greater (in sort
	/// order) on this prefix than the largest row currently stored, the operator
	/// safely terminates early.
	///
	/// ## Example
	///
	/// For input sorted by `(day DESC)`, but not by `timestamp`, a query such as:
	///
	/// ```sql
	/// SELECT day, timestamp FROM sensor ORDER BY day DESC, timestamp DESC LIMIT 10;
	/// ```
	///
	/// can terminate scanning early once sufficient rows from the latest days have been
	/// collected, skipping older data.
	///
	/// # Structure
	///
	/// This operator tracks the top K items using a `TopKHeap`.

	//! `GroupValues` implementations for single group by cases

	pub(crate) mod bytes;
	pub(crate) mod bytes_view;
	pub(crate) mod primitive;

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

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Option 1: Query rewrite

Option 2: Improve the TopK operators performance

Additional context

Activity

Thearas commented on Aug 11, 2025

alamb commented on Aug 12, 2025

Thearas commented on Aug 22, 2025

Thearas commented on Nov 3, 2025

alamb commented on Nov 3, 2025

Thearas commented on Nov 4, 2025

Dandandan commented on Nov 4, 2025

Thearas commented on Jan 29, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

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