Repository: duckdb/duckdb
File: src/common/sort/hashed_sort.cpp (C++)

Code before:
#include "duckdb/common/sorting/hashed_sort.hpp"
#include "duckdb/common/sorting/sorted_run.hpp"
#include "duckdb/common/radix_partitioning.hpp"
#include "duckdb/parallel/thread_context.hpp"
#include "duckdb/planner/expression/bound_reference_expression.hpp"

namespace duckdb {

//===--------------------------------------------------------------------===//
// HashedSortGroup
//===--------------------------------------------------------------------===//
// Formerly PartitionGlobalHashGroup
class HashedSortGroup {
public:
	using Orders = vector<BoundOrderByNode>;
	using Types = vector<LogicalType>;

	HashedSortGroup(ClientContext &client, optional_ptr<Sort> sort, idx_t group_idx);

	const idx_t group_idx;
	atomic<idx_t> count;

	//	Sink
	optional_ptr<Sort> sort;
	unique_ptr<GlobalSinkState> sort_global;

	//	Source
	atomic<idx_t> tasks_completed;
	unique_ptr<GlobalSourceState> sort_source;

	// Unsorted
	unique_ptr<ColumnDataCollection> columns;
	atomic<idx_t> get_columns;
};

HashedSortGroup::HashedSortGroup(ClientContext &client, optional_ptr<Sort> sort, idx_t group_idx)
    : group_idx(group_idx), count(0), sort(sort), tasks_completed(0), get_columns(0) {
	if (sort) {
		sort_global = sort->GetGlobalSinkState(client);
	}
}

//===--------------------------------------------------------------------===//
// HashedSortGlobalSinkState
//===--------------------------------------------------------------------===//
// Formerly PartitionGlobalSinkState

Review comment (performance):
Not sure why this wouldn't be thread-safe, pretty much the first thing that happens in the parallel scan code is grabbing a lock. However, this isn't well tested, as this newly added scan is now the only usage of parallel tuple data scanning in the entire code base.

I'd expect this to work when scanning with `UNPIN_AFTER_DONE` (which is the default being used here), but not with `DESTROY_AFTER_DONE` (which was used before - now removed).

I think what should happen is that each `TupleDataCollection` should be scanned independently. After the `Sink` phase, each thread has a `PartitionedTupleData`, so we should be able to do fine-grained scheduling of small tasks (one per thread per partition), single-threaded scans on each `TupleDataCollection` that can be sunk into the `Sort`. This then also allows using `DESTROY_AFTER_DONE`.

I've checkout out this branch and ran #19499, and this now finishes the query! However, it is ~30% slower than 1.3.2, as it spills more data. I think this is due to data not being freed up quickly.

Code after:
#include "duckdb/common/sorting/hashed_sort.hpp"
#include "duckdb/common/sorting/sorted_run.hpp"
#include "duckdb/common/radix_partitioning.hpp"
#include "duckdb/parallel/thread_context.hpp"
#include "duckdb/planner/expression/bound_reference_expression.hpp"

namespace duckdb {

//===--------------------------------------------------------------------===//
// HashedSortGroup
//===--------------------------------------------------------------------===//
// Formerly PartitionGlobalHashGroup
class HashedSortGroup {
public:
	using Orders = vector<BoundOrderByNode>;
	using Types = vector<LogicalType>;

	HashedSortGroup(ClientContext &client, optional_ptr<Sort> sort, idx_t group_idx);

	bool Scan(TupleDataCollection &payload, TupleDataLocalScanState &local_scan, DataChunk &chunk) {
		//	Despite the name, TupleDataParallelScanState is not thread safe...
		lock_guard<mutex> guard(scan_lock);
		return payload.Scan(parallel_scan, local_scan, chunk);
	}

	const idx_t group_idx;
	atomic<idx_t> count;

	//	Sink
	optional_ptr<Sort> sort;
	unique_ptr<GlobalSinkState> sort_global;

	//	Source
	mutex scan_lock;
	TupleDataParallelScanState parallel_scan;
	atomic<idx_t> tasks_completed;
	unique_ptr<GlobalSourceState> sort_source;

	// Unsorted
	unique_ptr<ColumnDataCollection> columns;
	atomic<idx_t> get_columns;
};

HashedSortGroup::HashedSortGroup(ClientContext &client, optional_ptr<Sort> sort, idx_t group_idx)
    : group_idx(group_idx), count(0), sort(sort), tasks_completed(0), get_columns(0) {
	if (sort) {