Repository: duckdb/duckdb (language: python)

Issue:
A confusion about Top-N operator
### What happens?

In `TopNHeap::Reduce`, the reduce threshold is determined by `payload_chunk.size()`. However, the `payload_chunk` is only used as a temporary data holder in `TopNHeap::Combine`. It's `heap_data` right?

### To Reproduce

```cpp
void TopNHeap::Combine(TopNHeap &other) {
	other.Finalize();

	// FIXME: heaps can be merged directly instead of doing it like this
	// that only really speeds things up if heaps are very large, however
	TopNScanState state;
	other.InitializeScan(state, false);
	while (true) {
		payload_chunk.Reset();
		other.Scan(state, payload_chunk);
		if (payload_chunk.size() == 0) {
			break;
		}
		Sink(payload_chunk);
	}
	Reduce();
}

void TopNHeap::Reduce() {
	if (payload_chunk.size() < ReduceThreshold()) {
		// only reduce when we pass the reduce threshold
		return;
	}
	...
}
```

### OS:

x86_64

### DuckDB Version:

newest commit id: b470dea7ee47dc2debcc37a4e94976f8eff6670c

### DuckDB Client:

none

### Hardware:

_No response_

### Full Name:

Yiliang Qiu

### Affiliation:

Baidu Inc.

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have not tested with any build

### Did you include all relevant data sets for reproducing the issue?

No - Other reason (please specify in the issue body)

### Did you include all code required to reproduce the issue?

- [x] Yes, I have

### Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

- [x] Yes, I have


Discussion / hints:
There are some addition to note: I benchmarked Top-N operator in v1.0.0 and newest commit and found that the performance of top-n in newest commit is degradation in some cases.

benchmark env:
- cpu: 96core
- memory: 754gb
- dataset: tpch sf100
- sql:
```sql
-- q1
select l_orderkey
from lineitem
where l_linestatus = 'O'
order by l_quantity limit 100 offset 100;

-- q2
select l_orderkey
from lineitem
where l_linestatus = 'O'
order by l_quantity limit 100 offset 1000000;

-- q3
select l_orderkey, sum(l_quantity)
from lineitem
group by l_orderkey
order by sum(l_quantity) desc
limit 100 offset 100;

-- q4
select l_orderkey, sum(l_quantity)
from lineitem
group by l_orderkey
order by sum(l_quantity) desc
limit 100 offset 1000000;
```

result (unit: sec):
| sql | v1.0.0 | main |
|:--------|:-------|:--------|
| q1 | 0.223 | 0.834 |
| q2 | 19.009 | 82.766 |
| q3 | 4.190 | 1.498 |
| q4 | 27.769 | 78.018 |

Fix patch:
diff --git a/src/execution/operator/order/physical_top_n.cpp b/src/execution/operator/order/physical_top_n.cpp
index 1b3ae11aed7a..dbee3bf617f7 100644
--- a/src/execution/operator/order/physical_top_n.cpp
+++ b/src/execution/operator/order/physical_top_n.cpp
@@ -90,14 +90,53 @@ class TopNHeap {
 	void InitializeScan(TopNScanState &state, bool exclude_offset);
 	void Scan(TopNScanState &state, DataChunk &chunk);
 
+	void AddSmallHeap(DataChunk &input, Vector &sort_keys_vec, const string &boundary_val,
+	                  const string_t &global_boundary_val);
+	void AddLargeHeap(DataChunk &input, Vector &sort_keys_vec, const string &boundary_val,
+	                  const string_t &global_boundary_val);
+
 public:
 	idx_t ReduceThreshold() const {
 		return MaxValue<idx_t>(STANDARD_VECTOR_SIZE * 5ULL, 2ULL * heap_size);
 	}
 
-	idx_t HeapAllocSize() const {
+	idx_t InitialHeapAllocSize() const {
 		return MinValue<idx_t>(STANDARD_VECTOR_SIZE * 100ULL, ReduceThreshold()) + STANDARD_VECTOR_SIZE;
 	}
+
+private:
+	inline bool EntryShouldBeAdded(const string_t &sort_key) {
+		if (heap.size() < heap_size) {
+			// heap is full - check the latest entry
+			return true;
+		}
+		if (sort_key < heap.front().sort_key) {
+			// sort key is smaller than current max value
+			return true;
+		}
+		// heap is full and there is no room for the entry
+		return false;
+	}
+
+	inline bool EntryShouldBeAdded(const string_t &sort_key, const string &boundary_val,
+	                               const string_t &global_boundary_val) {
+		// first compare against the global boundary value (if there is any)
+		if (!boundary_val.empty() && sort_key > global_boundary_val) {
+			// this entry is out-of-range for the global boundary val
+			// it will never be in the final result even if it fits in this heap
+			return false;
+		}
+		return EntryShouldBeAdded(sort_key);
+	}
+
+	inline void AddEntryToHeap(const TopNEntry &entry) {
+		if (heap.size() >= heap_size) {
+			std::pop_heap(heap.begin(), heap.end());
+			heap.pop_back();
+		}
+		heap.push_back(entry);
+		std::push_heap(heap.begin(), heap.end());
+	}
 };
 
 //===--------------------------------------------------------------------===//
@@ -115,9 +154,10 @@ TopNHeap::TopNHeap(ClientContext &context, Allocator &allocator, const vector<Lo
 		sort_types.push_back(expr->return_type);
 		executor.AddExpression(*expr);
 	}
+	heap.reserve(InitialHeapAllocSize());
 	vector<LogicalType> sort_keys_type {LogicalType::BLOB};
 	sort_keys.Initialize(allocator, sort_keys_type);
-	heap_data.Initialize(allocator, payload_types, HeapAllocSize());
+	heap_data.Initialize(allocator, payload_types, InitialHeapAllocSize());
 	payload_chunk.Initialize(allocator, payload_types);
 	sort_chunk.Initialize(allocator, sort_types);
 }
