Repository: duckdb/duckdb (language: python)

Issue:
Out of memory on basic hash aggregations with large values/aggregates
### What happens?

We see persistent out of memory issues with certain hash aggregations and hash joins for slightly bigger data sets, which also reproduce on surprisingly simple queries.

The problem occurs if there are many groups and the values/aggregates, rather than the keys, are of non-trivial size (strings, geometries, etc.). The spilling to disk seems no longer effective in avoiding memory_limit.

Possibly by design, but it's a very common scenario. 

### To Reproduce

Example with 100M MD5 values, distinct on, Parquet:
```sql
copy (select s a, md5(s::text) b from generate_series(1,100_000_000) as g(s)) to '/tmp/test.parquet';
set memory_limit to '4GB';
select distinct on (a) b from '/tmp/test.parquet' limit 10;
```
```consoke
Out of Memory Error: could not allocate block of size 256.0 KiB (3.7 GiB/3.7 GiB used)
```

Example with 1M long string values, group by, DuckDB table:
```sql
create table tbl as select s a, repeat('#', 10_000) b from generate_series(1,1_000_000) as g(s);
set memory_limit to '4GB';
select a, max(b) from tbl group by a limit 10;
```
```console
Out of Memory Error: could not allocate block of size 256.0 KiB (3.7 GiB/3.7 GiB used)
```

Large key small value, e.g. `select distinct on (b) a from '/tmp/test.parquet' limit 10;` works fine.

Using a persistent database with max_temp_directory_size 400GB.

Also noted in https://news.ycombinator.com/item?id=40645100

### OS:

Ubuntu 22.04 x86_64

### DuckDB Version:

1.1.1

### DuckDB Client:

CLI

### Hardware:

_No response_

### Full Name:

Marco Slot

### Affiliation:

Crunchy Data

### What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a stable release

### Did you include all relevant data sets for reproducing the issue?

No - I cannot share the data sets because they are confidential

### 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:

Fix patch:
diff --git a/extension/json/json_scan.cpp b/extension/json/json_scan.cpp
index 761e4a5e63b9..c5dee4b1aaa5 100644
--- a/extension/json/json_scan.cpp
+++ b/extension/json/json_scan.cpp
@@ -144,8 +144,7 @@ string JSONScanData::GetTimestampFormat() const {
 }
 
 JSONScanGlobalState::JSONScanGlobalState(ClientContext &context, const JSONScanData &bind_data_p)
