Repository: duckdb/duckdb
File: src/execution/operator/aggregate/physical_streaming_window.cpp (C++)

Code before:
#include "duckdb/parallel/thread_context.hpp"
#include "duckdb/planner/expression/bound_window_expression.hpp"

namespace duckdb {

PhysicalStreamingWindow::PhysicalStreamingWindow(PhysicalPlan &physical_plan, vector<LogicalType> types,
                                                 vector<unique_ptr<Expression>> select_list,
                                                 idx_t estimated_cardinality, PhysicalOperatorType type)
    : PhysicalOperator(physical_plan, type, std::move(types), estimated_cardinality),
      select_list(std::move(select_list)) {
}

class StreamingWindowGlobalState : public GlobalOperatorState {
public:
	StreamingWindowGlobalState() : row_number(1) {
	}

	//! The next row number.
	std::atomic<int64_t> row_number;
};

class StreamingWindowState : public OperatorState {
public:
	struct AggregateState {
		AggregateState(ClientContext &client, BoundWindowExpression &wexpr, Allocator &allocator)
		    : wexpr(wexpr), arena_allocator(Allocator::Get(client)), executor(client), filter_executor(client),
		      statev(LogicalType::POINTER, data_ptr_cast(&state_ptr)), hashes(LogicalType::HASH),
		      addresses(LogicalType::POINTER) {
			D_ASSERT(wexpr.GetExpressionType() == ExpressionType::WINDOW_AGGREGATE);
			auto &aggregate = *wexpr.aggregate;
			bind_data = wexpr.bind_info.get();
			dtor = aggregate.destructor;
			state.resize(aggregate.state_size(aggregate));
			state_ptr = state.data();
			aggregate.initialize(aggregate, state.data());
			for (auto &child : wexpr.children) {
				arg_types.push_back(child->return_type);
				executor.AddExpression(*child);
			}
			if (!arg_types.empty()) {
				arg_chunk.Initialize(allocator, arg_types);
				arg_cursor.Initialize(allocator, arg_types);
			}
			if (wexpr.filter_expr) {
				filter_executor.AddExpression(*wexpr.filter_expr);
				filter_sel.Initialize();
			}
			if (wexpr.distinct) {
				distinct = make_uniq<GroupedAggregateHashTable>(client, allocator, arg_types);
				distinct_args.Initialize(allocator, arg_types);
				distinct_sel.Initialize();

Review comment (suggestion):
Could we use `BufferAllocator::Get(client)` instead? Aggregate states can get pretty large, and this makes it count for the memory usage in `BufferPool`, so that pages will be evicted if we use too much memory.

Code after:
#include "duckdb/parallel/thread_context.hpp"
#include "duckdb/planner/expression/bound_window_expression.hpp"

namespace duckdb {

PhysicalStreamingWindow::PhysicalStreamingWindow(PhysicalPlan &physical_plan, vector<LogicalType> types,
                                                 vector<unique_ptr<Expression>> select_list,
                                                 idx_t estimated_cardinality, PhysicalOperatorType type)
    : PhysicalOperator(physical_plan, type, std::move(types), estimated_cardinality),
      select_list(std::move(select_list)) {
}

class StreamingWindowGlobalState : public GlobalOperatorState {
public:
	StreamingWindowGlobalState() : row_number(1) {
	}

	//! The next row number.
	std::atomic<int64_t> row_number;
};

class StreamingWindowState : public OperatorState {
public:
	struct AggregateState {
		AggregateState(ClientContext &client, BoundWindowExpression &wexpr, Allocator &allocator)
		    : wexpr(wexpr), arena_allocator(BufferAllocator::Get((client))), executor(client), filter_executor(client),
		      statev(LogicalType::POINTER, data_ptr_cast(&state_ptr)), hashes(LogicalType::HASH),
		      addresses(LogicalType::POINTER) {
			D_ASSERT(wexpr.GetExpressionType() == ExpressionType::WINDOW_AGGREGATE);
			auto &aggregate = *wexpr.aggregate;
			bind_data = wexpr.bind_info.get();
			dtor = aggregate.destructor;
			state.resize(aggregate.state_size(aggregate));
			state_ptr = state.data();
			aggregate.initialize(aggregate, state.data());
			for (auto &child : wexpr.children) {
				arg_types.push_back(child->return_type);
				executor.AddExpression(*child);
			}
			if (!arg_types.empty()) {
				arg_chunk.Initialize(allocator, arg_types);
				arg_cursor.Initialize(allocator, arg_types);
			}
			if (wexpr.filter_expr) {
				filter_executor.AddExpression(*wexpr.filter_expr);
				filter_sel.Initialize();
			}
			if (wexpr.distinct) {
				distinct = make_uniq<GroupedAggregateHashTable>(client, allocator, arg_types);
				distinct_args.Initialize(allocator, arg_types);
				distinct_sel.Initialize();