@@ -132,28 +172,8 @@ TopNHeap::TopNHeap(ExecutionContext &context, const vector<LogicalType> &payload
     : TopNHeap(context.client, Allocator::Get(context.client), payload_types, orders, limit, offset) {
 }
 
-void TopNHeap::Sink(DataChunk &input, optional_ptr<TopNBoundaryValue> global_boundary) {
-	// compute the ordering values for the new chunk
-	sort_chunk.Reset();
-	executor.Execute(input, sort_chunk);
-
-	// construct the sort key from the sort chunk
-	vector<OrderModifiers> modifiers;
-	for (auto &order : orders) {
-		modifiers.emplace_back(order.type, order.null_order);
-	}
-	sort_keys.Reset();
-	auto &sort_keys_vec = sort_keys.data[0];
-	CreateSortKeyHelpers::CreateSortKey(sort_chunk, modifiers, sort_keys_vec);
-
-	// fetch the current global boundary (if any)
-	string boundary_val;
-	string_t global_boundary_val;
-	if (global_boundary) {
-		boundary_val = global_boundary->GetBoundaryValue();
-		global_boundary_val = string_t(boundary_val);
-	}
-
+void TopNHeap::AddSmallHeap(DataChunk &input, Vector &sort_keys_vec, const string &boundary_val,
+                            const string_t &global_boundary_val) {
 	// insert the sort keys into the priority queue
 	constexpr idx_t BASE_INDEX = NumericLimits<uint32_t>::Maximum();
 
@@ -161,37 +181,20 @@ void TopNHeap::Sink(DataChunk &input, optional_ptr<TopNBoundaryValue> global_bou
 	auto sort_key_values = FlatVector::GetData<string_t>(sort_keys_vec);
 	for (idx_t r = 0; r < input.size(); r++) {
 		auto &sort_key = sort_key_values[r];
-		if (!boundary_val.empty() && sort_key > global_boundary_val) {
+		if (!EntryShouldBeAdded(sort_key, boundary_val, global_boundary_val)) {
 			continue;
 		}
-		if (heap.size() >= heap_size) {
-			// heap is full - check the latest entry
-			if (sort_key > heap.front().sort_key) {
-				// current max in the heap is smaller than the new key - skip this entry
-				continue;
-			}
-		}
 		// replace the previous top entry with the new entry
 		TopNEntry entry;
 		entry.sort_key = sort_key;
 		entry.index = BASE_INDEX + r;
-		if (heap.size() >= heap_size) {
-			std::pop_heap(heap.begin(), heap.end());
-			heap.pop_back();
-		}
-		heap.push_back(entry);
-		std::push_heap(heap.begin(), heap.end());
+		AddEntryToHeap(entry);
 		any_added = true;
 	}
 	if (!any_added) {
 		// early-out: no matches
 		return;
 	}
-	// if we modified the heap we might be able to update the global boundary
-	// note that the global boundary only applies to FULL heaps
-	if (heap.size() >= heap_size && global_boundary) {
-		global_boundary->UpdateValue(heap.front().sort_key);
-	}
 
 	// for all matching entries we need to copy over the corresponding payload values
 	idx_t match_count = 0;
@@ -214,20 +217,98 @@ void TopNHeap::Sink(DataChunk &input, optional_ptr<TopNBoundaryValue> global_bou
 	heap_data.Append(input, true, &matching_sel, match_count);
 }
 
+void TopNHeap::AddLargeHeap(DataChunk &input, Vector &sort_keys_vec, const string &boundary_val,
+                            const string_t &global_boundary_val) {
+	auto sort_key_values = FlatVector::GetData<string_t>(sort_keys_vec);
+	idx_t base_index = heap_data.size();
+	idx_t match_count = 0;
+	for (idx_t r = 0; r < input.size(); r++) {
+		auto &sort_key = sort_key_values[r];
+		if (!EntryShouldBeAdded(sort_key, boundary_val, global_boundary_val)) {
+			continue;
+		}
+		// replace the previous top entry with the new entry
+		TopNEntry entry;
+		entry.sort_key = sort_key.IsInlined() ? sort_key : sort_key_heap.AddBlob(sort_key);
+		entry.index = base_index + match_count;
+		AddEntryToHeap(entry);
+		matching_sel.set_index(match_count++, r);
+	}
+	if (match_count == 0) {
+		// early-out: no matches
+		return;
+	}
+
+	// copy over the input rows to the payload chunk
+	heap_data.Append(input, true, &matching_sel, match_count);
+}
+
+void TopNHeap::Sink(DataChunk &input, optional_ptr<TopNBoundaryValue> global_boundary) {
+	static constexpr idx_t SMALL_HEAP_THRESHOLD = 100;
+
+	// compute the ordering values for the new chunk
+	sort_chunk.Reset();
+	executor.Execute(input, sort_chunk);
+
+	// construct the sort key from the sort chunk
+	vector<OrderModifiers> modifiers;
+	for (auto &order : orders) {
+		modifiers.emplace_back(order.type, order.null_order);
+	}
+	sort_keys.Reset();
+	auto &sort_keys_vec = sort_keys.data[0];
+	CreateSortKeyHelpers::CreateSortKey(sort_chunk, modifiers, sort_keys_vec);
+
+	// fetch the current global boundary (if any)
+	string boundary_val;
+	string_t global_boundary_val;
+	if (global_boundary) {
+		boundary_val = global_boundary->GetBoundaryValue();
+		global_boundary_val = string_t(boundary_val);
+	}
+
+	if (heap_size <= SMALL_HEAP_THRESHOLD) {
+		AddSmallHeap(input, sort_keys_vec, boundary_val, global_boundary_val);
+	} else {
+		AddLargeHeap(input, sort_keys_vec, boundary_val, global_boundary_val);
+	}
+
+	// if we modified the heap we might be able to update the global boundary
+	// note that the global boundary only applies to FULL heaps
+	if (heap.size() >= heap_size && global_boundary) {
+		global_boundary->UpdateValue(heap.front().sort_key);
+	}
+}
+
 void TopNHeap::Combine(TopNHeap &other) {
 	other.Finalize();
 
-	// FIXME: heaps can be merged directly instead of doing it like this
-	// that only really speeds things up if heaps are very large, however
-	TopNScanState state;
-	other.InitializeScan(state, false);
-	while (true) {
-		payload_chunk.Reset();
-		other.Scan(state, payload_chunk);
-		if (payload_chunk.size() == 0) {
-			break;
+	idx_t match_count = 0;
+	// merge the heap of other into this
+	for (idx_t i = 0; i < other.heap.size(); i++) {
+		// heap is full - check the latest entry
+		auto &other_entry = other.heap[i];
+		auto &sort_key = other_entry.sort_key;
+		if (!EntryShouldBeAdded(sort_key)) {
+			continue;
+		}
+		// add this entry
+		TopNEntry new_entry;
+		new_entry.sort_key = sort_key.IsInlined() ? sort_key : sort_key_heap.AddBlob(sort_key);
+		new_entry.index = heap_data.size() + match_count;
+		AddEntryToHeap(new_entry);
+
+		matching_sel.set_index(match_count++, other_entry.index);
+		if (match_count >= STANDARD_VECTOR_SIZE) {
+			// flush
+			heap_data.Append(other.heap_data, true, &matching_sel, match_count);
+			match_count = 0;
 		}
-		Sink(payload_chunk);
+	}
+	if (match_count > 0) {
+		// flush
+		heap_data.Append(other.heap_data, true, &matching_sel, match_count);
+		match_count = 0;
 	}
 	Reduce();
 }
@@ -236,14 +317,14 @@ void TopNHeap::Finalize() {
 }
 
 void TopNHeap::Reduce() {
-	if (payload_chunk.size() < ReduceThreshold()) {
+	if (heap_data.size() < ReduceThreshold()) {
 		// only reduce when we pass the reduce threshold
 		return;
 	}
 	// we have too many values in the heap - reduce them
 	StringHeap new_sort_heap;
-	DataChunk new_payload_chunk;
-	new_payload_chunk.Initialize(allocator, payload_types, HeapAllocSize());
+	DataChunk new_heap_data;
+	new_heap_data.Initialize(allocator, payload_types, heap.size());
 
 	SelectionVector new_payload_sel(heap.size());
 	for (idx_t i = 0; i < heap.size(); i++) {
@@ -258,21 +339,22 @@ void TopNHeap::Reduce() {
 	}
 
 	// copy over the data from the current payload chunk to the new payload chunk
-	payload_chunk.Copy(new_payload_chunk, new_payload_sel, heap.size());
+	new_heap_data.Slice(heap_data, new_payload_sel, heap.size());
+	new_heap_data.Flatten();
 
-	new_sort_heap.Move(sort_key_heap);
-	payload_chunk.Reference(new_payload_chunk);
+	sort_key_heap.Move(new_sort_heap);
+	heap_data.Reference(new_heap_data);
 }
 
 void TopNHeap::InitializeScan(TopNScanState &state, bool exclude_offset) {
 	auto heap_copy = heap;
 	// traverse the rest of the heap
+	state.scan_order.resize(heap_copy.size());
 	while (!heap_copy.empty()) {
 		std::pop_heap(heap_copy.begin(), heap_copy.end());
-		state.scan_order.push_back(UnsafeNumericCast<sel_t>(heap_copy.back().index));
+		state.scan_order[heap_copy.size() - 1] = UnsafeNumericCast<sel_t>(heap_copy.back().index);
 		heap_copy.pop_back();
 	}
-	std::reverse(state.scan_order.begin(), state.scan_order.end());
 	state.pos = exclude_offset ? offset : 0;
 }
 


Test patch:
diff --git a/test/optimizer/compressed_materialization.test_coverage b/test/optimizer/compressed_materialization.test_slow
similarity index 98%
rename from test/optimizer/compressed_materialization.test_coverage
rename to test/optimizer/compressed_materialization.test_slow
index 4088c1efab2d..a39f17f8a4af 100644
--- a/test/optimizer/compressed_materialization.test_coverage
+++ b/test/optimizer/compressed_materialization.test_slow
@@ -1,4 +1,4 @@
-# name: test/optimizer/compressed_materialization.test_coverage
+# name: test/optimizer/compressed_materialization.test_slow
 # description: Compressed materialization test
 # group: [optimizer]
 
@@ -18,7 +18,7 @@ Binder Error: Compressed materialization functions are for internal use only!
 statement ok
 create table t0 as select range%400000 a, range%400000 b from range(500000);
 
-query III
+query III rowsort
 select * from (
       select *, row_number() OVER () as row_number from (
           SELECT * FROM t0 ORDER BY 1) ta
diff --git a/test/sql/topn/tpch_top_n.test_slow b/test/sql/topn/tpch_top_n.test_slow
new file mode 100644
index 000000000000..a3657bb05faf
--- /dev/null
+++ b/test/sql/topn/tpch_top_n.test_slow
@@ -0,0 +1,78 @@
+# name: test/sql/topn/tpch_top_n.test_slow
+# description: Test Top N NULLS FIRST/LAST with few rows
+# group: [topn]
+
+require tpch
+
+statement ok
+CALL dbgen(sf=1);
+
+query I
+select l_quantity
+from lineitem
+where l_linestatus = 'O'
+order by l_quantity limit 10 offset 100;
+----
+1.00
+1.00
+1.00
+1.00
+1.00
+1.00
+1.00
+1.00
+1.00
+1.00
+
+query I
+select l_quantity
+from lineitem
+where l_linestatus = 'O'
+order by l_quantity limit 10 offset 1000000;
+----
+17.00
+17.00
+17.00
+17.00
+17.00
+17.00
+17.00
+17.00
+17.00
+17.00
+
+query I
+select sum(l_quantity)
+from lineitem
+group by l_orderkey
+order by sum(l_quantity) desc
+limit 10 offset 100;
+----
+297.00
+296.00
+296.00
+296.00
+296.00
+296.00
+296.00
+296.00
+295.00
+295.00
+
+query I
+select sum(l_quantity)
+from lineitem
+group by l_orderkey
+order by sum(l_quantity) desc
+limit 10 offset 100000;
+----
+195.00
+195.00
+195.00
+195.00
+195.00
+195.00
+195.00
+195.00
+195.00
+195.00