-    : bind_data(bind_data_p), transform_options(bind_data.transform_options),
-      allocator(BufferManager::GetBufferManager(context).GetBufferAllocator()),
+    : bind_data(bind_data_p), transform_options(bind_data.transform_options), allocator(BufferAllocator::Get(context)),
       buffer_capacity(bind_data.maximum_object_size * 2), file_index(0), batch_index(0),
       system_threads(TaskScheduler::GetScheduler(context).NumberOfThreads()),
       enable_parallel_scans(bind_data.files.size() < system_threads) {
diff --git a/src/common/radix_partitioning.cpp b/src/common/radix_partitioning.cpp
index 3e8dee30d59f..d62a1329a44a 100644
--- a/src/common/radix_partitioning.cpp
+++ b/src/common/radix_partitioning.cpp
@@ -112,6 +112,7 @@ RadixPartitionedColumnData::RadixPartitionedColumnData(ClientContext &context_p,
 	allocators->allocators.reserve(num_partitions);
 	for (idx_t i = 0; i < num_partitions; i++) {
 		CreateAllocator();
+		allocators->allocators.back()->SetPartitionIndex(i);
 	}
 	D_ASSERT(allocators->allocators.size() == num_partitions);
 }
@@ -174,8 +175,10 @@ RadixPartitionedTupleData::~RadixPartitionedTupleData() {
 }
 
 void RadixPartitionedTupleData::Initialize() {
-	for (idx_t i = 0; i < RadixPartitioning::NumberOfPartitions(radix_bits); i++) {
+	const auto num_partitions = RadixPartitioning::NumberOfPartitions(radix_bits);
+	for (idx_t i = 0; i < num_partitions; i++) {
 		partitions.emplace_back(CreatePartitionCollection(i));
+		partitions.back()->SetPartitionIndex(i);
 	}
 }
 
diff --git a/src/common/types/column/column_data_allocator.cpp b/src/common/types/column/column_data_allocator.cpp
index bec0751e0d6c..66a1e612f2b3 100644
--- a/src/common/types/column/column_data_allocator.cpp
+++ b/src/common/types/column/column_data_allocator.cpp
@@ -1,5 +1,6 @@
 #include "duckdb/common/types/column/column_data_allocator.hpp"
 
+#include "duckdb/common/radix_partitioning.hpp"
 #include "duckdb/common/types/column/column_data_collection_segment.hpp"
 #include "duckdb/storage/buffer/block_handle.hpp"
 #include "duckdb/storage/buffer/buffer_pool.hpp"
@@ -84,6 +85,9 @@ BufferHandle ColumnDataAllocator::AllocateBlock(idx_t size) {
 	auto pin = alloc.buffer_manager->Allocate(MemoryTag::COLUMN_DATA, max_size, false);
 	data.handle = pin.GetBlockHandle();
 	blocks.push_back(std::move(data));
+	if (partition_index.IsValid()) { // Set the eviction queue index logarithmically using RadixBits
+		blocks.back().handle->SetEvictionQueueIndex(RadixPartitioning::RadixBits(partition_index.GetIndex()));
+	}
 	allocated_size += max_size;
 	return pin;
 }
diff --git a/src/common/types/column/column_data_collection.cpp b/src/common/types/column/column_data_collection.cpp
index 41626d470633..dd3439e5523a 100644
--- a/src/common/types/column/column_data_collection.cpp
+++ b/src/common/types/column/column_data_collection.cpp
@@ -119,6 +119,13 @@ idx_t ColumnDataCollection::AllocationSize() const {
 	return total_size;
 }
 
+void ColumnDataCollection::SetPartitionIndex(const idx_t index) {
+	D_ASSERT(!partition_index.IsValid());
+	D_ASSERT(Count() == 0);
+	partition_index = index;
+	allocator->SetPartitionIndex(index);
+}
+
 //===--------------------------------------------------------------------===//
 // ColumnDataRow
 //===--------------------------------------------------------------------===//
diff --git a/src/common/types/row/partitioned_tuple_data.cpp b/src/common/types/row/partitioned_tuple_data.cpp
index 17cd306f4045..b77463d8cf06 100644
--- a/src/common/types/row/partitioned_tuple_data.cpp
+++ b/src/common/types/row/partitioned_tuple_data.cpp
@@ -262,15 +262,8 @@ void PartitionedTupleData::Repartition(PartitionedTupleData &new_partitioned_dat
 	PartitionedTupleDataAppendState append_state;
 	new_partitioned_data.InitializeAppendState(append_state);
 
-	const auto reverse = RepartitionReverseOrder();
-	const idx_t start_idx = reverse ? partitions.size() : 0;
-	const idx_t end_idx = reverse ? 0 : partitions.size();
-	const int64_t update = reverse ? -1 : 1;
-	const int64_t adjustment = reverse ? -1 : 0;
-
-	for (idx_t partition_idx = start_idx; partition_idx != end_idx; partition_idx += idx_t(update)) {
-		auto actual_partition_idx = partition_idx + idx_t(adjustment);
-		auto &partition = *partitions[actual_partition_idx];
+	for (idx_t partition_idx = 0; partition_idx < partitions.size(); partition_idx++) {
+		auto &partition = *partitions[partition_idx];
 
 		if (partition.Count() > 0) {
 			TupleDataChunkIterator iterator(partition, TupleDataPinProperties::DESTROY_AFTER_DONE, true);
@@ -279,9 +272,9 @@ void PartitionedTupleData::Repartition(PartitionedTupleData &new_partitioned_dat
 				new_partitioned_data.Append(append_state, chunk_state, iterator.GetCurrentChunkCount());
 			} while (iterator.Next());
 
-			RepartitionFinalizeStates(*this, new_partitioned_data, append_state, actual_partition_idx);
+			RepartitionFinalizeStates(*this, new_partitioned_data, append_state, partition_idx);
 		}
-		partitions[actual_partition_idx]->Reset();
+		partitions[partition_idx]->Reset();
 	}
 	new_partitioned_data.FlushAppendState(append_state);
 
diff --git a/src/common/types/row/tuple_data_allocator.cpp b/src/common/types/row/tuple_data_allocator.cpp
index 8f391c4983aa..11895e977d61 100644
--- a/src/common/types/row/tuple_data_allocator.cpp
+++ b/src/common/types/row/tuple_data_allocator.cpp
@@ -1,6 +1,7 @@
 #include "duckdb/common/types/row/tuple_data_allocator.hpp"
 
 #include "duckdb/common/fast_mem.hpp"
+#include "duckdb/common/radix_partitioning.hpp"
 #include "duckdb/common/types/row/tuple_data_segment.hpp"
 #include "duckdb/common/types/row/tuple_data_states.hpp"
 #include "duckdb/storage/buffer/block_handle.hpp"
@@ -73,6 +74,12 @@ idx_t TupleDataAllocator::HeapBlockCount() const {
 	return heap_blocks.size();
 }
 
+void TupleDataAllocator::SetPartitionIndex(const idx_t index) {
+	D_ASSERT(!partition_index.IsValid());
+	D_ASSERT(row_blocks.empty() && heap_blocks.empty());
+	partition_index = index;
+}
+
 void TupleDataAllocator::Build(TupleDataSegment &segment, TupleDataPinState &pin_state,
                                TupleDataChunkState &chunk_state, const idx_t append_offset, const idx_t append_count) {
 	D_ASSERT(this == segment.allocator.get());
@@ -142,6 +149,9 @@ TupleDataChunkPart TupleDataAllocator::BuildChunkPart(TupleDataPinState &pin_sta
 	// Allocate row block (if needed)
 	if (row_blocks.empty() || row_blocks.back().RemainingCapacity() < layout.GetRowWidth()) {
 		row_blocks.emplace_back(buffer_manager, block_size);
+		if (partition_index.IsValid()) { // Set the eviction queue index logarithmically using RadixBits
+			row_blocks.back().handle->SetEvictionQueueIndex(RadixPartitioning::RadixBits(partition_index.GetIndex()));
+		}
 	}
 	result.row_block_index = NumericCast<uint32_t>(row_blocks.size() - 1);
 	auto &row_block = row_blocks[result.row_block_index];
@@ -188,6 +198,10 @@ TupleDataChunkPart TupleDataAllocator::BuildChunkPart(TupleDataPinState &pin_sta
 				if (heap_blocks.empty() || heap_blocks.back().RemainingCapacity() < heap_sizes[append_offset]) {
 					const auto size = MaxValue<idx_t>(block_size, heap_sizes[append_offset]);
 					heap_blocks.emplace_back(buffer_manager, size);
+					if (partition_index.IsValid()) { // Set the eviction queue index logarithmically using RadixBits
+						heap_blocks.back().handle->SetEvictionQueueIndex(
+						    RadixPartitioning::RadixBits(partition_index.GetIndex()));
+					}
 				}
 				result.heap_block_index = NumericCast<uint32_t>(heap_blocks.size() - 1);
 				auto &heap_block = heap_blocks[result.heap_block_index];
diff --git a/src/common/types/row/tuple_data_collection.cpp b/src/common/types/row/tuple_data_collection.cpp
index a5215d0302eb..c44732847e9d 100644
--- a/src/common/types/row/tuple_data_collection.cpp
+++ b/src/common/types/row/tuple_data_collection.cpp
@@ -79,6 +79,13 @@ void TupleDataCollection::Unpin() {
 	}
 }
 
+void TupleDataCollection::SetPartitionIndex(const idx_t index) {
+	D_ASSERT(!partition_index.IsValid());
+	D_ASSERT(Count() == 0);
+	partition_index = index;
+	allocator->SetPartitionIndex(index);
+}
+
 // LCOV_EXCL_START
 void VerifyAppendColumns(const TupleDataLayout &layout, const vector<column_t> &column_ids) {
 #ifdef DEBUG
diff --git a/src/execution/aggregate_hashtable.cpp b/src/execution/aggregate_hashtable.cpp
index e09fd9b70fad..077d846cd6d9 100644
--- a/src/execution/aggregate_hashtable.cpp
+++ b/src/execution/aggregate_hashtable.cpp
@@ -59,7 +59,8 @@ GroupedAggregateHashTable::GroupedAggregateHashTable(ClientContext &context, All
 }
 
 void GroupedAggregateHashTable::InitializePartitionedData() {
-	if (!partitioned_data || RadixPartitioning::RadixBits(partitioned_data->PartitionCount()) != radix_bits) {
+	if (!partitioned_data ||
+	    RadixPartitioning::RadixBitsOfPowerOfTwo(partitioned_data->PartitionCount()) != radix_bits) {
 		D_ASSERT(!partitioned_data || partitioned_data->Count() == 0);
 		partitioned_data =
 		    make_uniq<RadixPartitionedTupleData>(buffer_manager, layout, radix_bits, layout.ColumnCount() - 1);
@@ -131,7 +132,11 @@ idx_t GroupedAggregateHashTable::Capacity() const {
 }
 
 idx_t GroupedAggregateHashTable::ResizeThreshold() const {
-	return LossyNumericCast<idx_t>(static_cast<double>(Capacity()) / LOAD_FACTOR);
+	return ResizeThreshold(Capacity());
+}
+
+idx_t GroupedAggregateHashTable::ResizeThreshold(const idx_t capacity) {
+	return LossyNumericCast<idx_t>(static_cast<double>(capacity) / LOAD_FACTOR);
 }
 
 idx_t GroupedAggregateHashTable::ApplyBitMask(hash_t hash) const {
@@ -169,8 +174,8 @@ void GroupedAggregateHashTable::SetRadixBits(idx_t radix_bits_p) {
 void GroupedAggregateHashTable::Resize(idx_t size) {
 	D_ASSERT(size >= STANDARD_VECTOR_SIZE);
 	D_ASSERT(IsPowerOfTwo(size));
-	if (size < capacity) {
-		throw InternalException("Cannot downsize a hash table!");
+	if (Count() != 0 && size < capacity) {
+		throw InternalException("Cannot downsize a non-empty hash table!");
 	}
 
 	capacity = size;
diff --git a/src/execution/operator/join/physical_hash_join.cpp b/src/execution/operator/join/physical_hash_join.cpp
index 0283f79945af..a5fe270f0a9b 100644
--- a/src/execution/operator/join/physical_hash_join.cpp
+++ b/src/execution/operator/join/physical_hash_join.cpp
@@ -426,18 +426,11 @@ class HashJoinFinalizeEvent : public BasePipelineEvent {
 			    make_uniq<HashJoinFinalizeTask>(shared_from_this(), context, sink, 0U, chunk_count, false, sink.op));
 		} else {
 			// Parallel finalize
-			auto chunks_per_thread = MaxValue<idx_t>((chunk_count + num_threads - 1) / num_threads, 1);
-
-			idx_t chunk_idx = 0;
-			for (idx_t thread_idx = 0; thread_idx < num_threads; thread_idx++) {
-				auto chunk_idx_from = chunk_idx;
-				auto chunk_idx_to = MinValue<idx_t>(chunk_idx_from + chunks_per_thread, chunk_count);
-				finalize_tasks.push_back(make_uniq<HashJoinFinalizeTask>(shared_from_this(), context, sink,
-				                                                         chunk_idx_from, chunk_idx_to, true, sink.op));
-				chunk_idx = chunk_idx_to;
-				if (chunk_idx == chunk_count) {
-					break;
-				}
+			const idx_t chunks_per_task = context.config.verify_parallelism ? 1 : CHUNKS_PER_TASK;
+			for (idx_t chunk_idx = 0; chunk_idx < chunk_count; chunk_idx += chunks_per_task) {
+				auto chunk_idx_to = MinValue<idx_t>(chunk_idx + chunks_per_task, chunk_count);
+				finalize_tasks.push_back(make_uniq<HashJoinFinalizeTask>(shared_from_this(), context, sink, chunk_idx,
+				                                                         chunk_idx_to, true, sink.op));
 			}
 		}
 		SetTasks(std::move(finalize_tasks));
@@ -448,7 +441,8 @@ class HashJoinFinalizeEvent : public BasePipelineEvent {
 		sink.hash_table->finalized = true;
 	}
 
-	static constexpr const idx_t PARALLEL_CONSTRUCT_THRESHOLD = 1048576;
+	static constexpr idx_t PARALLEL_CONSTRUCT_THRESHOLD = 1048576;
+	static constexpr idx_t CHUNKS_PER_TASK = 64;
 };
 
 void HashJoinGlobalSinkState::ScheduleFinalize(Pipeline &pipeline, Event &event) {
diff --git a/src/execution/radix_partitioned_hashtable.cpp b/src/execution/radix_partitioned_hashtable.cpp
index cfd968100662..179f7fd67d92 100644
--- a/src/execution/radix_partitioned_hashtable.cpp
+++ b/src/execution/radix_partitioned_hashtable.cpp
@@ -119,8 +119,6 @@ struct RadixHTConfig {
 	static constexpr const idx_t MAXIMUM_INITIAL_SINK_RADIX_BITS = 3;
 	//! Maximum Sink radix bits (independent of threads)
 	static constexpr const idx_t MAXIMUM_FINAL_SINK_RADIX_BITS = 7;
-	//! By how many radix bits to increment if we go external
-	static constexpr const idx_t EXTERNAL_RADIX_BITS_INCREMENT = 3;
 
 	//! The global sink state
 	RadixHTGlobalSinkState &sink;
@@ -128,8 +126,6 @@ struct RadixHTConfig {
 	atomic<idx_t> sink_radix_bits;
 	//! Maximum Sink radix bits (set based on number of threads)
 	const idx_t maximum_sink_radix_bits;
-	//! Radix bits if we go external
-	const idx_t external_radix_bits;
 
 public:
 	//! Capacity of HTs during the Sink
@@ -153,6 +149,7 @@ class RadixHTGlobalSinkState : public GlobalSinkState {
 	ClientContext &context;
 	//! Temporary memory state for managing this hash table's memory usage
 	unique_ptr<TemporaryMemoryState> temporary_memory_state;
+	idx_t minimum_reservation;
 
 	//! The radix HT
 	const RadixPartitionedHashTable &radix_ht;
@@ -174,6 +171,7 @@ class RadixHTGlobalSinkState : public GlobalSinkState {
 	unique_ptr<PartitionedTupleData> uncombined_data;
 	//! Allocators used during the Sink/Finalize
 	vector<shared_ptr<ArenaAllocator>> stored_allocators;
+	idx_t stored_allocators_size;
 
 	//! Partitions that are finalized during GetData
 	vector<unique_ptr<AggregatePartition>> partitions;
@@ -192,8 +190,9 @@ RadixHTGlobalSinkState::RadixHTGlobalSinkState(ClientContext &context_p, const R
     : context(context_p), temporary_memory_state(TemporaryMemoryManager::Get(context).Register(context)),
       radix_ht(radix_ht_p), config(context, *this), finalized(false), external(false), active_threads(0),
       number_of_threads(NumericCast<idx_t>(TaskScheduler::GetScheduler(context).NumberOfThreads())),
-      any_combined(false), finalize_done(0), scan_pin_properties(TupleDataPinProperties::DESTROY_AFTER_DONE),
-      count_before_combining(0), max_partition_size(0) {
+      any_combined(false), stored_allocators_size(0), finalize_done(0),
+      scan_pin_properties(TupleDataPinProperties::DESTROY_AFTER_DONE), count_before_combining(0),
+      max_partition_size(0) {
 
 	// Compute minimum reservation
 	auto block_alloc_size = BufferManager::GetBufferManager(context).GetBlockAllocSize();
@@ -210,7 +209,7 @@ RadixHTGlobalSinkState::RadixHTGlobalSinkState(ClientContext &context_p, const R
 
 	// This really is the minimum reservation that we can do
 	auto num_threads = NumericCast<idx_t>(TaskScheduler::GetScheduler(context).NumberOfThreads());
-	auto minimum_reservation = num_threads * ht_size;
+	minimum_reservation = num_threads * ht_size;
 
 	temporary_memory_state->SetMinimumReservation(minimum_reservation);
 	temporary_memory_state->SetRemainingSizeAndUpdateReservation(context, minimum_reservation);
@@ -253,8 +252,7 @@ void RadixHTGlobalSinkState::Destroy() {
 
 RadixHTConfig::RadixHTConfig(ClientContext &context, RadixHTGlobalSinkState &sink_p)
     : sink(sink_p), sink_radix_bits(InitialSinkRadixBits(context)),
-      maximum_sink_radix_bits(MaximumSinkRadixBits(context)),
-      external_radix_bits(ExternalRadixBits(maximum_sink_radix_bits)), sink_capacity(SinkCapacity(context)) {
+      maximum_sink_radix_bits(MaximumSinkRadixBits(context)), sink_capacity(SinkCapacity(context)) {
 }
 
 void RadixHTConfig::SetRadixBits(idx_t radix_bits_p) {
@@ -262,7 +260,7 @@ void RadixHTConfig::SetRadixBits(idx_t radix_bits_p) {
 }
 
 bool RadixHTConfig::SetRadixBitsToExternal() {
-	SetRadixBitsInternal(external_radix_bits, true);
+	SetRadixBitsInternal(MAXIMUM_FINAL_SINK_RADIX_BITS, true);
 	return sink.external;
 }
 
@@ -284,21 +282,18 @@ void RadixHTConfig::SetRadixBitsInternal(const idx_t radix_bits_p, bool external
 		sink.external = true;
 	}
 	sink_radix_bits = radix_bits_p;
-	return;
 }
 
 idx_t RadixHTConfig::InitialSinkRadixBits(ClientContext &context) {
 	const auto active_threads = NumericCast<idx_t>(TaskScheduler::GetScheduler(context).NumberOfThreads());
-	return MinValue(RadixPartitioning::RadixBits(NextPowerOfTwo(active_threads)), MAXIMUM_INITIAL_SINK_RADIX_BITS);
+	return MinValue(RadixPartitioning::RadixBitsOfPowerOfTwo(NextPowerOfTwo(active_threads)),
+	                MAXIMUM_INITIAL_SINK_RADIX_BITS);
 }
 
 idx_t RadixHTConfig::MaximumSinkRadixBits(ClientContext &context) {
 	const auto active_threads = NumericCast<idx_t>(TaskScheduler::GetScheduler(context).NumberOfThreads());
-	return MinValue(RadixPartitioning::RadixBits(NextPowerOfTwo(active_threads)), MAXIMUM_FINAL_SINK_RADIX_BITS);
-}
-
-idx_t RadixHTConfig::ExternalRadixBits(const idx_t &maximum_sink_radix_bits_p) {
-	return MinValue(maximum_sink_radix_bits_p + EXTERNAL_RADIX_BITS_INCREMENT, MAXIMUM_FINAL_SINK_RADIX_BITS);
+	return MinValue(RadixPartitioning::RadixBitsOfPowerOfTwo(NextPowerOfTwo(active_threads)),
+	                MAXIMUM_FINAL_SINK_RADIX_BITS);
 }
 
 idx_t RadixHTConfig::SinkCapacity(ClientContext &context) {
@@ -370,7 +365,9 @@ bool MaybeRepartition(ClientContext &context, RadixHTGlobalSinkState &gstate, Ra
 
 	// Check if we're approaching the memory limit
 	auto &temporary_memory_state = *gstate.temporary_memory_state;
-	const auto total_size = partitioned_data->SizeInBytes() + ht.Capacity() * sizeof(ht_entry_t);
+	const auto aggregate_allocator_size = ht.GetAggregateAllocator()->AllocationSize();
+	const auto total_size =
+	    aggregate_allocator_size + partitioned_data->SizeInBytes() + ht.Capacity() * sizeof(ht_entry_t);
 	idx_t thread_limit = temporary_memory_state.GetReservation() / gstate.number_of_threads;
 	if (total_size > thread_limit) {
 		// We're over the thread memory limit
@@ -379,7 +376,9 @@ bool MaybeRepartition(ClientContext &context, RadixHTGlobalSinkState &gstate, Ra
 			auto guard = gstate.Lock();
 			thread_limit = temporary_memory_state.GetReservation() / gstate.number_of_threads;
 			if (total_size > thread_limit) {
-				// Out-of-core would be triggered below, try to increase the reservation
+				// Out-of-core would be triggered below, update minimum reservation and try to increase the reservation
+				temporary_memory_state.SetMinimumReservation(aggregate_allocator_size * gstate.number_of_threads +
+				                                             gstate.minimum_reservation);
 				auto remaining_size =
 				    MaxValue<idx_t>(gstate.number_of_threads * total_size, temporary_memory_state.GetRemainingSize());
 				temporary_memory_state.SetRemainingSizeAndUpdateReservation(context, 2 * remaining_size);
@@ -411,7 +410,7 @@ bool MaybeRepartition(ClientContext &context, RadixHTGlobalSinkState &gstate, Ra
 	}
 
 	const auto partition_count = partitioned_data->PartitionCount();
-	const auto current_radix_bits = RadixPartitioning::RadixBits(partition_count);
+	const auto current_radix_bits = RadixPartitioning::RadixBitsOfPowerOfTwo(partition_count);
 	D_ASSERT(current_radix_bits <= config.GetRadixBits());
 
 	const auto block_size = BufferManager::GetBufferManager(context).GetBlockSize();
@@ -441,7 +440,8 @@ void RadixPartitionedHashTable::Sink(ExecutionContext &context, DataChunk &chunk
 	auto &gstate = input.global_state.Cast<RadixHTGlobalSinkState>();
 	auto &lstate = input.local_state.Cast<RadixHTLocalSinkState>();
 	if (!lstate.ht) {
-		lstate.ht = CreateHT(context.client, gstate.config.sink_capacity, gstate.config.GetRadixBits());
+		lstate.ht =
+		    CreateHT(context.client, GroupedAggregateHashTable::InitialCapacity(), gstate.config.GetRadixBits());
 		gstate.active_threads++;
 	}
 
@@ -451,11 +451,11 @@ void RadixPartitionedHashTable::Sink(ExecutionContext &context, DataChunk &chunk
 	auto &ht = *lstate.ht;
 	ht.AddChunk(group_chunk, payload_input, filter);
 
-	if (ht.Count() + STANDARD_VECTOR_SIZE < ht.ResizeThreshold()) {
+	if (ht.Count() + STANDARD_VECTOR_SIZE < GroupedAggregateHashTable::ResizeThreshold(gstate.config.sink_capacity)) {
 		return; // We can fit another chunk
 	}
 
-	if (gstate.number_of_threads > 2) {
+	if (gstate.number_of_threads > 2 || gstate.external) {
 		// 'Reset' the HT without taking its data, we can just keep appending to the same collection
 		// This only works because we never resize the HT
 		ht.ClearPointerTable();
@@ -470,6 +470,9 @@ void RadixPartitionedHashTable::Sink(ExecutionContext &context, DataChunk &chunk
 		// We repartitioned, but we didn't clear the pointer table / reset the count because we're on 1 or 2 threads
 		ht.ClearPointerTable();
 		ht.ResetCount();
+		if (gstate.external) {
+			ht.Resize(gstate.config.sink_capacity);
+		}
 	}
 
 	// TODO: combine early and often
@@ -507,6 +510,7 @@ void RadixPartitionedHashTable::Combine(ExecutionContext &context, GlobalSinkSta
 		gstate.uncombined_data = std::move(lstate.abandoned_data);
 	}
 	gstate.stored_allocators.emplace_back(ht.GetAggregateAllocator());
+	gstate.stored_allocators_size += gstate.stored_allocators.back()->AllocationSize();
 }
 
 void RadixPartitionedHashTable::Finalize(ClientContext &context, GlobalSinkState &gstate_p) const {
@@ -541,7 +545,7 @@ void RadixPartitionedHashTable::Finalize(ClientContext &context, GlobalSinkState
 	}
 
 	// Minimum of combining one partition at a time
-	gstate.temporary_memory_state->SetMinimumReservation(gstate.max_partition_size);
+	gstate.temporary_memory_state->SetMinimumReservation(gstate.stored_allocators_size + gstate.max_partition_size);
 	// Set size to 0 until the scan actually starts
 	gstate.temporary_memory_state->SetZero();
 	gstate.finalized = true;
@@ -558,12 +562,15 @@ idx_t RadixPartitionedHashTable::MaxThreads(GlobalSinkState &sink_p) const {
 
 	const auto max_threads = MinValue<idx_t>(
 	    NumericCast<idx_t>(TaskScheduler::GetScheduler(sink.context).NumberOfThreads()), sink.partitions.size());
-	sink.temporary_memory_state->SetRemainingSizeAndUpdateReservation(sink.context,
-	                                                                  max_threads * sink.max_partition_size);
+	sink.temporary_memory_state->SetRemainingSizeAndUpdateReservation(
+	    sink.context, sink.stored_allocators_size + max_threads * sink.max_partition_size);
 
+	// we cannot spill aggregate state memory
+	const auto usable_memory = sink.temporary_memory_state->GetReservation() > sink.stored_allocators_size
+	                               ? sink.temporary_memory_state->GetReservation() - sink.max_partition_size
+	                               : 0;
 	// This many partitions will fit given our reservation (at least 1))
-	const auto partitions_fit =
-	    MaxValue<idx_t>(sink.temporary_memory_state->GetReservation() / sink.max_partition_size, 1);
+	const auto partitions_fit = MaxValue<idx_t>(usable_memory / sink.max_partition_size, 1);
 
 	// Mininum of the two
 	return MinValue<idx_t>(partitions_fit, max_threads);
diff --git a/src/function/aggregate/distributive/first.cpp b/src/function/aggregate/distributive/first.cpp
index 86d3e4bcb3e9..cbd63116f677 100644
--- a/src/function/aggregate/distributive/first.cpp
+++ b/src/function/aggregate/distributive/first.cpp
@@ -1,7 +1,7 @@
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/vector_operations/vector_operations.hpp"
-#include "duckdb/function/create_sort_key.hpp"
 #include "duckdb/function/aggregate/distributive_functions.hpp"
+#include "duckdb/function/create_sort_key.hpp"
 #include "duckdb/planner/expression.hpp"
 
 namespace duckdb {
@@ -68,7 +68,7 @@ struct FirstFunction : public FirstFunctionBase {
 
 template <bool LAST, bool SKIP_NULLS>
 struct FirstFunctionStringBase : public FirstFunctionBase {
-	template <class STATE>
+	template <class STATE, bool COMBINE = false>
 	static void SetValue(STATE &state, AggregateInputData &input_data, string_t value, bool is_null) {
 		if (LAST && state.is_set) {
 			Destroy(state, input_data);
@@ -81,7 +81,9 @@ struct FirstFunctionStringBase : public FirstFunctionBase {
 		} else {
 			state.is_set = true;
 			state.is_null = false;
-			if (value.IsInlined()) {
+			if ((COMBINE && !LAST) || value.IsInlined()) {
+				// We use the aggregate allocator for 'first', so the allocation is already done when combining
+				// Of course, if the value is inlined, we also don't need to allocate
 				state.value = value;
 			} else {
 				// non-inlined string, need to allocate space for it
@@ -97,7 +99,7 @@ struct FirstFunctionStringBase : public FirstFunctionBase {
 	template <class STATE, class OP>
 	static void Combine(const STATE &source, STATE &target, AggregateInputData &input_data) {
 		if (source.is_set && (LAST || !target.is_set)) {
-			SetValue(target, input_data, source.value, source.is_null);
+			SetValue<STATE, true>(target, input_data, source.value, source.is_null);
 		}
 	}
 
diff --git a/src/include/duckdb/common/file_buffer.hpp b/src/include/duckdb/common/file_buffer.hpp
index 1a3e6e9cb081..f32a2c416972 100644
--- a/src/include/duckdb/common/file_buffer.hpp
+++ b/src/include/duckdb/common/file_buffer.hpp
@@ -17,7 +17,7 @@ struct FileHandle;
 
 enum class FileBufferType : uint8_t { BLOCK = 1, MANAGED_BUFFER = 2, TINY_BUFFER = 3 };
 
-static constexpr const idx_t FILE_BUFFER_TYPE_COUNT = 3;
+static constexpr idx_t FILE_BUFFER_TYPE_COUNT = 3;
 
 //! The FileBuffer represents a buffer that can be read or written to a Direct IO FileHandle.
 class FileBuffer {
diff --git a/src/include/duckdb/common/radix_partitioning.hpp b/src/include/duckdb/common/radix_partitioning.hpp
index aa5efc2860f1..66b74ef7a4a2 100644
--- a/src/include/duckdb/common/radix_partitioning.hpp
+++ b/src/include/duckdb/common/radix_partitioning.hpp
@@ -8,7 +8,7 @@
 
 #pragma once
 
-#include "duckdb/common/fast_mem.hpp"
+#include "duckdb/common/bit_utils.hpp"
 #include "duckdb/common/types/column/partitioned_column_data.hpp"
 #include "duckdb/common/types/row/partitioned_tuple_data.hpp"
 
@@ -30,15 +30,15 @@ struct RadixPartitioning {
 		return idx_t(1) << radix_bits;
 	}
 
+	template <class T>
+	static inline idx_t RadixBits(T n) {
+		return sizeof(T) * 8 - CountZeros<T>::Leading(n);
+	}
+
 	//! Inverse of NumberOfPartitions, given a number of partitions, get the number of radix bits
-	static inline idx_t RadixBits(idx_t n_partitions) {
+	static inline idx_t RadixBitsOfPowerOfTwo(idx_t n_partitions) {
 		D_ASSERT(IsPowerOfTwo(n_partitions));
-		for (idx_t r = 0; r < sizeof(idx_t) * 8; r++) {
-			if (n_partitions == NumberOfPartitions(r)) {
-				return r;
-			}
-		}
-		throw InternalException("RadixPartitioning::RadixBits unable to find partition count!");
+		return RadixBits(n_partitions) - 1;
 	}
 
 	//! Radix bits begin after uint16_t because these bits are used as salt in the aggregate HT
@@ -132,9 +132,6 @@ class RadixPartitionedTupleData : public PartitionedTupleData {
 		return RadixPartitioning::NumberOfPartitions(radix_bits) - 1;
 	}
 
-	bool RepartitionReverseOrder() const override {
-		return true;
-	}
 	void RepartitionFinalizeStates(PartitionedTupleData &old_partitioned_data,
 	                               PartitionedTupleData &new_partitioned_data, PartitionedTupleDataAppendState &state,
 	                               idx_t finished_partition_idx) const override;
diff --git a/src/include/duckdb/common/types/column/column_data_allocator.hpp b/src/include/duckdb/common/types/column/column_data_allocator.hpp
index 194b40ca3ea1..38a29532013d 100644
--- a/src/include/duckdb/common/types/column/column_data_allocator.hpp
+++ b/src/include/duckdb/common/types/column/column_data_allocator.hpp
@@ -62,6 +62,12 @@ class ColumnDataAllocator {
 	idx_t AllocationSize() const {
 		return allocated_size;
 	}
+	//! Sets the partition index of this tuple data collection
+	void SetPartitionIndex(idx_t index) {
+		D_ASSERT(!partition_index.IsValid());
+		D_ASSERT(blocks.empty() && allocated_data.empty());
+		partition_index = index;
+	}
 
 public:
 	void AllocateData(idx_t size, uint32_t &block_id, uint32_t &offset, ChunkManagementState *chunk_state);
@@ -107,6 +113,8 @@ class ColumnDataAllocator {
 	mutex lock;
 	//! Total allocated size
 	idx_t allocated_size = 0;
+	//! Partition index (optional, if partitioned)
+	optional_idx partition_index;
 };
 
 } // namespace duckdb
diff --git a/src/include/duckdb/common/types/column/column_data_collection.hpp b/src/include/duckdb/common/types/column/column_data_collection.hpp
index 571826617a3d..f02d4900169a 100644
--- a/src/include/duckdb/common/types/column/column_data_collection.hpp
+++ b/src/include/duckdb/common/types/column/column_data_collection.hpp
@@ -65,6 +65,8 @@ class ColumnDataCollection {
 	idx_t SizeInBytes() const;
 	//! The allocation size (in bytes) of this ColumnDataCollection - this property is cached
 	idx_t AllocationSize() const;
+	//! Sets the partition index of this ColumnDataCollection
+	void SetPartitionIndex(idx_t index);
 
 	//! Get the allocator
 	DUCKDB_API Allocator &GetAllocator() const;
@@ -185,6 +187,8 @@ class ColumnDataCollection {
 	vector<ColumnDataCopyFunction> copy_functions;
 	//! When the column data collection is marked as finished - new tuples can no longer be appended to it
 	bool finished_append;
+	//! Partition index (optional, if partitioned)
+	optional_idx partition_index;
 };
 
 //! The ColumnDataRowCollection represents a set of materialized rows, as obtained from the ColumnDataCollection
diff --git a/src/include/duckdb/common/types/row/partitioned_tuple_data.hpp b/src/include/duckdb/common/types/row/partitioned_tuple_data.hpp
index 999c7218a31d..878b1bfa095d 100644
--- a/src/include/duckdb/common/types/row/partitioned_tuple_data.hpp
+++ b/src/include/duckdb/common/types/row/partitioned_tuple_data.hpp
@@ -153,10 +153,6 @@ class PartitionedTupleData {
 		return DConstants::INVALID_INDEX;
 	}
 
-	//! Whether or not to iterate over the original partitions in reverse order when repartitioning (optional)
-	virtual bool RepartitionReverseOrder() const {
-		return false;
-	}
 	//! Finalize states while repartitioning - useful for unpinning blocks that are no longer needed (optional)
 	virtual void RepartitionFinalizeStates(PartitionedTupleData &old_partitioned_data,
 	                                       PartitionedTupleData &new_partitioned_data,
diff --git a/src/include/duckdb/common/types/row/tuple_data_allocator.hpp b/src/include/duckdb/common/types/row/tuple_data_allocator.hpp
index 840d48602152..b68d3606b62c 100644
--- a/src/include/duckdb/common/types/row/tuple_data_allocator.hpp
+++ b/src/include/duckdb/common/types/row/tuple_data_allocator.hpp
@@ -67,6 +67,8 @@ class TupleDataAllocator {
 	idx_t RowBlockCount() const;
 	//! Number of heap blocks
 	idx_t HeapBlockCount() const;
+	//! Sets the partition index of this tuple data allocator
+	void SetPartitionIndex(idx_t index);
 
 public:
 	//! Builds out the chunks for next append, given the metadata in the append state
@@ -113,6 +115,8 @@ class TupleDataAllocator {
 	BufferManager &buffer_manager;
 	//! The layout of the data
 	const TupleDataLayout layout;
+	//! Partition index (optional, if partitioned)
+	optional_idx partition_index;
 	//! Blocks storing the fixed-size rows
 	unsafe_vector<TupleDataBlock> row_blocks;
 	//! Blocks storing the variable-size data of the fixed-size rows (e.g., string, list)
diff --git a/src/include/duckdb/common/types/row/tuple_data_collection.hpp b/src/include/duckdb/common/types/row/tuple_data_collection.hpp
index b87b4002401f..71b70092231f 100644
--- a/src/include/duckdb/common/types/row/tuple_data_collection.hpp
+++ b/src/include/duckdb/common/types/row/tuple_data_collection.hpp
@@ -66,6 +66,8 @@ class TupleDataCollection {
 	idx_t SizeInBytes() const;
 	//! Unpins all held pins
 	void Unpin();
+	//! Sets the partition index of this tuple data collection
+	void SetPartitionIndex(idx_t index);
 
 	//! Gets the scatter function for the given type
 	static TupleDataScatterFunction GetScatterFunction(const LogicalType &type, bool within_collection = false);
@@ -252,6 +254,8 @@ class TupleDataCollection {
 	vector<TupleDataScatterFunction> scatter_functions;
 	//! The set of gather functions
 	vector<TupleDataGatherFunction> gather_functions;
+	//! Partition index (optional, if partitioned)
+	optional_idx partition_index;
 };
 
 } // namespace duckdb
diff --git a/src/include/duckdb/execution/aggregate_hashtable.hpp b/src/include/duckdb/execution/aggregate_hashtable.hpp
index 2b424152f2c1..ac3c4a00ace0 100644
--- a/src/include/duckdb/execution/aggregate_hashtable.hpp
+++ b/src/include/duckdb/execution/aggregate_hashtable.hpp
@@ -57,6 +57,7 @@ class GroupedAggregateHashTable : public BaseAggregateHashTable {
 	idx_t Capacity() const;
 	//! Threshold at which to resize the HT
 	idx_t ResizeThreshold() const;
+	static idx_t ResizeThreshold(idx_t capacity);
 
 	//! Add the given data to the HT, computing the aggregates grouped by the
 	//! data in the group chunk. When resize = true, aggregates will not be
diff --git a/src/include/duckdb/storage/buffer/block_handle.hpp b/src/include/duckdb/storage/buffer/block_handle.hpp
index be1128faebad..1b07ff0f643c 100644
--- a/src/include/duckdb/storage/buffer/block_handle.hpp
+++ b/src/include/duckdb/storage/buffer/block_handle.hpp
@@ -15,6 +15,7 @@
 #include "duckdb/common/file_buffer.hpp"
 #include "duckdb/common/mutex.hpp"
 #include "duckdb/common/numeric_utils.hpp"
+#include "duckdb/common/optional_idx.hpp"
 #include "duckdb/storage/storage_info.hpp"
 
 namespace duckdb {
@@ -116,10 +117,17 @@ class BlockHandle : public enable_shared_from_this<BlockHandle> {
 	inline const idx_t &GetMemoryUsage() const {
 		return memory_usage;
 	}
+
 	bool IsUnloaded() {
 		return state == BlockState::BLOCK_UNLOADED;
 	}
 
+	void SetEvictionQueueIndex(const idx_t index) {
+		D_ASSERT(!eviction_queue_idx.IsValid());                  // Cannot overwrite
+		D_ASSERT(buffer->type == FileBufferType::MANAGED_BUFFER); // MANAGED_BUFFER only (at least, for now)
+		eviction_queue_idx = index;
+	}
+
 private:
 	BufferHandle Load(unique_ptr<FileBuffer> buffer = nullptr);
 	BufferHandle LoadFromBuffer(data_ptr_t data, unique_ptr<FileBuffer> reusable_buffer);
@@ -152,6 +160,8 @@ class BlockHandle : public enable_shared_from_this<BlockHandle> {
 	BufferPoolReservation memory_charge;
 	//! Does the block contain any memory pointers?
 	const char *unswizzled;
+	//! Index for eviction queue (FileBufferType::MANAGED_BUFFER only, for now)
+	optional_idx eviction_queue_idx;
 };
 
 } // namespace duckdb
diff --git a/src/include/duckdb/storage/buffer/buffer_pool.hpp b/src/include/duckdb/storage/buffer/buffer_pool.hpp
index 50166a51fa05..f4548e4305e2 100644
--- a/src/include/duckdb/storage/buffer/buffer_pool.hpp
+++ b/src/include/duckdb/storage/buffer/buffer_pool.hpp
@@ -82,14 +82,21 @@ class BufferPool {
 	idx_t PurgeAgedBlocks(uint32_t max_age_sec);
 	idx_t PurgeAgedBlocksInternal(EvictionQueue &queue, uint32_t max_age_sec, int64_t now, int64_t limit);
 	//! Garbage collect dead nodes in the eviction queue.
-	void PurgeQueue(FileBufferType type);
+	void PurgeQueue(const BlockHandle &handle);
 	//! Add a buffer handle to the eviction queue. Returns true, if the queue is
 	//! ready to be purged, and false otherwise.
 	bool AddToEvictionQueue(shared_ptr<BlockHandle> &handle);
 	//! Gets the eviction queue for the specified type
-	EvictionQueue &GetEvictionQueueForType(FileBufferType type);
+	EvictionQueue &GetEvictionQueueForBlockHandle(const BlockHandle &handle);
 	//! Increments the dead nodes for the queue with specified type
-	void IncrementDeadNodes(FileBufferType type);
+	void IncrementDeadNodes(const BlockHandle &handle);
+
+	//! How many eviction queues we have for the different FileBufferTypes
+	static constexpr idx_t BLOCK_QUEUE_SIZE = 1;
+	static constexpr idx_t MANAGED_BUFFER_QUEUE_SIZE = 6;
+	static constexpr idx_t TINY_BUFFER_QUEUE_SIZE = 1;
+	//! Mapping and priority order for the eviction queues
+	const array<idx_t, FILE_BUFFER_TYPE_COUNT> eviction_queue_sizes;
 
 protected:
 	enum class MemoryUsageCaches {
diff --git a/src/include/duckdb/storage/buffer_manager.hpp b/src/include/duckdb/storage/buffer_manager.hpp
index e2a3b95e0775..16708416e83f 100644
--- a/src/include/duckdb/storage/buffer_manager.hpp
+++ b/src/include/duckdb/storage/buffer_manager.hpp
@@ -98,7 +98,7 @@ class BufferManager {
 	virtual TemporaryMemoryManager &GetTemporaryMemoryManager();
 
 protected:
-	virtual void PurgeQueue(FileBufferType type) = 0;
+	virtual void PurgeQueue(const BlockHandle &handle) = 0;
 	virtual void AddToEvictionQueue(shared_ptr<BlockHandle> &handle);
 	virtual void WriteTemporaryBuffer(MemoryTag tag, block_id_t block_id, FileBuffer &buffer);
 	virtual unique_ptr<FileBuffer> ReadTemporaryBuffer(MemoryTag tag, BlockHandle &block,
diff --git a/src/include/duckdb/storage/standard_buffer_manager.hpp b/src/include/duckdb/storage/standard_buffer_manager.hpp
index e4de96089b6a..ef52365fa5ea 100644
--- a/src/include/duckdb/storage/standard_buffer_manager.hpp
+++ b/src/include/duckdb/storage/standard_buffer_manager.hpp
@@ -120,7 +120,7 @@ class StandardBufferManager : public BufferManager {
 	shared_ptr<BlockHandle> RegisterMemory(MemoryTag tag, idx_t block_size, bool can_destroy);
 
 	//! Garbage collect eviction queue
-	void PurgeQueue(FileBufferType type) final;
+	void PurgeQueue(const BlockHandle &handle) final;
 
 	BufferPool &GetBufferPool() const final;
 	TemporaryMemoryManager &GetTemporaryMemoryManager() final;
diff --git a/src/storage/buffer/block_handle.cpp b/src/storage/buffer/block_handle.cpp
index 9523b2962560..8a24834a808f 100644
--- a/src/storage/buffer/block_handle.cpp
+++ b/src/storage/buffer/block_handle.cpp
@@ -36,7 +36,7 @@ BlockHandle::~BlockHandle() { // NOLINT: allow internal exceptions
 	if (buffer && buffer->type != FileBufferType::TINY_BUFFER) {
 		// we kill the latest version in the eviction queue
 		auto &buffer_manager = block_manager.buffer_manager;
-		buffer_manager.GetBufferPool().IncrementDeadNodes(buffer->type);
+		buffer_manager.GetBufferPool().IncrementDeadNodes(*this);
 	}
 
 	// no references remain to this block: erase
diff --git a/src/storage/buffer/block_manager.cpp b/src/storage/buffer/block_manager.cpp
index 22cb54d13572..2f7839179aa4 100644
--- a/src/storage/buffer/block_manager.cpp
+++ b/src/storage/buffer/block_manager.cpp
@@ -64,7 +64,7 @@ shared_ptr<BlockHandle> BlockManager::ConvertToPersistent(block_id_t block_id, s
 	// potentially purge the queue
 	auto purge_queue = buffer_manager.GetBufferPool().AddToEvictionQueue(new_block);
 	if (purge_queue) {
-		buffer_manager.GetBufferPool().PurgeQueue(new_block->buffer->type);
+		buffer_manager.GetBufferPool().PurgeQueue(*new_block);
 	}
 
 	return new_block;
diff --git a/src/storage/buffer/buffer_pool.cpp b/src/storage/buffer/buffer_pool.cpp
index 9ed31f6f3262..607ef09db464 100644
--- a/src/storage/buffer/buffer_pool.cpp
+++ b/src/storage/buffer/buffer_pool.cpp
@@ -41,7 +41,8 @@ typedef duckdb_moodycamel::ConcurrentQueue<BufferEvictionNode> eviction_queue_t;
 
 struct EvictionQueue {
 public:
-	EvictionQueue() : evict_queue_insertions(0), total_dead_nodes(0) {
+	explicit EvictionQueue(const FileBufferType file_buffer_type_p)
+	    : file_buffer_type(file_buffer_type_p), evict_queue_insertions(0), total_dead_nodes(0) {
 	}
 
 public:
@@ -69,6 +70,8 @@ struct EvictionQueue {
 	void PurgeIteration(const idx_t purge_size);
 
 public:
+	//! The type of the buffers in this queue
+	const FileBufferType file_buffer_type;
 	//! The concurrent queue
 	eviction_queue_t q;
 
@@ -196,20 +199,24 @@ void EvictionQueue::PurgeIteration(const idx_t purge_size) {
 
 BufferPool::BufferPool(idx_t maximum_memory, bool track_eviction_timestamps,
                        idx_t allocator_bulk_deallocation_flush_threshold)
-    : maximum_memory(maximum_memory),
+    : eviction_queue_sizes({BLOCK_QUEUE_SIZE, MANAGED_BUFFER_QUEUE_SIZE, TINY_BUFFER_QUEUE_SIZE}),
+      maximum_memory(maximum_memory),
       allocator_bulk_deallocation_flush_threshold(allocator_bulk_deallocation_flush_threshold),
       track_eviction_timestamps(track_eviction_timestamps),
       temporary_memory_manager(make_uniq<TemporaryMemoryManager>()) {
-	queues.reserve(FILE_BUFFER_TYPE_COUNT);
-	for (idx_t i = 0; i < FILE_BUFFER_TYPE_COUNT; i++) {
-		queues.push_back(make_uniq<EvictionQueue>());
+	for (uint8_t type_idx = 0; type_idx < FILE_BUFFER_TYPE_COUNT; type_idx++) {
+		const auto type = static_cast<FileBufferType>(type_idx + 1);
+		const auto &type_queue_size = eviction_queue_sizes[type_idx];
+		for (idx_t queue_idx = 0; queue_idx < type_queue_size; queue_idx++) {
+			queues.push_back(make_uniq<EvictionQueue>(type));
+		}
 	}
 }
 BufferPool::~BufferPool() {
 }
 
 bool BufferPool::AddToEvictionQueue(shared_ptr<BlockHandle> &handle) {
-	auto &queue = GetEvictionQueueForType(handle->buffer->type);
+	auto &queue = GetEvictionQueueForBlockHandle(*handle);
 
 	// The block handle is locked during this operation (Unpin),
 	// or the block handle is still a local variable (ConvertToPersistent)
@@ -227,16 +234,36 @@ bool BufferPool::AddToEvictionQueue(shared_ptr<BlockHandle> &handle) {
 		queue.IncrementDeadNodes();
 	}
 
-	// Get the eviction queue for the buffer type and add it
+	// Get the eviction queue for the block and add it
 	return queue.AddToEvictionQueue(BufferEvictionNode(weak_ptr<BlockHandle>(handle), ts));
 }
 
-EvictionQueue &BufferPool::GetEvictionQueueForType(FileBufferType type) {
-	return *queues[uint8_t(type) - 1];
+EvictionQueue &BufferPool::GetEvictionQueueForBlockHandle(const BlockHandle &handle) {
+	const auto &handle_buffer_type = handle.buffer->type;
+
+	// Get offset into eviction queues for this FileBufferType
+	idx_t queue_index = 0;
+	for (uint8_t type_idx = 0; type_idx < FILE_BUFFER_TYPE_COUNT; type_idx++) {
+		const auto queue_buffer_type = static_cast<FileBufferType>(type_idx + 1);
+		if (handle_buffer_type == queue_buffer_type) {
+			break;
+		}
+		const auto &type_queue_size = eviction_queue_sizes[type_idx];
+		queue_index += type_queue_size;
+	}
+
+	const auto &queue_size = eviction_queue_sizes[static_cast<uint8_t>(handle_buffer_type) - 1];
+	// Adjust if eviction_queue_idx is set (idx == 0 -> add at back, idx >= queue_size -> add at front)
+	if (handle.eviction_queue_idx.IsValid() && handle.eviction_queue_idx.GetIndex() < queue_size) {
+		queue_index += queue_size - handle.eviction_queue_idx.GetIndex() - 1;
+	}
+
+	D_ASSERT(queues[queue_index]->file_buffer_type == handle_buffer_type);
+	return *queues[queue_index];
 }
 
-void BufferPool::IncrementDeadNodes(FileBufferType type) {
-	GetEvictionQueueForType(type).IncrementDeadNodes();
+void BufferPool::IncrementDeadNodes(const BlockHandle &handle) {
+	GetEvictionQueueForBlockHandle(handle).IncrementDeadNodes();
 }
 
 void BufferPool::UpdateUsedMemory(MemoryTag tag, int64_t size) {
@@ -261,23 +288,14 @@ TemporaryMemoryManager &BufferPool::GetTemporaryMemoryManager() {
 
 BufferPool::EvictionResult BufferPool::EvictBlocks(MemoryTag tag, idx_t extra_memory, idx_t memory_limit,
                                                    unique_ptr<FileBuffer> *buffer) {
-	// First, we try to evict persistent table data
-	auto block_result =
-	    EvictBlocksInternal(GetEvictionQueueForType(FileBufferType::BLOCK), tag, extra_memory, memory_limit, buffer);
-	if (block_result.success) {
-		return block_result;
-	}
-
-	// If that does not succeed, we try to evict temporary data
-	auto managed_buffer_result = EvictBlocksInternal(GetEvictionQueueForType(FileBufferType::MANAGED_BUFFER), tag,
-	                                                 extra_memory, memory_limit, buffer);
-	if (managed_buffer_result.success) {
-		return managed_buffer_result;
+	for (auto &queue : queues) {
+		auto block_result = EvictBlocksInternal(*queue, tag, extra_memory, memory_limit, buffer);
+		if (block_result.success || RefersToSameObject(*queue, *queues.back())) {
+			return block_result; // Return upon success or upon last queue
+		}
 	}
-
-	// Finally, we try to evict tiny buffers
-	return EvictBlocksInternal(GetEvictionQueueForType(FileBufferType::TINY_BUFFER), tag, extra_memory, memory_limit,
-	                           buffer);
+	// This can never happen since we always return when i == 1. Exception to silence compiler warning
+	throw InternalException("Exited BufferPool::EvictBlocksInternal without obtaining BufferPool::EvictionResult");
 }
 
 BufferPool::EvictionResult BufferPool::EvictBlocksInternal(EvictionQueue &queue, MemoryTag tag, idx_t extra_memory,
@@ -381,8 +399,8 @@ void EvictionQueue::IterateUnloadableBlocks(FN fn) {
 	}
 }
 
-void BufferPool::PurgeQueue(FileBufferType type) {
-	GetEvictionQueueForType(type).Purge();
+void BufferPool::PurgeQueue(const BlockHandle &block) {
+	GetEvictionQueueForBlockHandle(block).Purge();
 }
 
 void BufferPool::SetLimit(idx_t limit, const char *exception_postscript) {
diff --git a/src/storage/standard_buffer_manager.cpp b/src/storage/standard_buffer_manager.cpp
index da10afce297d..9b4c305d40a4 100644
--- a/src/storage/standard_buffer_manager.cpp
+++ b/src/storage/standard_buffer_manager.cpp
@@ -356,8 +356,8 @@ BufferHandle StandardBufferManager::Pin(shared_ptr<BlockHandle> &handle) {
 	return buf;
 }
 
-void StandardBufferManager::PurgeQueue(FileBufferType type) {
-	buffer_pool.PurgeQueue(type);
+void StandardBufferManager::PurgeQueue(const BlockHandle &handle) {
+	buffer_pool.PurgeQueue(handle);
 }
 
 void StandardBufferManager::AddToEvictionQueue(shared_ptr<BlockHandle> &handle) {
@@ -395,7 +395,7 @@ void StandardBufferManager::Unpin(shared_ptr<BlockHandle> &handle) {
 
 	// We do not have to keep the handle locked while purging.
 	if (purge) {
-		PurgeQueue(handle->buffer->type);
+		PurgeQueue(*handle);
 	}
 }
 


Test patch:
diff --git a/test/sql/aggregate/aggregates/first_memory_usage.test_slow b/test/sql/aggregate/aggregates/first_memory_usage.test_slow
new file mode 100644
index 000000000000..427c9f961ea5
--- /dev/null
+++ b/test/sql/aggregate/aggregates/first_memory_usage.test_slow
@@ -0,0 +1,14 @@
+# name: test/sql/aggregate/aggregates/first_memory_usage.test_slow
+# description: Issue 14132 - Out of memory on basic hash aggregations with large values/aggregates
+# group: [aggregates]
+
+load __TEST_DIR__/first_memory_usage.db
+
+statement ok
+set memory_limit='400mb';
+
+# this query uses the first() aggregate, which used to use too much memory (it did redundant allocation in Combine)
+# we also limit the number of threads in RadixPartitionedHashtable to limit memory usage when close to the limit
+# we can now easily complete this query
+statement ok
+select distinct on (a) b from (select s a, md5(s::text) b from generate_series(1,10_000_000) as g(s)) limit 10;