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

Code before:
	case ExpressionType::WINDOW_FIRST_VALUE:
	case ExpressionType::WINDOW_PERCENT_RANK:
	case ExpressionType::WINDOW_RANK:
	case ExpressionType::WINDOW_RANK_DENSE:
	case ExpressionType::WINDOW_ROW_NUMBER:
		return true;
	case ExpressionType::WINDOW_LAG:
	case ExpressionType::WINDOW_LEAD: {
		// We can stream LEAD/LAG if the arguments are constant and the delta is less than a block behind
		Value dflt;
		int64_t offset;
		return StreamingWindowState::LeadLagState::ComputeDefault(context, wexpr, dflt) &&
		       StreamingWindowState::LeadLagState::ComputeOffset(context, wexpr, offset);
	}
	default:
		return false;
	}
}

unique_ptr<GlobalOperatorState> PhysicalStreamingWindow::GetGlobalOperatorState(ClientContext &context) const {
	return make_uniq<StreamingWindowGlobalState>();
}

unique_ptr<OperatorState> PhysicalStreamingWindow::GetOperatorState(ExecutionContext &context) const {
	auto &gstate = op_state->Cast<StreamingWindowGlobalState>();
	lock_guard<mutex> lock(gstate.local_lock);
	if (!gstate.local_state) {
		gstate.local_state = make_uniq<StreamingWindowState>(context.client);
	}
	return PhysicalOperator::GetOperatorState(context);
}

void StreamingWindowState::AggregateState::Execute(ExecutionContext &context, DataChunk &input, Vector &result) {
	//	Establish the aggregation environment
	const idx_t count = input.size();
	auto &aggregate = *wexpr.aggregate;
	auto &aggr_state = *this;
	auto &statev = aggr_state.statev;

	// Compute the FILTER mask (if any)
	ValidityMask filter_mask;
	auto filtered = count;
	auto &filter_sel = aggr_state.filter_sel;
	if (wexpr.filter_expr) {
		filtered = filter_executor.SelectExpression(input, filter_sel);
		if (filtered < count) {
			filter_mask.Initialize(count);
			filter_mask.SetAllInvalid(count);
			for (idx_t f = 0; f < filtered; ++f) {
				filter_mask.SetValid(filter_sel.get_index(f));
			}

Review comment (suggestion):
This seems unnecessarily complex - can we instead remove the `PhysicalStreamingWindow::GetOperatorState` method and create the `local_state` in the global state constructor? Then the lock wouldn't be necessary.

Code after:
		return wexpr.start == WindowBoundary::UNBOUNDED_PRECEDING && wexpr.end == WindowBoundary::CURRENT_ROW_ROWS;
	case ExpressionType::WINDOW_FIRST_VALUE:
	case ExpressionType::WINDOW_PERCENT_RANK:
	case ExpressionType::WINDOW_RANK:
	case ExpressionType::WINDOW_RANK_DENSE:
	case ExpressionType::WINDOW_ROW_NUMBER:
		return true;
	case ExpressionType::WINDOW_LAG:
	case ExpressionType::WINDOW_LEAD: {
		// We can stream LEAD/LAG if the arguments are constant and the delta is less than a block behind
		Value dflt;
		int64_t offset;
		return StreamingWindowState::LeadLagState::ComputeDefault(context, wexpr, dflt) &&
		       StreamingWindowState::LeadLagState::ComputeOffset(context, wexpr, offset);
	}
	default:
		return false;
	}
}

unique_ptr<GlobalOperatorState> PhysicalStreamingWindow::GetGlobalOperatorState(ClientContext &client) const {
	return make_uniq<StreamingWindowGlobalState>(client);
}

void StreamingWindowState::AggregateState::Execute(ExecutionContext &context, DataChunk &input, Vector &result) {
	//	Establish the aggregation environment
	const idx_t count = input.size();
	auto &aggregate = *wexpr.aggregate;
	auto &aggr_state = *this;
	auto &statev = aggr_state.statev;

	// Compute the FILTER mask (if any)
	ValidityMask filter_mask;
	auto filtered = count;
	auto &filter_sel = aggr_state.filter_sel;
	if (wexpr.filter_expr) {
		filtered = filter_executor.SelectExpression(input, filter_sel);
		if (filtered < count) {
			filter_mask.Initialize(count);
			filter_mask.SetAllInvalid(count);
			for (idx_t f = 0; f < filtered; ++f) {
				filter_mask.SetValid(filter_sel.get_index(f));
			}
		}
	}

	// Check for COUNT(*)
	if (wexpr.children.empty()) {
		D_ASSERT(GetTypeIdSize(result.GetType().InternalType()) == sizeof(int64_t));
		auto data = FlatVector::GetData<int64_t>(result);
		auto &unfiltered = aggr_state.unfiltered;