Repository: duckdb/duckdb (language: python)

Issue:
JDBC import database error
#### What happens?
JDBC import database error

#### To Reproduce
`@Test 
   public void importDuckDB() throws SQLException, IOException { 

       final String EXPORT_SQL_TEMPLATE = "EXPORT DATABASE '%s' (FORMAT PARQUET);"; 
       String path = "export"; 
       File export = new File(path); 
       if (export.exists()) { 
           export.delete(); 
       } 
       export.mkdir(); 
       System.out.println(export.getAbsolutePath()); 
       try (Connection conn = DriverManager.getConnection("jdbc:duckdb:db001"); Statement stmt = conn.createStatement();) { 
           stmt.execute("CREATE SCHEMA IF NOT EXISTS mydb"); 
           stmt.execute("CREATE TABLE IF NOT EXISTS mydb.student(id INTEGER,name VARCHAR,PRIMARY KEY(id))"); 
           stmt.execute(String.format(EXPORT_SQL_TEMPLATE, path)); 
       } 

       final String IMPORT_SQL_TEMPLATE = "IMPORT DATABASE '%s';"; 

       try (Connection conn = DriverManager.getConnection("jdbc:duckdb:db002"); Statement stmt = conn.createStatement();) { 
           stmt.execute(String.format(IMPORT_SQL_TEMPLATE, path)); 
       } 

   }

error info:
java.sql.SQLException: Cannot prepare multiple statements at once! 

     at org.duckdb.DuckDBNative.duckdb_jdbc_prepare(Native Method) 
     at org.duckdb.DuckDBPreparedStatement.prepare(DuckDBPreparedStatement.java:73) 
     at org.duckdb.DuckDBPreparedStatement.execute(DuckDBPreparedStatement.java:121)
`

#### Environment (please complete the following information):
 - OS: win10
 - DuckDB Version: 0.3.1
 - DuckDB Client: jdbc-0.3.1



Discussion / hints:
Thanks for using DuckDB! Does this same issue happen on version 0.3.2?


> Thanks for using DuckDB! Does this same issue happen on version 0.3.2?

yes,this same issue also  happen on version 0.3.2
Probably needs something similar than in #1102 and #3107.
In any case, you could try to execute the statements in `schema.sql` and `load.sql` (generated by `EXPORT` to the same folder as exported tables) manually (or in a loop) one by one.

Fix patch:
diff --git a/extension/core_functions/aggregate/holistic/mad.cpp b/extension/core_functions/aggregate/holistic/mad.cpp
index 94b7d7097f4a..03c07df3f38a 100644
--- a/extension/core_functions/aggregate/holistic/mad.cpp
+++ b/extension/core_functions/aggregate/holistic/mad.cpp
@@ -196,10 +196,7 @@ struct MedianAbsoluteDeviationOperation : QuantileOperation {
 		auto &state = *reinterpret_cast<STATE *>(l_state);
 		auto gstate = reinterpret_cast<const STATE *>(g_state);
 
-		D_ASSERT(partition.inputs);
-		const auto &inputs = *partition.inputs;
-		D_ASSERT(inputs.ColumnCount() == 1);
-		auto &data = state.GetOrCreateWindowCursor(inputs, partition.all_valid);
+		auto &data = state.GetOrCreateWindowCursor(partition);
 		const auto &fmask = partition.filter_mask;
 
 		auto rdata = FlatVector::GetData<RESULT_TYPE>(result);
diff --git a/extension/core_functions/aggregate/holistic/mode.cpp b/extension/core_functions/aggregate/holistic/mode.cpp
index 2c0de366c1f1..c0bb248ebbc7 100644
--- a/extension/core_functions/aggregate/holistic/mode.cpp
+++ b/extension/core_functions/aggregate/holistic/mode.cpp
@@ -121,11 +121,13 @@ struct ModeState {
 		}
 	}
 
-	void InitializePage(const ColumnDataCollection &inputs) {
+	void InitializePage(const WindowPartitionInput &partition) {
 		if (page.ColumnCount() == 0) {
-			this->inputs = &inputs;
-			inputs.InitializeScan(scan);
-			inputs.InitializeScanChunk(scan, page);
+			D_ASSERT(partition.inputs);
+			inputs = partition.inputs;
+			D_ASSERT(partition.column_ids.size() == 1);
+			inputs->InitializeScan(scan, partition.column_ids);
+			inputs->InitializeScanChunk(scan, page);
 		}
 	}
 
@@ -346,10 +348,7 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 	                   idx_t rid) {
 		auto &state = *reinterpret_cast<STATE *>(l_state);
 
-		D_ASSERT(partition.inputs);
-		const auto &inputs = *partition.inputs;
-		D_ASSERT(inputs.ColumnCount() == 1);
-		state.InitializePage(inputs);
+		state.InitializePage(partition);
 		const auto &fmask = partition.filter_mask;
 
 		auto rdata = FlatVector::GetData<RESULT_TYPE>(result);
diff --git a/extension/core_functions/aggregate/holistic/quantile.cpp b/extension/core_functions/aggregate/holistic/quantile.cpp
index 24caec942363..879be9637bf9 100644
--- a/extension/core_functions/aggregate/holistic/quantile.cpp
+++ b/extension/core_functions/aggregate/holistic/quantile.cpp
@@ -214,10 +214,7 @@ struct QuantileScalarOperation : public QuantileOperation {
 		auto &state = *reinterpret_cast<STATE *>(l_state);
 		auto gstate = reinterpret_cast<const STATE *>(g_state);
 
-		D_ASSERT(partition.inputs);
-		const auto &inputs = *partition.inputs;
-		D_ASSERT(inputs.ColumnCount() == 1);
-		auto &data = state.GetOrCreateWindowCursor(inputs, partition.all_valid);
+		auto &data = state.GetOrCreateWindowCursor(partition);
 		const auto &fmask = partition.filter_mask;
 
 		QuantileIncluded<INPUT_TYPE> included(fmask, data);
@@ -320,10 +317,7 @@ struct QuantileListOperation : QuantileOperation {
 		auto &state = *reinterpret_cast<STATE *>(l_state);
 		auto gstate = reinterpret_cast<const STATE *>(g_state);
 
-		D_ASSERT(partition.inputs);
-		const auto &inputs = *partition.inputs;
-		D_ASSERT(inputs.ColumnCount() == 1);
-		auto &data = state.GetOrCreateWindowCursor(inputs, partition.all_valid);
+		auto &data = state.GetOrCreateWindowCursor(partition);
 		const auto &fmask = partition.filter_mask;
 
 		D_ASSERT(aggr_input_data.bind_data);
diff --git a/extension/core_functions/include/core_functions/aggregate/quantile_sort_tree.hpp b/extension/core_functions/include/core_functions/aggregate/quantile_sort_tree.hpp
index c59ebb642f67..b669ac9f957d 100644
--- a/extension/core_functions/include/core_functions/aggregate/quantile_sort_tree.hpp
+++ b/extension/core_functions/include/core_functions/aggregate/quantile_sort_tree.hpp
@@ -26,9 +26,13 @@ namespace duckdb {
 // Paged access
 template <typename INPUT_TYPE>
 struct QuantileCursor {
-	explicit QuantileCursor(const ColumnDataCollection &inputs, bool all_valid) : inputs(inputs), all_valid(all_valid) {
-		inputs.InitializeScan(scan);
+	explicit QuantileCursor(const WindowPartitionInput &partition) : inputs(*partition.inputs) {
+		D_ASSERT(partition.column_ids.size() == 1);
+		inputs.InitializeScan(scan, partition.column_ids);
 		inputs.InitializeScanChunk(scan, page);
+
+		D_ASSERT(partition.all_valid.size() == 1);
+		all_valid = partition.all_valid[0];
 	}
 
 	inline sel_t RowOffset(idx_t row_idx) const {
@@ -330,7 +334,11 @@ static unique_ptr<GlobalSortState> SortQuantileIndices(const WindowPartitionInpu
 	SelectionVector filtered(capacity);
 
 	// TODO: Two pass parallel sorting using Build
-	auto order_expr = make_uniq<BoundConstantExpression>(Value(inputs.Types()[0]));
+	ColumnDataScanState state;
+	DataChunk sort;
+	inputs.InitializeScan(state, partition.column_ids);
+	inputs.InitializeScanChunk(state, sort);
+	auto order_expr = make_uniq<BoundConstantExpression>(Value(sort.GetTypes()[0]));
 	vector<BoundOrderByNode> orders;
 	orders.emplace_back(BoundOrderByNode(order_type, OrderByNullType::NULLS_LAST, std::move(order_expr)));
 
@@ -343,10 +351,6 @@ static unique_ptr<GlobalSortState> SortQuantileIndices(const WindowPartitionInpu
 	local_sort.Initialize(*global_sort, global_sort->buffer_manager);
 
 	//	Build the indirection array by scanning the valid indices
-	ColumnDataScanState state;
-	DataChunk sort;
-	inputs.InitializeScan(state);
-	inputs.InitializeScanChunk(state, sort);
 	while (inputs.Scan(state, sort)) {
 		// Match the payload to the scanned data
 		if (sort.size() > capacity) {
@@ -361,7 +365,7 @@ static unique_ptr<GlobalSortState> SortQuantileIndices(const WindowPartitionInpu
 		payload.SetCardinality(sort);
 		indices.Sequence(int64_t(state.current_row_index), 1, payload.size());
 
-		if (!filter_mask.AllValid() || !partition.all_valid) {
+		if (!filter_mask.AllValid() || !partition.all_valid[0]) {
 			auto &key = sort.data[0];
 			auto &validity = FlatVector::Validity(key);
 			idx_t valid = 0;
diff --git a/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp b/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp
index 3b4c9304792e..da78b8e143c7 100644
--- a/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp
+++ b/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp
@@ -300,9 +300,9 @@ struct QuantileState {
 		return *window_state;
 	}
 
-	CursorType &GetOrCreateWindowCursor(const ColumnDataCollection &inputs, bool all_valid) {
+	CursorType &GetOrCreateWindowCursor(const WindowPartitionInput &partition) {
 		if (!window_state) {
-			window_cursor = make_uniq<CursorType>(inputs, all_valid);
+			window_cursor = make_uniq<CursorType>(partition);
 		}
 		return *window_cursor;
 	}
diff --git a/src/execution/operator/aggregate/physical_window.cpp b/src/execution/operator/aggregate/physical_window.cpp
index 0af87505e584..dfdc7cae626f 100644
--- a/src/execution/operator/aggregate/physical_window.cpp
+++ b/src/execution/operator/aggregate/physical_window.cpp
@@ -102,6 +102,8 @@ class WindowHashGroup {
 	ValidityMask partition_mask;
 	//! The order boundary mask
 	OrderMasks order_masks;
+	//! The fully materialised data collection
+	unique_ptr<WindowCollection> collection;
 	//! External paging
 	bool external;
 	// The processing stage for this group
@@ -145,6 +147,8 @@ class WindowGlobalSinkState : public GlobalSinkState {
 	unique_ptr<WindowPartitionGlobalSinkState> global_partition;
 	//! The execution functions
 	Executors executors;
+	//! The shared expressions library
+	WindowSharedExpressions shared;
 };
 
 class WindowPartitionGlobalSinkState : public PartitionGlobalSinkState {
@@ -215,31 +219,31 @@ PhysicalWindow::PhysicalWindow(vector<LogicalType> types, vector<unique_ptr<Expr
 }
 
 static unique_ptr<WindowExecutor> WindowExecutorFactory(BoundWindowExpression &wexpr, ClientContext &context,
-                                                        WindowAggregationMode mode) {
+                                                        WindowSharedExpressions &shared, WindowAggregationMode mode) {
 	switch (wexpr.type) {
 	case ExpressionType::WINDOW_AGGREGATE:
-		return make_uniq<WindowAggregateExecutor>(wexpr, context, mode);
+		return make_uniq<WindowAggregateExecutor>(wexpr, context, shared, mode);
 	case ExpressionType::WINDOW_ROW_NUMBER:
-		return make_uniq<WindowRowNumberExecutor>(wexpr, context);
+		return make_uniq<WindowRowNumberExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_RANK_DENSE:
-		return make_uniq<WindowDenseRankExecutor>(wexpr, context);
+		return make_uniq<WindowDenseRankExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_RANK:
-		return make_uniq<WindowRankExecutor>(wexpr, context);
+		return make_uniq<WindowRankExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_PERCENT_RANK:
-		return make_uniq<WindowPercentRankExecutor>(wexpr, context);
+		return make_uniq<WindowPercentRankExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_CUME_DIST:
-		return make_uniq<WindowCumeDistExecutor>(wexpr, context);
+		return make_uniq<WindowCumeDistExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_NTILE:
-		return make_uniq<WindowNtileExecutor>(wexpr, context);
+		return make_uniq<WindowNtileExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_LEAD:
 	case ExpressionType::WINDOW_LAG:
-		return make_uniq<WindowLeadLagExecutor>(wexpr, context);
+		return make_uniq<WindowLeadLagExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_FIRST_VALUE:
-		return make_uniq<WindowFirstValueExecutor>(wexpr, context);
+		return make_uniq<WindowFirstValueExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_LAST_VALUE:
-		return make_uniq<WindowLastValueExecutor>(wexpr, context);
+		return make_uniq<WindowLastValueExecutor>(wexpr, context, shared);
 	case ExpressionType::WINDOW_NTH_VALUE:
-		return make_uniq<WindowNthValueExecutor>(wexpr, context);
+		return make_uniq<WindowNthValueExecutor>(wexpr, context, shared);
 		break;
 	default:
 		throw InternalException("Window aggregate type %s", ExpressionTypeToString(wexpr.type));
@@ -256,7 +260,7 @@ WindowGlobalSinkState::WindowGlobalSinkState(const PhysicalWindow &op, ClientCon
 	for (idx_t expr_idx = 0; expr_idx < op.select_list.size(); ++expr_idx) {
 		D_ASSERT(op.select_list[expr_idx]->GetExpressionClass() == ExpressionClass::BOUND_WINDOW);
 		auto &wexpr = op.select_list[expr_idx]->Cast<BoundWindowExpression>();
-		auto wexec = WindowExecutorFactory(wexpr, context, mode);
+		auto wexec = WindowExecutorFactory(wexpr, context, shared, mode);
 		executors.emplace_back(std::move(wexec));
 	}
 
@@ -568,6 +572,15 @@ WindowHashGroup::WindowHashGroup(WindowGlobalSinkState &gstate, const idx_t hash
 	if (rows) {
 		blocks = rows->blocks.size();
 	}
+
+	// Set up the collection for any fully materialised data
+	const auto &shared = WindowSharedExpressions::GetSortedExpressions(gstate.shared.coll_shared);
+	vector<LogicalType> types;
+	for (auto &expr : shared) {
+		types.emplace_back(expr->return_type);
+	}
+	auto &buffer_manager = BufferManager::GetBufferManager(gstate.context);
+	collection = make_uniq<WindowCollection>(buffer_manager, count, types);
 }
 
 // Per-thread scan state
@@ -606,6 +619,19 @@ class WindowLocalSourceState : public LocalSourceState {
 	void Sink();
 	void Finalize();
 	void GetData(DataChunk &chunk);
+
+	//! Storage and evaluation for the fully materialised data
+	unique_ptr<WindowBuilder> builder;
+	ExpressionExecutor coll_exec;
+	DataChunk coll_chunk;
+
+	//! Storage and evaluation for chunks used in the sink/build phase
+	ExpressionExecutor sink_exec;
+	DataChunk sink_chunk;
+
+	//! Storage and evaluation for chunks used in the evaluate phase
+	ExpressionExecutor eval_exec;
+	DataChunk eval_chunk;
 };
 
 WindowHashGroup::ExecutorGlobalStates &WindowHashGroup::Initialize(WindowGlobalSinkState &gsink) {
@@ -660,8 +686,30 @@ void WindowLocalSourceState::Sink() {
 				break;
 			}
 
+			//	Compute fully materialised expressions
+			if (coll_chunk.data.empty()) {
+				coll_chunk.SetCardinality(input_chunk);
+			} else {
+				coll_chunk.Reset();
+				coll_exec.Execute(input_chunk, coll_chunk);
+				auto collection = window_hash_group->collection.get();
+				if (!builder || &builder->collection != collection) {
+					builder = make_uniq<WindowBuilder>(*collection);
+				}
+
+				builder->Sink(coll_chunk, input_idx);
+			}
+
+			// Compute sink expressions
+			if (sink_chunk.data.empty()) {
+				sink_chunk.SetCardinality(input_chunk);
+			} else {
+				sink_chunk.Reset();
+				sink_exec.Execute(input_chunk, sink_chunk);
+			}
+
 			for (idx_t w = 0; w < executors.size(); ++w) {
-				executors[w]->Sink(input_chunk, input_idx, window_hash_group->count, *gestates[w], *local_states[w]);
+				executors[w]->Sink(sink_chunk, coll_chunk, input_idx, *gestates[w], *local_states[w]);
 			}
 
 			window_hash_group->sunk += input_chunk.size();
@@ -677,15 +725,20 @@ void WindowLocalSourceState::Finalize() {
 	D_ASSERT(task);
 	D_ASSERT(task->stage == WindowGroupStage::FINALIZE);
 
+	// First finalize the collection (so the executors can use it)
+	auto &gsink = gsource.gsink;
+	if (window_hash_group->collection) {
+		window_hash_group->collection->Combine(gsink.shared.coll_validity);
+	}
+
 	// Finalize all the executors.
 	// Parallel finalisation is handled internally by the executor,
 	// and should not return until all threads have completed work.
-	auto &gsink = gsource.gsink;
 	const auto &executors = gsink.executors;
 	auto &gestates = window_hash_group->gestates;
 	auto &local_states = window_hash_group->thread_states.at(task->thread_idx);
 	for (idx_t w = 0; w < executors.size(); ++w) {
-		executors[w]->Finalize(*gestates[w], *local_states[w]);
+		executors[w]->Finalize(*gestates[w], *local_states[w], window_hash_group->collection);
 	}
 
 	//	Mark this range as done
@@ -693,7 +746,9 @@ void WindowLocalSourceState::Finalize() {
 	task->begin_idx = task->end_idx;
 }
 
-WindowLocalSourceState::WindowLocalSourceState(WindowGlobalSourceState &gsource) : gsource(gsource), batch_index(0) {
+WindowLocalSourceState::WindowLocalSourceState(WindowGlobalSourceState &gsource)
+    : gsource(gsource), batch_index(0), coll_exec(gsource.context), sink_exec(gsource.context),
+      eval_exec(gsource.context) {
 	auto &gsink = gsource.gsink;
 	auto &global_partition = *gsink.global_partition;
 
@@ -706,6 +761,11 @@ WindowLocalSourceState::WindowLocalSourceState(WindowGlobalSourceState &gsource)
 	}
 	output_chunk.Initialize(global_partition.allocator, output_types);
 
+	auto &shared = gsink.shared;
+	shared.PrepareCollection(coll_exec, coll_chunk);
+	shared.PrepareSink(sink_exec, sink_chunk);
+	shared.PrepareEvaluate(eval_exec, eval_chunk);
+
 	++gsource.locals;
 }
 
@@ -819,7 +879,13 @@ void WindowLocalSourceState::GetData(DataChunk &result) {
 		auto &gstate = *gestates[expr_idx];
 		auto &lstate = *local_states[expr_idx];
 		auto &result = output_chunk.data[expr_idx];
-		executor.Evaluate(position, input_chunk, result, lstate, gstate);
+		if (eval_chunk.data.empty()) {
+			eval_chunk.SetCardinality(input_chunk);
+		} else {
+			eval_chunk.Reset();
+			eval_exec.Execute(input_chunk, eval_chunk);
+		}
+		executor.Evaluate(position, eval_chunk, result, lstate, gstate);
 	}
 	output_chunk.SetCardinality(input_chunk);
 	output_chunk.Verify();
diff --git a/src/execution/window_executor.cpp b/src/execution/window_executor.cpp
index 302dfce83732..6d36d98e50c0 100644
--- a/src/execution/window_executor.cpp
+++ b/src/execution/window_executor.cpp
@@ -2,6 +2,7 @@
 
 #include "duckdb/common/operator/add.hpp"
 #include "duckdb/common/operator/subtract.hpp"
+#include "duckdb/planner/expression/bound_reference_expression.hpp"
 
 #include "duckdb/common/array.hpp"
 
@@ -11,10 +12,17 @@ namespace duckdb {
 // WindowCollection
 //===--------------------------------------------------------------------===//
 WindowCollection::WindowCollection(BufferManager &buffer_manager, idx_t count, const vector<LogicalType> &types)
-    : all_valid(true), types(types), count(count), buffer_manager(buffer_manager) {
+    : all_valids(types.size()), types(types), count(count), buffer_manager(buffer_manager) {
 	if (!types.empty()) {
 		inputs = make_uniq<ColumnDataCollection>(buffer_manager, types);
 	}
+
+	validities.resize(types.size());
+
+	// Atomic vectors can't be constructed with a given value
+	for (auto &all_valid : all_valids) {
+		all_valid = true;
+	}
 }
 
 void WindowCollection::GetCollection(idx_t row_idx, ColumnDataCollectionSpec &spec) {
@@ -32,31 +40,61 @@ void WindowCollection::GetCollection(idx_t row_idx, ColumnDataCollectionSpec &sp
 	collections.emplace_back(std::move(collection));
 }
 
-void WindowCollection::Combine(bool build_validity) {
+void WindowCollection::Combine(const ColumnSet &validity_cols) {
 	lock_guard<mutex> collection_guard(lock);
 
-	if (collections.empty()) {
+	// If there are no columns (COUNT(*)) then this is a NOP
+	if (types.empty()) {
+		return;
+	}
+
+	// Have we already combined?
+	if (inputs->Count()) {
+		D_ASSERT(collections.empty());
+		D_ASSERT(ranges.empty());
 		return;
 	}
 
+	// If there are columns, we should have data
+	D_ASSERT(!collections.empty());
+	D_ASSERT(!ranges.empty());
+
 	for (auto &range : ranges) {
 		inputs->Combine(*collections[range.second]);
 	}
 	collections.clear();
 	ranges.clear();
 
-	if (build_validity && !all_valid) {
-		D_ASSERT(inputs.get());
-		validity.Initialize(inputs->Count());
-		WindowCursor cursor(*this);
-		idx_t target_offset = 0;
-		while (cursor.Scan()) {
-			const auto count = cursor.chunk.size();
-			auto &other = FlatVector::Validity(cursor.chunk.data[0]);
-			validity.SliceInPlace(other, target_offset, 0, count);
-			target_offset += count;
+	if (validity_cols.empty()) {
+		return;
+	}
+
+	D_ASSERT(inputs.get());
+
+	//	Find all columns with NULLs
+	vector<column_t> invalid_cols;
+	for (auto &col_idx : validity_cols) {
+		if (!all_valids[col_idx]) {
+			invalid_cols.emplace_back(col_idx);
+			validities[col_idx].Initialize(inputs->Count());
 		}
 	}
+
+	if (invalid_cols.empty()) {
+		return;
+	}
+
+	WindowCursor cursor(*this, invalid_cols);
+	idx_t target_offset = 0;
+	while (cursor.Scan()) {
+		const auto count = cursor.chunk.size();
+		for (idx_t i = 0; i < invalid_cols.size(); ++i) {
+			auto &other = FlatVector::Validity(cursor.chunk.data[i]);
+			const auto col_idx = invalid_cols[i];
+			validities[col_idx].SliceInPlace(other, target_offset, 0, count);
+		}
+		target_offset += count;
+	}
 }
 
 WindowBuilder::WindowBuilder(WindowCollection &collection) : collection(collection) {
@@ -72,34 +110,90 @@ void WindowBuilder::Sink(DataChunk &chunk, idx_t input_idx) {
 	sink.second->Append(appender, chunk);
 
 	// Record NULLs
-	if (all_valid) {
+	for (column_t col_idx = 0; col_idx < chunk.ColumnCount(); ++col_idx) {
+		if (!collection.all_valids[col_idx]) {
+			continue;
+		}
+
+		// Column was valid, make sure it still is.
 		UnifiedVectorFormat data;
-		chunk.data[0].ToUnifiedFormat(chunk.size(), data);
-		all_valid = data.validity.AllValid();
-		if (!all_valid) {
-			collection.all_valid = false;
+		chunk.data[col_idx].ToUnifiedFormat(chunk.size(), data);
+		if (!data.validity.AllValid()) {
+			collection.all_valids[col_idx] = false;
 		}
 	}
 }
 
-WindowCursor::WindowCursor(const WindowCollection &paged) : paged(paged) {
-	if (paged.GetTypes().empty()) {
+WindowCursor::WindowCursor(const WindowCollection &paged, vector<column_t> column_ids) : paged(paged) {
+	D_ASSERT(paged.collections.empty());
+	D_ASSERT(paged.ranges.empty());
+	if (column_ids.empty()) {
 		//	For things like COUNT(*) set the state up to contain the whole range
 		state.segment_index = 0;
 		state.chunk_index = 0;
 		state.current_row_index = 0;
 		state.next_row_index = paged.size();
 		state.properties = ColumnDataScanProperties::ALLOW_ZERO_COPY;
+		chunk.SetCapacity(state.next_row_index);
 		chunk.SetCardinality(state.next_row_index);
 		return;
 	} else if (chunk.data.empty()) {
 		auto &inputs = paged.inputs;
 		D_ASSERT(inputs.get());
-		inputs->InitializeScan(state);
+		inputs->InitializeScan(state, std::move(column_ids));
 		inputs->InitializeScanChunk(state, chunk);
 	}
 }
 
+WindowCursor::WindowCursor(const WindowCollection &paged, column_t col_idx)
+    : WindowCursor(paged, vector<column_t>(1, col_idx)) {
+}
+
+struct WindowInputExpression {
+	WindowInputExpression(DataChunk &chunk, column_t col_idx)
+	    : ptype(PhysicalType::INVALID), scalar(true), chunk(chunk), col_idx(col_idx) {
+		if (col_idx < chunk.data.size()) {
+			auto &col = chunk.data[col_idx];
+			ptype = col.GetType().InternalType();
+			scalar = (col.GetVectorType() == VectorType::CONSTANT_VECTOR);
+		}
+	}
+
+	inline PhysicalType InternalType() const {
+		return ptype;
+	}
+
+	template <typename T>
+	inline T GetCell(idx_t i) const {
+		D_ASSERT(!chunk.data.empty());
+		const auto data = FlatVector::GetData<T>(chunk.data[col_idx]);
+		return data[scalar ? 0 : i];
+	}
+
+	inline bool CellIsNull(idx_t i) const {
+		D_ASSERT(!chunk.data.empty());
+		auto &col = chunk.data[col_idx];
+
+		if (scalar) {
+			return ConstantVector::IsNull(col);
+		}
+		return FlatVector::IsNull(col, i);
+	}
+
+	inline void CopyCell(Vector &target, idx_t target_offset, idx_t width = 1) const {
+		D_ASSERT(!chunk.data.empty());
+		auto &source = chunk.data[col_idx];
+		auto source_offset = scalar ? 0 : target_offset;
+		VectorOperations::Copy(source, target, source_offset + width, source_offset, target_offset);
+	}
+
+private:
+	PhysicalType ptype;
+	bool scalar;
+	DataChunk &chunk;
+	const column_t col_idx;
+};
+
 static idx_t FindNextStart(const ValidityMask &mask, idx_t l, const idx_t r, idx_t &n) {
 	if (mask.AllValid()) {
 		auto start = MinValue(l + n - 1, r);
@@ -330,9 +424,7 @@ template <typename OP, bool FROM>
 static idx_t FindRangeBound(WindowCursor &over, const idx_t order_begin, const idx_t order_end,
                             const WindowBoundary range, WindowInputExpression &boundary, const idx_t chunk_idx,
                             const FrameBounds &prev) {
-	D_ASSERT(boundary.chunk.ColumnCount() == 1);
-
-	switch (boundary.chunk.data[0].GetType().InternalType()) {
+	switch (boundary.InternalType()) {
 	case PhysicalType::INT8:
 		return FindTypedRangeBound<int8_t, OP, FROM>(over, order_begin, order_end, range, boundary, chunk_idx, prev);
 	case PhysicalType::INT16:
@@ -695,34 +787,30 @@ class WindowExecutorBoundsState : public WindowExecutorLocalState {
 	~WindowExecutorBoundsState() override {
 	}
 
-	virtual void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range);
+	virtual void UpdateBounds(WindowExecutorGlobalState &gstate, idx_t row_idx, DataChunk &eval_chunk,
+	                          optional_ptr<WindowCursor> range);
 
 	// Frame management
 	const ValidityMask &partition_mask;
 	const ValidityMask &order_mask;
 	DataChunk bounds;
 	WindowBoundariesState state;
-
-	// evaluate boundaries if present. Parser has checked boundary types.
-	WindowInputExpression boundary_start;
-	WindowInputExpression boundary_end;
 };
 
 WindowExecutorBoundsState::WindowExecutorBoundsState(const WindowExecutorGlobalState &gstate)
     : WindowExecutorLocalState(gstate), partition_mask(gstate.partition_mask), order_mask(gstate.order_mask),
-      state(gstate.executor.wexpr, gstate.payload_count),
-      boundary_start(gstate.executor.wexpr.start_expr.get(), gstate.executor.context),
-      boundary_end(gstate.executor.wexpr.end_expr.get(), gstate.executor.context) {
+      state(gstate.executor.wexpr, gstate.payload_count) {
 	vector<LogicalType> bounds_types(6, LogicalType(LogicalTypeId::UBIGINT));
 	bounds.Initialize(Allocator::Get(gstate.executor.context), bounds_types);
 }
 
-void WindowExecutorBoundsState::UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range) {
+void WindowExecutorBoundsState::UpdateBounds(WindowExecutorGlobalState &gstate, idx_t row_idx, DataChunk &eval_chunk,
+                                             optional_ptr<WindowCursor> range) {
 	// Evaluate the row-level arguments
-	boundary_start.Execute(input_chunk);
-	boundary_end.Execute(input_chunk);
+	WindowInputExpression boundary_start(eval_chunk, gstate.executor.boundary_start_idx);
+	WindowInputExpression boundary_end(eval_chunk, gstate.executor.boundary_end_idx);
 
-	const auto count = input_chunk.size();
+	const auto count = eval_chunk.size();
 	bounds.Reset();
 	state.Bounds(bounds, row_idx, range, count, boundary_start, boundary_end, partition_mask, order_mask);
 }
@@ -824,76 +912,83 @@ void ExclusionFilter::ResetMask(idx_t row_idx, idx_t offset) {
 	}
 }
 
-//===--------------------------------------------------------------------===//
-// WindowExecutor
-//===--------------------------------------------------------------------===//
-static void PrepareInputExpressions(const vector<unique_ptr<Expression>> &exprs, ExpressionExecutor &executor,
-                                    DataChunk &chunk) {
-	if (exprs.empty()) {
-		return;
+column_t WindowSharedExpressions::RegisterExpr(const unique_ptr<Expression> &expr, Shared &shared) {
+	auto pexpr = expr.get();
+	if (!pexpr) {
+		return DConstants::INVALID_INDEX;
+	}
+
+	//	We need to make separate columns for volatile arguments
+	const auto is_volatile = expr->IsVolatile();
+	auto i = shared.columns.find(*pexpr);
+	if (i != shared.columns.end() && !is_volatile) {
+		return i->second.front();
+	}
+
+	// New column, find maximum column number
+	column_t result = shared.size++;
+	shared.columns[*pexpr].emplace_back(result);
+
+	return result;
+}
+
+vector<const Expression *> WindowSharedExpressions::GetSortedExpressions(Shared &shared) {
+	vector<const Expression *> sorted(shared.size, nullptr);
+	for (auto &col : shared.columns) {
+		auto &expr = col.first.get();
+		for (auto col_idx : col.second) {
+			sorted[col_idx] = &expr;
+		}
 	}
 
+	return sorted;
+}
+void WindowSharedExpressions::PrepareExecutors(Shared &shared, ExpressionExecutor &exec, DataChunk &chunk) {
+	const auto sorted = GetSortedExpressions(shared);
 	vector<LogicalType> types;
-	for (idx_t expr_idx = 0; expr_idx < exprs.size(); ++expr_idx) {
-		types.push_back(exprs[expr_idx]->return_type);
-		executor.AddExpression(*exprs[expr_idx]);
+	for (auto expr : sorted) {
+		exec.AddExpression(*expr);
+		types.emplace_back(expr->return_type);
 	}
 
 	if (!types.empty()) {
-		auto &allocator = executor.GetAllocator();
-		chunk.Initialize(allocator, types);
+		chunk.Initialize(exec.GetAllocator(), types);
 	}
 }
 
-WindowExecutor::WindowExecutor(BoundWindowExpression &wexpr, ClientContext &context) : wexpr(wexpr), context(context) {
+//===--------------------------------------------------------------------===//
+// WindowExecutor
+//===--------------------------------------------------------------------===//
+WindowExecutor::WindowExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared)
+    : wexpr(wexpr), context(context),
+      range_expr((HasPrecedingRange(wexpr) || HasFollowingRange(wexpr)) ? wexpr.orders[0].expression.get() : nullptr) {
+	if (range_expr) {
+		range_idx = shared.RegisterCollection(wexpr.orders[0].expression, false);
+	}
+
+	boundary_start_idx = shared.RegisterEvaluate(wexpr.start_expr);
+	boundary_end_idx = shared.RegisterEvaluate(wexpr.end_expr);
 }
 
 WindowExecutorGlobalState::WindowExecutorGlobalState(const WindowExecutor &executor, const idx_t payload_count,
                                                      const ValidityMask &partition_mask, const ValidityMask &order_mask)
-    : executor(executor), payload_count(payload_count), partition_mask(partition_mask), order_mask(order_mask),
-      range_expr((HasPrecedingRange(executor.wexpr) || HasFollowingRange(executor.wexpr))
-                     ? executor.wexpr.orders[0].expression.get()
-                     : nullptr) {
+    : executor(executor), payload_count(payload_count), partition_mask(partition_mask), order_mask(order_mask) {
 	for (const auto &child : executor.wexpr.children) {
 		arg_types.emplace_back(child->return_type);
 	}
-
-	if (range_expr) {
-		vector<LogicalType> types;
-		types.emplace_back(range_expr->return_type);
-		range = make_uniq<WindowCollection>(BufferManager::GetBufferManager(executor.context), payload_count, types);
-	}
 }
 
-WindowExecutorLocalState::WindowExecutorLocalState(const WindowExecutorGlobalState &gstate)
-    : payload_executor(gstate.executor.context), range_executor(gstate.executor.context) {
-	// TODO: child may be a scalar, don't need to materialize the whole collection then
-
-	// evaluate inner expressions of window functions, could be more complex
-	PrepareInputExpressions(gstate.executor.wexpr.children, payload_executor, payload_chunk);
-
-	if (gstate.range_expr) {
-		range_executor.AddExpression(*gstate.range_expr);
-
-		D_ASSERT(gstate.range.get());
-		auto &allocator = range_executor.GetAllocator();
-		range_chunk.Initialize(allocator, gstate.range->GetTypes());
-		range_builder = make_uniq<WindowBuilder>(*gstate.range);
-	}
+WindowExecutorLocalState::WindowExecutorLocalState(const WindowExecutorGlobalState &gstate) {
 }
 
-void WindowExecutorLocalState::Sink(WindowExecutorGlobalState &gstate, DataChunk &input_chunk, idx_t input_idx) {
-	if (gstate.range_expr) {
-		range_executor.Execute(input_chunk, range_chunk);
-		range_builder->Sink(range_chunk, input_idx);
-	}
+void WindowExecutorLocalState::Sink(WindowExecutorGlobalState &gstate, DataChunk &sink_chunk, DataChunk &coll_chunk,
+                                    idx_t input_idx) {
 }
 
-void WindowExecutorLocalState::Finalize(WindowExecutorGlobalState &gstate) {
-	if (gstate.range) {
-		if (!range_cursor) {
-			range_cursor = make_uniq<WindowCursor>(*gstate.range);
-		}
+void WindowExecutorLocalState::Finalize(WindowExecutorGlobalState &gstate, CollectionPtr collection) {
+	const auto range_idx = gstate.executor.range_idx;
+	if (range_idx != DConstants::INVALID_INDEX) {
+		range_cursor = make_uniq<WindowCursor>(*collection, range_idx);
 	}
 }
 
@@ -907,17 +1002,14 @@ unique_ptr<WindowExecutorLocalState> WindowExecutor::GetLocalState(const WindowE
 	return make_uniq<WindowExecutorBoundsState>(gstate);
 }
 
-void WindowExecutor::Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
+void WindowExecutor::Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, const idx_t input_idx,
                           WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	lstate.Sink(gstate, input_chunk, input_idx);
+	lstate.Sink(gstate, sink_chunk, coll_chunk, input_idx);
 }
 
-void WindowExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	if (gstate.range) {
-		gstate.range->Combine(false);
-	}
-
-	lstate.Finalize(gstate);
+void WindowExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+                              CollectionPtr collection) const {
+	lstate.Finalize(gstate, collection);
 }
 
 //===--------------------------------------------------------------------===//
@@ -925,22 +1017,17 @@ void WindowExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorL
 //===--------------------------------------------------------------------===//
 class WindowAggregateExecutorGlobalState : public WindowExecutorGlobalState {
 public:
-	bool IsConstantAggregate();
-	bool IsCustomAggregate();
-	bool IsDistinctAggregate();
-
 	WindowAggregateExecutorGlobalState(const WindowAggregateExecutor &executor, const idx_t payload_count,
 	                                   const ValidityMask &partition_mask, const ValidityMask &order_mask);
 
-	// aggregate computation algorithm
-	unique_ptr<WindowAggregator> aggregator;
 	// aggregate global state
 	unique_ptr<WindowAggregatorState> gsink;
-};
 
-bool WindowAggregateExecutorGlobalState::IsConstantAggregate() {
-	const auto &wexpr = executor.wexpr;
+	// the filter reference expression.
+	const Expression *filter_ref;
+};
 
+bool WindowAggregateExecutor::IsConstantAggregate() {
 	if (!wexpr.aggregate) {
 		return false;
 	}
@@ -998,9 +1085,7 @@ bool WindowAggregateExecutorGlobalState::IsConstantAggregate() {
 	return true;
 }
 
-bool WindowAggregateExecutorGlobalState::IsDistinctAggregate() {
-	const auto &wexpr = executor.wexpr;
-
+bool WindowAggregateExecutor::IsDistinctAggregate() {
 	if (!wexpr.aggregate) {
 		return false;
 	}
@@ -1008,10 +1093,7 @@ bool WindowAggregateExecutorGlobalState::IsDistinctAggregate() {
 	return wexpr.distinct;
 }
 
-bool WindowAggregateExecutorGlobalState::IsCustomAggregate() {
-	const auto &wexpr = executor.wexpr;
-	const auto &mode = reinterpret_cast<const WindowAggregateExecutor &>(executor).mode;
-
+bool WindowAggregateExecutor::IsCustomAggregate() {
 	if (!wexpr.aggregate) {
 		return false;
 	}
@@ -1023,53 +1105,56 @@ bool WindowAggregateExecutorGlobalState::IsCustomAggregate() {
 	return (mode < WindowAggregationMode::COMBINE);
 }
 
-void WindowExecutor::Evaluate(idx_t row_idx, DataChunk &input_chunk, Vector &result, WindowExecutorLocalState &lstate,
+void WindowExecutor::Evaluate(idx_t row_idx, DataChunk &eval_chunk, Vector &result, WindowExecutorLocalState &lstate,
                               WindowExecutorGlobalState &gstate) const {
 	auto &lbstate = lstate.Cast<WindowExecutorBoundsState>();
-	lbstate.UpdateBounds(row_idx, input_chunk, lstate.range_cursor);
+	lbstate.UpdateBounds(gstate, row_idx, eval_chunk, lstate.range_cursor);
 
-	const auto count = input_chunk.size();
-	EvaluateInternal(gstate, lstate, result, count, row_idx);
+	const auto count = eval_chunk.size();
+	EvaluateInternal(gstate, lstate, eval_chunk, result, count, row_idx);
 
 	result.Verify(count);
 }
 
 WindowAggregateExecutor::WindowAggregateExecutor(BoundWindowExpression &wexpr, ClientContext &context,
-                                                 WindowAggregationMode mode)
-    : WindowExecutor(wexpr, context), mode(mode) {
-}
-
-WindowAggregateExecutorGlobalState::WindowAggregateExecutorGlobalState(const WindowAggregateExecutor &executor,
-                                                                       const idx_t group_count,
-                                                                       const ValidityMask &partition_mask,
-                                                                       const ValidityMask &order_mask)
-    : WindowExecutorGlobalState(executor, group_count, partition_mask, order_mask) {
-	auto &wexpr = executor.wexpr;
-	auto &context = executor.context;
+                                                 WindowSharedExpressions &shared, WindowAggregationMode mode)
+    : WindowExecutor(wexpr, context, shared), mode(mode) {
 	auto return_type = wexpr.return_type;
-	const auto &mode = reinterpret_cast<const WindowAggregateExecutor &>(executor).mode;
 
 	// Force naive for SEPARATE mode or for (currently!) unsupported functionality
 	const auto force_naive =
 	    !ClientConfig::GetConfig(context).enable_optimizer || mode == WindowAggregationMode::SEPARATE;
-	AggregateObject aggr(wexpr);
 	if (force_naive || (wexpr.distinct && wexpr.exclude_clause != WindowExcludeMode::NO_OTHER)) {
-		aggregator = make_uniq<WindowNaiveAggregator>(aggr, arg_types, return_type, wexpr.exclude_clause);
+		aggregator = make_uniq<WindowNaiveAggregator>(wexpr, wexpr.exclude_clause, shared);
 	} else if (IsDistinctAggregate()) {
 		// build a merge sort tree
 		// see https://dl.acm.org/doi/pdf/10.1145/3514221.3526184
-		aggregator = make_uniq<WindowDistinctAggregator>(aggr, arg_types, return_type, wexpr.exclude_clause, context);
+		aggregator = make_uniq<WindowDistinctAggregator>(wexpr, wexpr.exclude_clause, shared, context);
 	} else if (IsConstantAggregate()) {
-		aggregator = make_uniq<WindowConstantAggregator>(aggr, arg_types, return_type, wexpr.exclude_clause);
+		aggregator = make_uniq<WindowConstantAggregator>(wexpr, wexpr.exclude_clause, shared);
 	} else if (IsCustomAggregate()) {
-		aggregator = make_uniq<WindowCustomAggregator>(aggr, arg_types, return_type, wexpr.exclude_clause);
+		aggregator = make_uniq<WindowCustomAggregator>(wexpr, wexpr.exclude_clause, shared);
 	} else {
 		// build a segment tree for frame-adhering aggregates
 		// see http://www.vldb.org/pvldb/vol8/p1058-leis.pdf
-		aggregator = make_uniq<WindowSegmentTree>(aggr, arg_types, return_type, mode, wexpr.exclude_clause);
+		aggregator = make_uniq<WindowSegmentTree>(wexpr, mode, wexpr.exclude_clause, shared);
 	}
 
-	gsink = aggregator->GetGlobalState(context, group_count, partition_mask);
+	// Compute the FILTER with the other eval columns.
+	// Anyone who needs it can then convert it to the form they need.
+	if (wexpr.filter_expr) {
+		const auto filter_idx = shared.RegisterSink(wexpr.filter_expr);
+		filter_ref = make_uniq<BoundReferenceExpression>(wexpr.filter_expr->return_type, filter_idx);
+	}
+}
+
+WindowAggregateExecutorGlobalState::WindowAggregateExecutorGlobalState(const WindowAggregateExecutor &executor,
+                                                                       const idx_t group_count,
+                                                                       const ValidityMask &partition_mask,
+                                                                       const ValidityMask &order_mask)
+    : WindowExecutorGlobalState(executor, group_count, partition_mask, order_mask),
+      filter_ref(executor.filter_ref.get()) {
+	gsink = executor.aggregator->GetGlobalState(executor.context, group_count, partition_mask);
 }
 
 unique_ptr<WindowExecutorGlobalState> WindowAggregateExecutor::GetGlobalState(const idx_t payload_count,
@@ -1087,9 +1172,9 @@ class WindowAggregateExecutorLocalState : public WindowExecutorBoundsState {
 		aggregator_state = aggregator.GetLocalState(*gastate.gsink);
 
 		// evaluate the FILTER clause and stuff it into a large mask for compactness and reuse
-		auto &wexpr = gstate.executor.wexpr;
-		if (wexpr.filter_expr) {
-			filter_executor.AddExpression(*wexpr.filter_expr);
+		auto filter_ref = gastate.filter_ref;
+		if (filter_ref) {
+			filter_executor.AddExpression(*filter_ref);
 			filter_sel.Initialize(STANDARD_VECTOR_SIZE);
 		}
 	}
@@ -1105,43 +1190,29 @@ class WindowAggregateExecutorLocalState : public WindowExecutorBoundsState {
 
 unique_ptr<WindowExecutorLocalState>
 WindowAggregateExecutor::GetLocalState(const WindowExecutorGlobalState &gstate) const {
-	auto &gastate = gstate.Cast<WindowAggregateExecutorGlobalState>();
-	auto res = make_uniq<WindowAggregateExecutorLocalState>(gstate, *gastate.aggregator);
-	return std::move(res);
+	return make_uniq<WindowAggregateExecutorLocalState>(gstate, *aggregator);
 }
 
-void WindowAggregateExecutor::Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
+void WindowAggregateExecutor::Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, const idx_t input_idx,
                                    WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
 	auto &gastate = gstate.Cast<WindowAggregateExecutorGlobalState>();
 	auto &lastate = lstate.Cast<WindowAggregateExecutorLocalState>();
 	auto &filter_sel = lastate.filter_sel;
 	auto &filter_executor = lastate.filter_executor;
-	auto &payload_executor = lastate.payload_executor;
-	auto &payload_chunk = lastate.payload_chunk;
-	auto &aggregator = gastate.aggregator;
 
 	idx_t filtered = 0;
 	SelectionVector *filtering = nullptr;
-	if (wexpr.filter_expr) {
+	if (gastate.filter_ref) {
 		filtering = &filter_sel;
-		filtered = filter_executor.SelectExpression(input_chunk, filter_sel);
-	}
-
-	if (!wexpr.children.empty()) {
-		payload_chunk.Reset();
-		payload_executor.Execute(input_chunk, payload_chunk);
-		payload_chunk.Verify();
-	} else if (aggregator) {
-		//	Zero-argument aggregate (e.g., COUNT(*)
-		payload_chunk.SetCardinality(input_chunk);
+		filtered = filter_executor.SelectExpression(sink_chunk, filter_sel);
 	}
 
 	D_ASSERT(aggregator);
 	auto &gestate = *gastate.gsink;
 	auto &lestate = *lastate.aggregator_state;
-	aggregator->Sink(gestate, lestate, payload_chunk, input_idx, filtering, filtered);
+	aggregator->Sink(gestate, lestate, sink_chunk, coll_chunk, input_idx, filtering, filtered);
 
-	WindowExecutor::Sink(input_chunk, input_idx, total_count, gstate, lstate);
+	WindowExecutor::Sink(sink_chunk, coll_chunk, input_idx, gstate, lstate);
 }
 
 static void ApplyWindowStats(const WindowBoundary &boundary, FrameDelta &delta, BaseStatistics *base, bool is_start) {
@@ -1202,11 +1273,11 @@ static void ApplyWindowStats(const WindowBoundary &boundary, FrameDelta &delta,
 	}
 }
 
-void WindowAggregateExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	WindowExecutor::Finalize(gstate, lstate);
+void WindowAggregateExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+                                       CollectionPtr collection) const {
+	WindowExecutor::Finalize(gstate, lstate, collection);
 
 	auto &gastate = gstate.Cast<WindowAggregateExecutorGlobalState>();
-	auto &aggregator = gastate.aggregator;
 	auto &gsink = gastate.gsink;
 	D_ASSERT(aggregator);
 
@@ -1226,14 +1297,14 @@ void WindowAggregateExecutor::Finalize(WindowExecutorGlobalState &gstate, Window
 	ApplyWindowStats(wexpr.end, stats[1], base, false);
 
 	auto &lastate = lstate.Cast<WindowAggregateExecutorLocalState>();
-	aggregator->Finalize(*gsink, *lastate.aggregator_state, stats);
+	aggregator->Finalize(*gsink, *lastate.aggregator_state, collection, stats);
 }
 
 void WindowAggregateExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                               Vector &result, idx_t count, idx_t row_idx) const {
+                                               DataChunk &eval_chunk, Vector &result, idx_t count,
+                                               idx_t row_idx) const {
 	auto &gastate = gstate.Cast<WindowAggregateExecutorGlobalState>();
 	auto &lastate = lstate.Cast<WindowAggregateExecutorLocalState>();
-	auto &aggregator = gastate.aggregator;
 	auto &gsink = gastate.gsink;
 	D_ASSERT(aggregator);
 
@@ -1245,12 +1316,14 @@ void WindowAggregateExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate
 //===--------------------------------------------------------------------===//
 // WindowRowNumberExecutor
 //===--------------------------------------------------------------------===//
-WindowRowNumberExecutor::WindowRowNumberExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowRowNumberExecutor::WindowRowNumberExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                                 WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
 }
 
 void WindowRowNumberExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                               Vector &result, idx_t count, idx_t row_idx) const {
+                                               DataChunk &eval_chunk, Vector &result, idx_t count,
+                                               idx_t row_idx) const {
 	auto &lbstate = lstate.Cast<WindowExecutorBoundsState>();
 	auto partition_begin = FlatVector::GetData<const idx_t>(lbstate.bounds.data[PARTITION_BEGIN]);
 	auto rdata = FlatVector::GetData<int64_t>(result);
@@ -1288,8 +1361,9 @@ void WindowPeerState::NextRank(idx_t partition_begin, idx_t peer_begin, idx_t ro
 	rank_equal++;
 }
 
-WindowRankExecutor::WindowRankExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowRankExecutor::WindowRankExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                       WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
 }
 
 unique_ptr<WindowExecutorLocalState> WindowRankExecutor::GetLocalState(const WindowExecutorGlobalState &gstate) const {
@@ -1297,7 +1371,7 @@ unique_ptr<WindowExecutorLocalState> WindowRankExecutor::GetLocalState(const Win
 }
 
 void WindowRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                          Vector &result, idx_t count, idx_t row_idx) const {
+                                          DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const {
 	auto &lpeer = lstate.Cast<WindowPeerState>();
 	auto partition_begin = FlatVector::GetData<const idx_t>(lpeer.bounds.data[PARTITION_BEGIN]);
 	auto peer_begin = FlatVector::GetData<const idx_t>(lpeer.bounds.data[PEER_BEGIN]);
@@ -1313,8 +1387,9 @@ void WindowRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, Win
 	}
 }
 
-WindowDenseRankExecutor::WindowDenseRankExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowDenseRankExecutor::WindowDenseRankExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                                 WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
 }
 
 unique_ptr<WindowExecutorLocalState>
@@ -1323,7 +1398,8 @@ WindowDenseRankExecutor::GetLocalState(const WindowExecutorGlobalState &gstate)
 }
 
 void WindowDenseRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                               Vector &result, idx_t count, idx_t row_idx) const {
+                                               DataChunk &eval_chunk, Vector &result, idx_t count,
+                                               idx_t row_idx) const {
 	auto &lpeer = lstate.Cast<WindowPeerState>();
 
 	auto &order_mask = gstate.order_mask;
@@ -1376,8 +1452,9 @@ void WindowDenseRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate
 	}
 }
 
-WindowPercentRankExecutor::WindowPercentRankExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowPercentRankExecutor::WindowPercentRankExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                                     WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
 }
 
 unique_ptr<WindowExecutorLocalState>
@@ -1386,7 +1463,8 @@ WindowPercentRankExecutor::GetLocalState(const WindowExecutorGlobalState &gstate
 }
 
 void WindowPercentRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                                 Vector &result, idx_t count, idx_t row_idx) const {
+                                                 DataChunk &eval_chunk, Vector &result, idx_t count,
+                                                 idx_t row_idx) const {
 	auto &lpeer = lstate.Cast<WindowPeerState>();
 	auto partition_begin = FlatVector::GetData<const idx_t>(lpeer.bounds.data[PARTITION_BEGIN]);
 	auto partition_end = FlatVector::GetData<const idx_t>(lpeer.bounds.data[PARTITION_END]);
@@ -1408,12 +1486,13 @@ void WindowPercentRankExecutor::EvaluateInternal(WindowExecutorGlobalState &gsta
 //===--------------------------------------------------------------------===//
 // WindowCumeDistExecutor
 //===--------------------------------------------------------------------===//
-WindowCumeDistExecutor::WindowCumeDistExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowCumeDistExecutor::WindowCumeDistExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                               WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
 }
 
 void WindowCumeDistExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                              Vector &result, idx_t count, idx_t row_idx) const {
+                                              DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const {
 	auto &lbstate = lstate.Cast<WindowExecutorBoundsState>();
 	auto partition_begin = FlatVector::GetData<const idx_t>(lbstate.bounds.data[PARTITION_BEGIN]);
 	auto partition_end = FlatVector::GetData<const idx_t>(lbstate.bounds.data[PARTITION_END]);
@@ -1432,23 +1511,18 @@ void WindowCumeDistExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 
 class WindowValueGlobalState : public WindowExecutorGlobalState {
 public:
-	using WindowDataChunkPtr = unique_ptr<WindowCollection>;
-	WindowValueGlobalState(const WindowExecutor &executor, const idx_t payload_count,
+	using WindowCollectionPtr = unique_ptr<WindowCollection>;
+	WindowValueGlobalState(const WindowValueExecutor &executor, const idx_t payload_count,
 	                       const ValidityMask &partition_mask, const ValidityMask &order_mask)
-	    : WindowExecutorGlobalState(executor, payload_count, partition_mask, order_mask)
-
-	{
-		if (!arg_types.empty()) {
-			auto &buffer_manager = BufferManager::GetBufferManager(executor.context);
-			payload_collection = make_uniq<WindowCollection>(buffer_manager, payload_count, arg_types);
-			ignore_nulls = &payload_collection->validity;
-		}
+	    : WindowExecutorGlobalState(executor, payload_count, partition_mask, order_mask), ignore_nulls(&all_valid),
+	      child_idx(executor.child_idx) {
 	}
 
-	// The partition values
-	WindowDataChunkPtr payload_collection;
 	// IGNORE NULLS
+	ValidityMask all_valid;
 	optional_ptr<ValidityMask> ignore_nulls;
+
+	const column_t child_idx;
 };
 
 //===--------------------------------------------------------------------===//
@@ -1459,13 +1533,11 @@ class WindowValueGlobalState : public WindowExecutorGlobalState {
 class WindowValueLocalState : public WindowExecutorBoundsState {
 public:
 	explicit WindowValueLocalState(const WindowValueGlobalState &gvstate)
-	    : WindowExecutorBoundsState(gvstate), gvstate(gvstate), builder(*gvstate.payload_collection) {
+	    : WindowExecutorBoundsState(gvstate), gvstate(gvstate) {
 	}
 
-	//! Sink a chunk into the collection
-	void Sink(DataChunk &chunk, idx_t row_idx);
 	//! Finish the sinking and prepare to scan
-	void Finalize(WindowExecutorGlobalState &gstate) override;
+	void Finalize(WindowExecutorGlobalState &gstate, CollectionPtr collection) override;
 
 	//! The corresponding global value state
 	const WindowValueGlobalState &gvstate;
@@ -1474,29 +1546,12 @@ class WindowValueLocalState : public WindowExecutorBoundsState {
 	//! The validity mask that combines both the NULLs and exclusion information
 	optional_ptr<ValidityMask> ignore_nulls_exclude;
 
-	//! The thread's current input collection
-	WindowBuilder builder;
 	//! The state used for reading the collection
 	unique_ptr<WindowCursor> cursor;
 };
 
-void WindowValueLocalState::Sink(DataChunk &input_chunk, const idx_t input_idx) {
-	payload_chunk.Reset();
-	payload_executor.Execute(input_chunk, payload_chunk);
-	payload_chunk.Verify();
-
-	// Check whether we need a a new collection
-	builder.Sink(payload_chunk, input_idx);
-}
-
-void WindowValueLocalState::Finalize(WindowExecutorGlobalState &gstate) {
-	WindowExecutorBoundsState::Finalize(gstate);
-
-	auto &payload_collection = gvstate.payload_collection;
-
-	// Thread-safe and idempotent
-	auto &wexpr = gvstate.executor.wexpr;
-	payload_collection->Combine(wexpr.ignore_nulls && wexpr.type != ExpressionType::WINDOW_NTILE);
+void WindowValueLocalState::Finalize(WindowExecutorGlobalState &gstate, CollectionPtr collection) {
+	WindowExecutorBoundsState::Finalize(gstate, collection);
 
 	// Set up the IGNORE NULLS state
 	auto ignore_nulls = gvstate.ignore_nulls;
@@ -1511,20 +1566,34 @@ void WindowValueLocalState::Finalize(WindowExecutorGlobalState &gstate) {
 	}
 
 	// Prepare to scan
-	if (!cursor) {
-		cursor = make_uniq<WindowCursor>(*payload_collection);
+	if (!cursor && gvstate.child_idx != DConstants::INVALID_INDEX) {
+		cursor = make_uniq<WindowCursor>(*collection, gvstate.child_idx);
 	}
 }
 
 //===--------------------------------------------------------------------===//
 // WindowValueExecutor
 //===--------------------------------------------------------------------===//
-WindowValueExecutor::WindowValueExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowExecutor(wexpr, context) {
+WindowValueExecutor::WindowValueExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                         WindowSharedExpressions &shared)
+    : WindowExecutor(wexpr, context, shared) {
+
+	//	The children have to be handled separately because only the first one is global
+	if (!wexpr.children.empty()) {
+		child_idx = shared.RegisterCollection(wexpr.children[0], wexpr.ignore_nulls);
+
+		if (wexpr.children.size() > 1) {
+			nth_idx = shared.RegisterEvaluate(wexpr.children[1]);
+		}
+	}
+
+	offset_idx = shared.RegisterEvaluate(wexpr.offset_expr);
+	default_idx = shared.RegisterEvaluate(wexpr.default_expr);
 }
 
-WindowNtileExecutor::WindowNtileExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowValueExecutor(wexpr, context) {
+WindowNtileExecutor::WindowNtileExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                         WindowSharedExpressions &shared)
+    : WindowValueExecutor(wexpr, context, shared) {
 }
 
 unique_ptr<WindowExecutorGlobalState> WindowValueExecutor::GetGlobalState(const idx_t payload_count,
@@ -1533,15 +1602,14 @@ unique_ptr<WindowExecutorGlobalState> WindowValueExecutor::GetGlobalState(const
 	return make_uniq<WindowValueGlobalState>(*this, payload_count, partition_mask, order_mask);
 }
 
-void WindowValueExecutor::Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
-                               WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	auto &lvstate = lstate.Cast<WindowValueLocalState>();
-
-	if (!wexpr.children.empty()) {
-		lvstate.Sink(input_chunk, input_idx);
+void WindowValueExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+                                   CollectionPtr collection) const {
+	if (child_idx != DConstants::INVALID_INDEX && wexpr.ignore_nulls) {
+		auto &gvstate = gstate.Cast<WindowValueGlobalState>();
+		gvstate.ignore_nulls = &collection->validities[child_idx];
 	}
 
-	WindowExecutor::Sink(input_chunk, input_idx, total_count, gstate, lstate);
+	WindowExecutor::Finalize(gstate, lstate, collection);
 }
 
 unique_ptr<WindowExecutorLocalState> WindowValueExecutor::GetLocalState(const WindowExecutorGlobalState &gstate) const {
@@ -1550,7 +1618,7 @@ unique_ptr<WindowExecutorLocalState> WindowValueExecutor::GetLocalState(const Wi
 }
 
 void WindowNtileExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                           Vector &result, idx_t count, idx_t row_idx) const {
+                                           DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const {
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
 	auto &cursor = *lvstate.cursor;
 	auto partition_begin = FlatVector::GetData<const idx_t>(lvstate.bounds.data[PARTITION_BEGIN]);
@@ -1599,30 +1667,13 @@ void WindowNtileExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, Wi
 //===--------------------------------------------------------------------===//
 class WindowLeadLagLocalState : public WindowValueLocalState {
 public:
-	explicit WindowLeadLagLocalState(const WindowValueGlobalState &gstate)
-	    : WindowValueLocalState(gstate),
-	      leadlag_offset(gstate.executor.wexpr.offset_expr.get(), gstate.executor.context),
-	      leadlag_default(gstate.executor.wexpr.default_expr.get(), gstate.executor.context) {
+	explicit WindowLeadLagLocalState(const WindowValueGlobalState &gstate) : WindowValueLocalState(gstate) {
 	}
-
-	void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range) override;
-
-public:
-	// LEAD/LAG Evaluation
-	WindowInputExpression leadlag_offset;
-	WindowInputExpression leadlag_default;
 };
 
-void WindowLeadLagLocalState::UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range) {
-	// Evaluate the row-level arguments
-	leadlag_offset.Execute(input_chunk);
-	leadlag_default.Execute(input_chunk);
-
-	WindowExecutorBoundsState::UpdateBounds(row_idx, input_chunk, range);
-}
-
-WindowLeadLagExecutor::WindowLeadLagExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowValueExecutor(wexpr, context) {
+WindowLeadLagExecutor::WindowLeadLagExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                             WindowSharedExpressions &shared)
+    : WindowValueExecutor(wexpr, context, shared) {
 }
 
 unique_ptr<WindowExecutorLocalState>
@@ -1632,12 +1683,15 @@ WindowLeadLagExecutor::GetLocalState(const WindowExecutorGlobalState &gstate) co
 }
 
 void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                             Vector &result, idx_t count, idx_t row_idx) const {
+                                             DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const {
 	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
 	auto &ignore_nulls = gvstate.ignore_nulls;
 	auto &llstate = lstate.Cast<WindowLeadLagLocalState>();
 	auto &cursor = *llstate.cursor;
 
+	WindowInputExpression leadlag_offset(eval_chunk, offset_idx);
+	WindowInputExpression leadlag_default(eval_chunk, default_idx);
+
 	bool can_shift = ignore_nulls->AllValid();
 	if (wexpr.offset_expr) {
 		can_shift = can_shift && wexpr.offset_expr->IsFoldable();
@@ -1652,7 +1706,7 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 	for (idx_t i = 0; i < count;) {
 		int64_t offset = 1;
 		if (wexpr.offset_expr) {
-			offset = llstate.leadlag_offset.GetCell<int64_t>(i);
+			offset = leadlag_offset.GetCell<int64_t>(i);
 		}
 		int64_t val_idx = (int64_t)row_idx;
 		if (wexpr.type == ExpressionType::WINDOW_LEAD) {
@@ -1693,7 +1747,7 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 				}
 			} else if (wexpr.default_expr) {
 				const auto width = MinValue(delta, target_limit);
-				llstate.leadlag_default.CopyCell(result, i, width);
+				leadlag_default.CopyCell(result, i, width);
 				i += width;
 				row_idx += width;
 			} else {
@@ -1705,7 +1759,7 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 			if (!delta) {
 				cursor.CopyCell(0, NumericCast<idx_t>(val_idx), result, i);
 			} else if (wexpr.default_expr) {
-				llstate.leadlag_default.CopyCell(result, i);
+				leadlag_default.CopyCell(result, i);
 			} else {
 				FlatVector::SetNull(result, i, true);
 			}
@@ -1715,12 +1769,14 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 	}
 }
 
-WindowFirstValueExecutor::WindowFirstValueExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowValueExecutor(wexpr, context) {
+WindowFirstValueExecutor::WindowFirstValueExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                                   WindowSharedExpressions &shared)
+    : WindowValueExecutor(wexpr, context, shared) {
 }
 
 void WindowFirstValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                                Vector &result, idx_t count, idx_t row_idx) const {
+                                                DataChunk &eval_chunk, Vector &result, idx_t count,
+                                                idx_t row_idx) const {
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
 	auto &cursor = *lvstate.cursor;
 	auto window_begin = FlatVector::GetData<const idx_t>(lvstate.bounds.data[WINDOW_BEGIN]);
@@ -1750,12 +1806,14 @@ void WindowFirstValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstat
 	}
 }
 
-WindowLastValueExecutor::WindowLastValueExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowValueExecutor(wexpr, context) {
+WindowLastValueExecutor::WindowLastValueExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                                 WindowSharedExpressions &shared)
+    : WindowValueExecutor(wexpr, context, shared) {
 }
 
 void WindowLastValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                               Vector &result, idx_t count, idx_t row_idx) const {
+                                               DataChunk &eval_chunk, Vector &result, idx_t count,
+                                               idx_t row_idx) const {
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
 	auto &cursor = *lvstate.cursor;
 	auto window_begin = FlatVector::GetData<const idx_t>(lvstate.bounds.data[WINDOW_BEGIN]);
@@ -1784,17 +1842,19 @@ void WindowLastValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate
 	}
 }
 
-WindowNthValueExecutor::WindowNthValueExecutor(BoundWindowExpression &wexpr, ClientContext &context)
-    : WindowValueExecutor(wexpr, context) {
+WindowNthValueExecutor::WindowNthValueExecutor(BoundWindowExpression &wexpr, ClientContext &context,
+                                               WindowSharedExpressions &shared)
+    : WindowValueExecutor(wexpr, context, shared) {
 }
 
 void WindowNthValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
-                                              Vector &result, idx_t count, idx_t row_idx) const {
+                                              DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const {
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
 	auto &cursor = *lvstate.cursor;
-	D_ASSERT(cursor.chunk.ColumnCount() == 2);
+	D_ASSERT(cursor.chunk.ColumnCount() == 1);
 	auto window_begin = FlatVector::GetData<const idx_t>(lvstate.bounds.data[WINDOW_BEGIN]);
 	auto window_end = FlatVector::GetData<const idx_t>(lvstate.bounds.data[WINDOW_END]);
+	WindowInputExpression nth_col(eval_chunk, nth_idx);
 	for (idx_t i = 0; i < count; ++i, ++row_idx) {
 
 		if (lvstate.exclusion_filter) {
@@ -1807,10 +1867,10 @@ void WindowNthValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 		}
 		// Returns value evaluated at the row that is the n'th row of the window frame (counting from 1);
 		// returns NULL if there is no such row.
-		if (cursor.CellIsNull(1, row_idx)) {
+		if (nth_col.CellIsNull(row_idx)) {
 			FlatVector::SetNull(result, i, true);
 		} else {
-			auto n_param = cursor.GetCell<int64_t>(1, row_idx);
+			auto n_param = nth_col.GetCell<int64_t>(row_idx);
 			if (n_param < 1) {
 				FlatVector::SetNull(result, i, true);
 			} else {
diff --git a/src/execution/window_segment_tree.cpp b/src/execution/window_segment_tree.cpp
index 930b526083ae..383b7b990e24 100644
--- a/src/execution/window_segment_tree.cpp
+++ b/src/execution/window_segment_tree.cpp
@@ -23,11 +23,9 @@ WindowAggregatorState::WindowAggregatorState() : allocator(Allocator::DefaultAll
 class WindowAggregatorGlobalState : public WindowAggregatorState {
 public:
 	WindowAggregatorGlobalState(ClientContext &context, const WindowAggregator &aggregator_p, idx_t group_count)
-	    : aggregator(aggregator_p),
-	      winputs(BufferManager::GetBufferManager(context), group_count, aggregator.arg_types), all_valid(true),
-	      locals(0), finalized(0) {
+	    : aggregator(aggregator_p), aggr(aggregator.wexpr), locals(0), finalized(0) {
 
-		if (aggregator.aggr.filter) {
+		if (aggr.filter) {
 			// 	Start with all invalid and set the ones that pass
 			filter_mask.Initialize(group_count, false);
 		}
@@ -36,11 +34,8 @@ class WindowAggregatorGlobalState : public WindowAggregatorState {
 	//! The aggregator data
 	const WindowAggregator &aggregator;
 
-	//! Partition data chunk
-	WindowCollection winputs;
-
-	//! Validity flag for the entire collection
-	atomic<bool> all_valid;
+	//! The aggregate function
+	const AggregateObject aggr;
 
 	//! The filtered rows in inputs.
 	ValidityArray filter_mask;
@@ -57,22 +52,33 @@ class WindowAggregatorGlobalState : public WindowAggregatorState {
 
 class WindowAggregatorLocalState : public WindowAggregatorState {
 public:
+	using CollectionPtr = optional_ptr<WindowCollection>;
+
 	WindowAggregatorLocalState() {
 	}
 
-	void Sink(WindowAggregatorGlobalState &gastate, DataChunk &input, idx_t row_idx);
-	virtual void Finalize(WindowAggregatorGlobalState &gastate);
+	void Sink(WindowAggregatorGlobalState &gastate, DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t row_idx);
+	virtual void Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection);
 
-	//! The state used for appending to the collection
-	unique_ptr<WindowBuilder> appender;
 	//! The state used for reading the collection
 	unique_ptr<WindowCursor> cursor;
 };
 
-WindowAggregator::WindowAggregator(AggregateObject aggr_p, const vector<LogicalType> &arg_types_p,
-                                   const LogicalType &result_type_p, const WindowExcludeMode exclude_mode_p)
-    : aggr(std::move(aggr_p)), arg_types(arg_types_p), result_type(result_type_p),
-      state_size(aggr.function.state_size(aggr.function)), exclude_mode(exclude_mode_p) {
+WindowAggregator::WindowAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode_p)
+    : wexpr(wexpr), aggr(wexpr), result_type(wexpr.return_type), state_size(aggr.function.state_size(aggr.function)),
+      exclude_mode(exclude_mode_p) {
+
+	for (auto &child : wexpr.children) {
+		arg_types.emplace_back(child->return_type);
+	}
+}
+
+WindowAggregator::WindowAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode_p,
+                                   WindowSharedExpressions &shared)
+    : WindowAggregator(wexpr, exclude_mode_p) {
+	for (auto &child : wexpr.children) {
+		child_idx.emplace_back(shared.RegisterCollection(child, false));
+	}
 }
 
 WindowAggregator::~WindowAggregator() {
@@ -83,21 +89,16 @@ unique_ptr<WindowAggregatorState> WindowAggregator::GetGlobalState(ClientContext
 	return make_uniq<WindowAggregatorGlobalState>(context, *this, group_count);
 }
 
-void WindowAggregatorLocalState::Sink(WindowAggregatorGlobalState &gastate, DataChunk &arg_chunk, idx_t input_idx) {
-	auto &winputs = gastate.winputs;
-	if (!winputs.GetTypes().empty()) {
-		if (!appender) {
-			appender = make_uniq<WindowBuilder>(winputs);
-		}
-		appender->Sink(arg_chunk, input_idx);
-	}
+void WindowAggregatorLocalState::Sink(WindowAggregatorGlobalState &gastate, DataChunk &sink_chunk,
+                                      DataChunk &coll_chunk, idx_t input_idx) {
 }
 
-void WindowAggregator::Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &arg_chunk,
-                            idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
+void WindowAggregator::Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &sink_chunk,
+                            DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+                            idx_t filtered) {
 	auto &gastate = gstate.Cast<WindowAggregatorGlobalState>();
 	auto &lastate = lstate.Cast<WindowAggregatorLocalState>();
-	lastate.Sink(gastate, arg_chunk, input_idx);
+	lastate.Sink(gastate, sink_chunk, coll_chunk, input_idx);
 	if (filter_sel) {
 		auto &filter_mask = gastate.filter_mask;
 		for (idx_t f = 0; f < filtered; ++f) {
@@ -106,22 +107,18 @@ void WindowAggregator::Sink(WindowAggregatorState &gstate, WindowAggregatorState
 	}
 }
 
-void WindowAggregatorLocalState::Finalize(WindowAggregatorGlobalState &gastate) {
-	auto &winputs = gastate.winputs;
-
-	// Thread-safe and idempotent
-	winputs.Combine(false);
-
+void WindowAggregatorLocalState::Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) {
 	// Prepare to scan
 	if (!cursor) {
-		cursor = make_uniq<WindowCursor>(winputs);
+		cursor = make_uniq<WindowCursor>(*collection, gastate.aggregator.child_idx);
 	}
 }
 
-void WindowAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) {
+void WindowAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+                                const FrameStats &stats) {
 	auto &gasink = gstate.Cast<WindowAggregatorGlobalState>();
 	auto &lastate = lstate.Cast<WindowAggregatorLocalState>();
-	lastate.Finalize(gasink);
+	lastate.Finalize(gasink, collection);
 }
 
 //===--------------------------------------------------------------------===//
@@ -232,7 +229,8 @@ class WindowConstantAggregatorLocalState : public WindowAggregatorLocalState {
 	~WindowConstantAggregatorLocalState() override {
 	}
 
-	void Sink(DataChunk &payload_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered);
+	void Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+	          idx_t filtered);
 	void Combine(WindowConstantAggregatorGlobalState &gstate);
 
 public:
@@ -240,6 +238,8 @@ class WindowConstantAggregatorLocalState : public WindowAggregatorLocalState {
 	const WindowConstantAggregatorGlobalState &gstate;
 	//! Reusable chunk for sinking
 	DataChunk inputs;
+	//! Chunk for referencing the input columns
+	DataChunk payload_chunk;
 	//! A vector of pointers to "state", used for intermediate window segment aggregation
 	Vector statep;
 	//! Reused result state container for the window functions
@@ -254,7 +254,7 @@ WindowConstantAggregatorGlobalState::WindowConstantAggregatorGlobalState(ClientC
                                                                          const WindowConstantAggregator &aggregator,
                                                                          idx_t group_count,
                                                                          const ValidityMask &partition_mask)
-    : WindowAggregatorGlobalState(context, aggregator, STANDARD_VECTOR_SIZE), statef(aggregator.aggr) {
+    : WindowAggregatorGlobalState(context, aggregator, STANDARD_VECTOR_SIZE), statef(aggr) {
 
 	// Locate the partition boundaries
 	if (partition_mask.AllValid()) {
@@ -304,14 +304,20 @@ WindowConstantAggregatorLocalState::WindowConstantAggregatorLocalState(
 
 	// Set up shared buffer
 	inputs.Initialize(Allocator::DefaultAllocator(), aggregator.arg_types);
+	payload_chunk.InitializeEmpty(inputs.GetTypes());
 
 	gstate.locals++;
 }
 
-WindowConstantAggregator::WindowConstantAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types,
-                                                   const LogicalType &result_type,
-                                                   const WindowExcludeMode exclude_mode_p)
-    : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode_p) {
+WindowConstantAggregator::WindowConstantAggregator(const BoundWindowExpression &wexpr,
+                                                   const WindowExcludeMode exclude_mode_p,
+                                                   WindowSharedExpressions &shared)
+    : WindowAggregator(wexpr, exclude_mode_p) {
+
+	// We only need these values for Sink
+	for (auto &child : wexpr.children) {
+		child_idx.emplace_back(shared.RegisterSink(child));
+	}
 }
 
 unique_ptr<WindowAggregatorState> WindowConstantAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
@@ -319,20 +325,20 @@ unique_ptr<WindowAggregatorState> WindowConstantAggregator::GetGlobalState(Clien
 	return make_uniq<WindowConstantAggregatorGlobalState>(context, *this, group_count, partition_mask);
 }
 
-void WindowConstantAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk,
-                                    idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
+void WindowConstantAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &sink_chunk,
+                                    DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+                                    idx_t filtered) {
 	auto &lastate = lstate.Cast<WindowConstantAggregatorLocalState>();
 
-	lastate.Sink(arg_chunk, input_idx, filter_sel, filtered);
+	lastate.Sink(sink_chunk, coll_chunk, input_idx, filter_sel, filtered);
 }
 
-void WindowConstantAggregatorLocalState::Sink(DataChunk &payload_chunk, idx_t row,
+void WindowConstantAggregatorLocalState::Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t row,
                                               optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
 	auto &partition_offsets = gstate.partition_offsets;
-	auto &aggregator = gstate.aggregator;
-	const auto &aggr = aggregator.aggr;
+	const auto &aggr = gstate.aggr;
 	const auto chunk_begin = row;
-	const auto chunk_end = chunk_begin + payload_chunk.size();
+	const auto chunk_end = chunk_begin + sink_chunk.size();
 	idx_t partition =
 	    idx_t(std::upper_bound(partition_offsets.begin(), partition_offsets.end(), row) - partition_offsets.begin()) -
 	    1;
@@ -340,6 +346,11 @@ void WindowConstantAggregatorLocalState::Sink(DataChunk &payload_chunk, idx_t ro
 	auto state_f_data = statef.GetData();
 	auto state_p_data = FlatVector::GetData<data_ptr_t>(statep);
 
+	auto &child_idx = gstate.aggregator.child_idx;
+	for (column_t c = 0; c < child_idx.size(); ++c) {
+		payload_chunk.data[c].Reference(sink_chunk.data[child_idx[c]]);
+	}
+
 	AggregateInputData aggr_input_data(aggr.GetFunctionData(), allocator);
 	idx_t begin = 0;
 	idx_t filter_idx = 0;
@@ -407,7 +418,7 @@ void WindowConstantAggregatorLocalState::Sink(DataChunk &payload_chunk, idx_t ro
 }
 
 void WindowConstantAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate,
-                                        const FrameStats &stats) {
+                                        CollectionPtr collection, const FrameStats &stats) {
 	auto &gastate = gstate.Cast<WindowConstantAggregatorGlobalState>();
 	auto &lastate = lstate.Cast<WindowConstantAggregatorLocalState>();
 
@@ -469,9 +480,9 @@ void WindowConstantAggregator::Evaluate(const WindowAggregatorState &gsink, Wind
 //===--------------------------------------------------------------------===//
 // WindowCustomAggregator
 //===--------------------------------------------------------------------===//
-WindowCustomAggregator::WindowCustomAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types,
-                                               const LogicalType &result_type, const WindowExcludeMode exclude_mode)
-    : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode) {
+WindowCustomAggregator::WindowCustomAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode,
+                                               WindowSharedExpressions &shared)
+    : WindowAggregator(wexpr, exclude_mode, shared) {
 }
 
 WindowCustomAggregator::~WindowCustomAggregator() {
@@ -484,7 +495,7 @@ class WindowCustomAggregatorState : public WindowAggregatorLocalState {
 
 public:
 	//! The aggregate function
-	const AggregateObject &aggr;
+	const AggregateObject aggr;
 	//! Data pointer that contains a single state, shared by all the custom evaluators
 	vector<data_t> state;
 	//! Reused result state container for the window functions
@@ -516,7 +527,7 @@ class WindowCustomAggregatorGlobalState : public WindowAggregatorGlobalState {
 	                                           idx_t group_count)
 	    : WindowAggregatorGlobalState(context, aggregator, group_count), context(context) {
 
-		gcstate = make_uniq<WindowCustomAggregatorState>(aggregator.aggr, aggregator.exclude_mode);
+		gcstate = make_uniq<WindowCustomAggregatorState>(aggr, aggregator.exclude_mode);
 	}
 
 	//! Buffer manager for paging custom accelerator data
@@ -551,26 +562,29 @@ unique_ptr<WindowAggregatorState> WindowCustomAggregator::GetGlobalState(ClientC
 	return make_uniq<WindowCustomAggregatorGlobalState>(context, *this, group_count);
 }
 
-void WindowCustomAggregator::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate,
-                                      const FrameStats &stats) {
+void WindowCustomAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate,
+                                      CollectionPtr collection, const FrameStats &stats) {
 	//	Single threaded Finalize for now
-	auto &gcsink = gsink.Cast<WindowCustomAggregatorGlobalState>();
+	auto &gcsink = gstate.Cast<WindowCustomAggregatorGlobalState>();
 	lock_guard<mutex> gestate_guard(gcsink.lock);
 	if (gcsink.finalized) {
 		return;
 	}
 
-	WindowAggregator::Finalize(gsink, lstate, stats);
+	WindowAggregator::Finalize(gstate, lstate, collection, stats);
 
-	auto inputs = gcsink.winputs.inputs.get();
-	const auto count = gcsink.winputs.size();
-	const auto all_valid = gcsink.winputs.all_valid.load();
+	auto inputs = collection->inputs.get();
+	const auto count = collection->size();
+	vector<bool> all_valids;
+	for (auto col_idx : child_idx) {
+		all_valids.push_back(collection->all_valids[col_idx]);
+	}
 	auto &filter_mask = gcsink.filter_mask;
 	auto &filter_packed = gcsink.filter_packed;
 	filter_mask.Pack(filter_packed, filter_mask.target_count);
 
 	gcsink.partition_input =
-	    make_uniq<WindowPartitionInput>(gcsink.context, inputs, count, all_valid, filter_packed, stats);
+	    make_uniq<WindowPartitionInput>(gcsink.context, inputs, count, child_idx, all_valids, filter_packed, stats);
 
 	if (aggr.function.window_init) {
 		auto &gcstate = *gcsink.gcstate;
@@ -664,9 +678,9 @@ void WindowCustomAggregator::Evaluate(const WindowAggregatorState &gsink, Window
 //===--------------------------------------------------------------------===//
 // WindowNaiveAggregator
 //===--------------------------------------------------------------------===//
-WindowNaiveAggregator::WindowNaiveAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types,
-                                             const LogicalType &result_type, const WindowExcludeMode exclude_mode)
-    : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode) {
+WindowNaiveAggregator::WindowNaiveAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode,
+                                             WindowSharedExpressions &shared)
+    : WindowAggregator(wexpr, exclude_mode, shared) {
 }
 
 WindowNaiveAggregator::~WindowNaiveAggregator() {
@@ -675,34 +689,32 @@ WindowNaiveAggregator::~WindowNaiveAggregator() {
 class WindowNaiveState : public WindowAggregatorLocalState {
 public:
 	struct HashRow {
-		explicit HashRow(WindowNaiveState &state, const WindowCollection &winputs) : state(state), winputs(winputs) {
+		explicit HashRow(WindowNaiveState &state) : state(state) {
 		}
 
 		inline size_t operator()(const idx_t &i) const {
-			return state.Hash(winputs, i);
+			return state.Hash(i);
 		}
 
 		WindowNaiveState &state;
-		const WindowCollection &winputs;
 	};
 
 	struct EqualRow {
-		EqualRow(WindowNaiveState &state, const WindowCollection &winputs) : state(state), winputs(winputs) {
+		explicit EqualRow(WindowNaiveState &state) : state(state) {
 		}
 
 		inline bool operator()(const idx_t &lhs, const idx_t &rhs) const {
-			return state.KeyEqual(winputs, lhs, rhs);
+			return state.KeyEqual(lhs, rhs);
 		}
 
 		WindowNaiveState &state;
-		const WindowCollection &winputs;
 	};
 
 	using RowSet = std::unordered_set<idx_t, HashRow, EqualRow>;
 
 	explicit WindowNaiveState(const WindowNaiveAggregator &gsink);
 
-	void Finalize(WindowAggregatorGlobalState &gastate) override;
+	void Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) override;
 
 	void Evaluate(const WindowAggregatorGlobalState &gsink, const DataChunk &bounds, Vector &result, idx_t count,
 	              idx_t row_idx);
@@ -712,9 +724,9 @@ class WindowNaiveState : public WindowAggregatorLocalState {
 	void FlushStates(const WindowAggregatorGlobalState &gsink);
 
 	//! Hashes a value for the hash table
-	size_t Hash(const WindowCollection &inputs, idx_t rid);
+	size_t Hash(idx_t rid);
 	//! Compares two values for the hash table
-	bool KeyEqual(const WindowCollection &inputs, const idx_t &lhs, const idx_t &rhs);
+	bool KeyEqual(const idx_t &lhs, const idx_t &rhs);
 
 	//! The global state
 	const WindowNaiveAggregator &aggregator;
@@ -757,12 +769,12 @@ WindowNaiveState::WindowNaiveState(const WindowNaiveAggregator &aggregator_p)
 	}
 }
 
-void WindowNaiveState::Finalize(WindowAggregatorGlobalState &gastate) {
-	WindowAggregatorLocalState::Finalize(gastate);
+void WindowNaiveState::Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) {
+	WindowAggregatorLocalState::Finalize(gastate, collection);
 
 	//	Set up the comparison scanner just in case
 	if (!comparer) {
-		comparer = make_uniq<WindowCursor>(gastate.winputs);
+		comparer = make_uniq<WindowCursor>(*collection, gastate.aggregator.child_idx);
 	}
 }
 
@@ -774,14 +786,14 @@ void WindowNaiveState::FlushStates(const WindowAggregatorGlobalState &gsink) {
 	auto &scanned = cursor->chunk;
 	leaves.Slice(scanned, update_sel, flush_count);
 
-	auto &aggr = aggregator.aggr;
+	const auto &aggr = gsink.aggr;
 	AggregateInputData aggr_input_data(aggr.GetFunctionData(), allocator);
 	aggr.function.update(leaves.data.data(), aggr_input_data, leaves.ColumnCount(), statep, flush_count);
 
 	flush_count = 0;
 }
 
-size_t WindowNaiveState::Hash(const WindowCollection &winputs, idx_t rid) {
+size_t WindowNaiveState::Hash(idx_t rid) {
 	D_ASSERT(cursor->RowIsVisible(rid));
 	auto s = cursor->RowOffset(rid);
 	auto &scanned = cursor->chunk;
@@ -792,7 +804,7 @@ size_t WindowNaiveState::Hash(const WindowCollection &winputs, idx_t rid) {
 	return *FlatVector::GetData<hash_t>(hashes);
 }
 
-bool WindowNaiveState::KeyEqual(const WindowCollection &winputs, const idx_t &lidx, const idx_t &ridx) {
+bool WindowNaiveState::KeyEqual(const idx_t &lidx, const idx_t &ridx) {
 	//	One of the indices will be scanned, so make it the left one
 	auto lhs = lidx;
 	auto rhs = ridx;
@@ -831,10 +843,9 @@ bool WindowNaiveState::KeyEqual(const WindowCollection &winputs, const idx_t &li
 
 void WindowNaiveState::Evaluate(const WindowAggregatorGlobalState &gsink, const DataChunk &bounds, Vector &result,
                                 idx_t count, idx_t row_idx) {
-	auto &aggr = aggregator.aggr;
+	const auto &aggr = gsink.aggr;
 	auto &filter_mask = gsink.filter_mask;
-	auto &winputs = gsink.winputs;
-	auto &types = winputs.GetTypes();
+	const auto types = cursor->chunk.GetTypes();
 
 	if (leaves.ColumnCount() == 0 && !types.empty()) {
 		leaves.Initialize(Allocator::DefaultAllocator(), types);
@@ -843,8 +854,8 @@ void WindowNaiveState::Evaluate(const WindowAggregatorGlobalState &gsink, const
 	auto fdata = FlatVector::GetData<data_ptr_t>(statef);
 	auto pdata = FlatVector::GetData<data_ptr_t>(statep);
 
-	HashRow hash_row(*this, winputs);
-	EqualRow equal_row(*this, winputs);
+	HashRow hash_row(*this);
+	EqualRow equal_row(*this);
 	RowSet row_set(STANDARD_VECTOR_SIZE, hash_row, equal_row);
 
 	EvaluateSubFrames(bounds, aggregator.exclude_mode, count, row_idx, frames, [&](idx_t rid) {
@@ -940,10 +951,9 @@ class WindowSegmentTreeGlobalState : public WindowAggregatorGlobalState {
 	static constexpr idx_t TREE_FANOUT = 16;
 };
 
-WindowSegmentTree::WindowSegmentTree(AggregateObject aggr, const vector<LogicalType> &arg_types,
-                                     const LogicalType &result_type, WindowAggregationMode mode_p,
-                                     const WindowExcludeMode exclude_mode_p)
-    : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode_p), mode(mode_p) {
+WindowSegmentTree::WindowSegmentTree(const BoundWindowExpression &wexpr, WindowAggregationMode mode_p,
+                                     const WindowExcludeMode exclude_mode_p, WindowSharedExpressions &shared)
+    : WindowAggregator(wexpr, exclude_mode_p, shared), mode(mode_p) {
 }
 
 class WindowSegmentTreePart {
@@ -953,12 +963,12 @@ class WindowSegmentTreePart {
 
 	enum FramePart : uint8_t { FULL = 0, LEFT = 1, RIGHT = 2 };
 
-	WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr, const WindowCollection &inputs,
+	WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr, unique_ptr<WindowCursor> cursor,
 	                      const ValidityArray &filter_mask);
 	~WindowSegmentTreePart();
 
 	unique_ptr<WindowSegmentTreePart> Copy() const {
-		return make_uniq<WindowSegmentTreePart>(allocator, aggr, inputs, filter_mask);
+		return make_uniq<WindowSegmentTreePart>(allocator, aggr, cursor->Copy(), filter_mask);
 	}
 
 	void FlushStates(bool combining);
@@ -989,8 +999,6 @@ class WindowSegmentTreePart {
 	const AggregateObject &aggr;
 	//! Order insensitive aggregate (we can optimise internal combines)
 	const bool order_insensitive;
-	//! The partition arguments
-	const WindowCollection &inputs;
 	//! The filtered rows in inputs
 	const ValidityArray &filter_mask;
 	//! The size of a single aggregate state
@@ -1020,7 +1028,7 @@ class WindowSegmentTreeState : public WindowAggregatorLocalState {
 	WindowSegmentTreeState() {
 	}
 
-	void Finalize(WindowAggregatorGlobalState &gastate) override;
+	void Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) override;
 	void Evaluate(const WindowSegmentTreeGlobalState &gsink, const DataChunk &bounds, Vector &result, idx_t count,
 	              idx_t row_idx);
 	//! The left (default) segment tree part
@@ -1029,29 +1037,23 @@ class WindowSegmentTreeState : public WindowAggregatorLocalState {
 	unique_ptr<WindowSegmentTreePart> right_part;
 };
 
-void WindowSegmentTree::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate, const FrameStats &stats) {
+void WindowSegmentTree::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate, CollectionPtr collection,
+                                 const FrameStats &stats) {
+	WindowAggregator::Finalize(gsink, lstate, collection, stats);
 
 	auto &gasink = gsink.Cast<WindowSegmentTreeGlobalState>();
-	auto &types = gasink.winputs.GetTypes();
-
-	WindowAggregator::Finalize(gsink, lstate, stats);
-
-	if (!types.empty()) {
-		if (aggr.function.combine && UseCombineAPI()) {
-			lstate.Cast<WindowSegmentTreeState>().Finalize(gasink);
-		}
-	}
-
 	++gasink.finalized;
 }
 
 WindowSegmentTreePart::WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr,
-                                             const WindowCollection &inputs, const ValidityArray &filter_mask)
+                                             unique_ptr<WindowCursor> cursor_p, const ValidityArray &filter_mask)
     : allocator(allocator), aggr(aggr),
-      order_insensitive(aggr.function.order_dependent == AggregateOrderDependent::NOT_ORDER_DEPENDENT), inputs(inputs),
+      order_insensitive(aggr.function.order_dependent == AggregateOrderDependent::NOT_ORDER_DEPENDENT),
       filter_mask(filter_mask), state_size(aggr.function.state_size(aggr.function)),
-      state(state_size * STANDARD_VECTOR_SIZE), statep(LogicalType::POINTER), statel(LogicalType::POINTER),
-      statef(LogicalType::POINTER), flush_count(0) {
+      state(state_size * STANDARD_VECTOR_SIZE), cursor(std::move(cursor_p)), statep(LogicalType::POINTER),
+      statel(LogicalType::POINTER), statef(LogicalType::POINTER), flush_count(0) {
+
+	auto &inputs = cursor->chunk;
 	if (inputs.ColumnCount() > 0) {
 		leaves.Initialize(Allocator::DefaultAllocator(), inputs.GetTypes());
 		filter_sel.Initialize();
@@ -1067,11 +1069,6 @@ WindowSegmentTreePart::WindowSegmentTreePart(ArenaAllocator &allocator, const Ag
 		fdata[i] = state_ptr;
 		state_ptr += state_size;
 	}
-
-	// Set up for scanning
-	if (!cursor) {
-		cursor = make_uniq<WindowCursor>(inputs);
-	}
 }
 
 WindowSegmentTreePart::~WindowSegmentTreePart() {
@@ -1141,6 +1138,7 @@ void WindowSegmentTreePart::ExtractFrame(idx_t begin, idx_t end, data_ptr_t stat
 void WindowSegmentTreePart::WindowSegmentValue(const WindowSegmentTreeGlobalState &tree, idx_t l_idx, idx_t begin,
                                                idx_t end, data_ptr_t state_ptr) {
 	D_ASSERT(begin <= end);
+	auto &inputs = cursor->chunk;
 	if (begin == end || inputs.ColumnCount() == 0) {
 		return;
 	}
@@ -1186,10 +1184,9 @@ void WindowSegmentTreePart::Finalize(Vector &result, idx_t count) {
 
 WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(ClientContext &context, const WindowSegmentTree &aggregator,
                                                            idx_t group_count)
-    : WindowAggregatorGlobalState(context, aggregator, group_count), tree(aggregator),
-      levels_flat_native(aggregator.aggr) {
+    : WindowAggregatorGlobalState(context, aggregator, group_count), tree(aggregator), levels_flat_native(aggr) {
 
-	D_ASSERT(!winputs.GetTypes().empty());
+	D_ASSERT(!aggregator.wexpr.children.empty());
 
 	// compute space required to store internal nodes of segment tree
 	levels_flat_start.push_back(0);
@@ -1230,15 +1227,15 @@ WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(ClientContext &contex
 	}
 }
 
-void WindowSegmentTreeState::Finalize(WindowAggregatorGlobalState &gastate) {
-	WindowAggregatorLocalState::Finalize(gastate);
+void WindowSegmentTreeState::Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) {
+	WindowAggregatorLocalState::Finalize(gastate, collection);
 
 	//	Single part for constructing the tree
 	auto &gstate = gastate.Cast<WindowSegmentTreeGlobalState>();
-	auto &inputs = gstate.winputs;
-	auto &tree = gstate.tree;
+	auto cursor = make_uniq<WindowCursor>(*collection, gastate.aggregator.child_idx);
+	const auto leaf_count = collection->size();
 	auto &filter_mask = gstate.filter_mask;
-	WindowSegmentTreePart gtstate(gstate.CreateTreeAllocator(), tree.aggr, inputs, filter_mask);
+	WindowSegmentTreePart gtstate(gstate.CreateTreeAllocator(), gastate.aggr, std::move(cursor), filter_mask);
 
 	auto &levels_flat_native = gstate.levels_flat_native;
 	const auto &levels_flat_start = gstate.levels_flat_start;
@@ -1251,8 +1248,7 @@ void WindowSegmentTreeState::Finalize(WindowAggregatorGlobalState &gastate) {
 
 		// level 0 is data itself
 		const auto level_size =
-		    (level_current == 0 ? inputs.size()
-		                        : levels_flat_start[level_current] - levels_flat_start[level_current - 1]);
+		    (level_current == 0 ? leaf_count : levels_flat_start[level_current] - levels_flat_start[level_current - 1]);
 		if (level_size <= 1) {
 			break;
 		}
@@ -1301,8 +1297,7 @@ void WindowSegmentTreeState::Evaluate(const WindowSegmentTreeGlobalState &gtstat
 	auto peer_end = FlatVector::GetData<const idx_t>(bounds.data[PEER_END]);
 
 	if (!part) {
-		auto &inputs = gtstate.winputs;
-		part = make_uniq<WindowSegmentTreePart>(allocator, gtstate.aggregator.aggr, inputs, gtstate.filter_mask);
+		part = make_uniq<WindowSegmentTreePart>(allocator, gtstate.aggr, cursor->Copy(), gtstate.filter_mask);
 	}
 
 	if (gtstate.aggregator.exclude_mode != WindowExcludeMode::NO_OTHER) {
@@ -1500,10 +1495,10 @@ void WindowSegmentTreePart::EvaluateLeaves(const WindowSegmentTreeGlobalState &t
 //===--------------------------------------------------------------------===//
 // WindowDistinctAggregator
 //===--------------------------------------------------------------------===//
-WindowDistinctAggregator::WindowDistinctAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types,
-                                                   const LogicalType &result_type,
-                                                   const WindowExcludeMode exclude_mode_p, ClientContext &context)
-    : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode_p), context(context) {
+WindowDistinctAggregator::WindowDistinctAggregator(const BoundWindowExpression &wexpr,
+                                                   const WindowExcludeMode exclude_mode_p,
+                                                   WindowSharedExpressions &shared, ClientContext &context)
+    : WindowAggregator(wexpr, exclude_mode_p, shared), context(context) {
 }
 
 class WindowDistinctAggregatorLocalState;
@@ -1592,7 +1587,7 @@ WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(ClientC
                                                                          idx_t group_count)
     : WindowAggregatorGlobalState(context, aggregator, group_count), context(aggregator.context),
       stage(PartitionSortStage::INIT), tasks_completed(0), merge_sort_tree(*this, group_count),
-      levels_flat_native(aggregator.aggr) {
+      levels_flat_native(aggr) {
 	payload_types.emplace_back(LogicalType::UBIGINT);
 
 	//	1:	functionComputePrevIdcs(𝑖𝑛)
@@ -1651,7 +1646,9 @@ class WindowDistinctAggregatorLocalState : public WindowAggregatorLocalState {
 public:
 	explicit WindowDistinctAggregatorLocalState(const WindowDistinctAggregatorGlobalState &aggregator);
 
-	void Sink(DataChunk &arg_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered);
+	void Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+	          idx_t filtered);
+	void Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) override;
 	void Sorted();
 	void ExecuteTask();
 	void Evaluate(const WindowDistinctAggregatorGlobalState &gdstate, const DataChunk &bounds, Vector &result,
@@ -1693,7 +1690,7 @@ class WindowDistinctAggregatorLocalState : public WindowAggregatorLocalState {
 WindowDistinctAggregatorLocalState::WindowDistinctAggregatorLocalState(
     const WindowDistinctAggregatorGlobalState &gastate)
     : update_v(LogicalType::POINTER), source_v(LogicalType::POINTER), target_v(LogicalType::POINTER), gastate(gastate),
-      statef(gastate.aggregator.aggr), statep(LogicalType::POINTER), statel(LogicalType::POINTER), flush_count(0) {
+      statef(gastate.aggr), statep(LogicalType::POINTER), statel(LogicalType::POINTER), flush_count(0) {
 	InitSubFrames(frames, gastate.aggregator.exclude_mode);
 	payload_chunk.Initialize(Allocator::DefaultAllocator(), gastate.payload_types);
 
@@ -1703,11 +1700,6 @@ WindowDistinctAggregatorLocalState::WindowDistinctAggregatorLocalState(
 	sort_chunk.Initialize(Allocator::DefaultAllocator(), gastate.sort_types);
 	sort_chunk.data.back().Reference(payload_chunk.data[0]);
 
-	//! Input data chunk, used for leaf segment aggregation
-	auto &inputs = gastate.winputs;
-	leaves.Initialize(Allocator::DefaultAllocator(), inputs.GetTypes());
-	sel.Initialize();
-
 	gastate.locals++;
 }
 
@@ -1716,29 +1708,33 @@ unique_ptr<WindowAggregatorState> WindowDistinctAggregator::GetGlobalState(Clien
 	return make_uniq<WindowDistinctAggregatorGlobalState>(context, *this, group_count);
 }
 
-void WindowDistinctAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk,
-                                    idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
-	WindowAggregator::Sink(gsink, lstate, arg_chunk, input_idx, filter_sel, filtered);
+void WindowDistinctAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &sink_chunk,
+                                    DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+                                    idx_t filtered) {
+	WindowAggregator::Sink(gsink, lstate, sink_chunk, coll_chunk, input_idx, filter_sel, filtered);
 
 	auto &ldstate = lstate.Cast<WindowDistinctAggregatorLocalState>();
-	ldstate.Sink(arg_chunk, input_idx, filter_sel, filtered);
+	ldstate.Sink(sink_chunk, coll_chunk, input_idx, filter_sel, filtered);
 }
 
-void WindowDistinctAggregatorLocalState::Sink(DataChunk &arg_chunk, idx_t input_idx,
+void WindowDistinctAggregatorLocalState::Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t input_idx,
                                               optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
 	//	3: 	for i ← 0 to in.size do
 	//	4: 		sorted[i] ← (in[i], i)
-	const auto count = arg_chunk.size();
+	const auto count = sink_chunk.size();
 	payload_chunk.Reset();
 	auto &sorted_vec = payload_chunk.data[0];
 	auto sorted = FlatVector::GetData<idx_t>(sorted_vec);
 	std::iota(sorted, sorted + count, input_idx);
 
-	for (column_t c = 0; c < arg_chunk.ColumnCount(); ++c) {
-		sort_chunk.data[c].Reference(arg_chunk.data[c]);
+	// Our arguments are being fully materialised,
+	// but we also need them as sort keys.
+	auto &child_idx = gastate.aggregator.child_idx;
+	for (column_t c = 0; c < child_idx.size(); ++c) {
+		sort_chunk.data[c].Reference(coll_chunk.data[child_idx[c]]);
 	}
 	sort_chunk.data.back().Reference(sorted_vec);
-	sort_chunk.SetCardinality(arg_chunk);
+	sort_chunk.SetCardinality(sink_chunk);
 	payload_chunk.SetCardinality(sort_chunk);
 
 	//	Apply FILTER clause, if any
@@ -1754,6 +1750,14 @@ void WindowDistinctAggregatorLocalState::Sink(DataChunk &arg_chunk, idx_t input_
 	}
 }
 
+void WindowDistinctAggregatorLocalState::Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) {
+	WindowAggregatorLocalState::Finalize(gastate, collection);
+
+	//! Input data chunk, used for leaf segment aggregation
+	leaves.Initialize(Allocator::DefaultAllocator(), cursor->chunk.GetTypes());
+	sel.Initialize();
+}
+
 void WindowDistinctAggregatorLocalState::ExecuteTask() {
 	auto &global_sort = *gastate.global_sort;
 	switch (stage) {
@@ -1864,10 +1868,10 @@ bool WindowDistinctAggregatorGlobalState::TryPrepareNextStage(WindowDistinctAggr
 }
 
 void WindowDistinctAggregator::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate,
-                                        const FrameStats &stats) {
+                                        CollectionPtr collection, const FrameStats &stats) {
 	auto &gdsink = gsink.Cast<WindowDistinctAggregatorGlobalState>();
 	auto &ldstate = lstate.Cast<WindowDistinctAggregatorLocalState>();
-	ldstate.Finalize(gdsink);
+	ldstate.Finalize(gdsink, collection);
 
 	//	5: Sort sorted lexicographically increasing
 	ldstate.ExecuteTask();
@@ -2031,7 +2035,7 @@ void WindowDistinctSortTree::Build(WindowDistinctAggregatorLocalState &ldastate)
 }
 
 void WindowDistinctSortTree::BuildRun(idx_t level_nr, idx_t run_idx, WindowDistinctAggregatorLocalState &ldastate) {
-	auto &aggr = gdastate.aggregator.aggr;
+	auto &aggr = gdastate.aggr;
 	auto &allocator = gdastate.allocator;
 	auto &inputs = ldastate.cursor->chunk;
 	auto &levels_flat_native = gdastate.levels_flat_native;
@@ -2136,7 +2140,7 @@ void WindowDistinctAggregatorLocalState::FlushStates() {
 		return;
 	}
 
-	const auto &aggr = gastate.aggregator.aggr;
+	const auto &aggr = gastate.aggr;
 	AggregateInputData aggr_input_data(aggr.GetFunctionData(), allocator);
 	statel.Verify(flush_count);
 	aggr.function.combine(statel, statep, aggr_input_data, flush_count);
diff --git a/src/function/aggregate/distributive/count.cpp b/src/function/aggregate/distributive/count.cpp
index d4e837aaa00d..55c0a2a93d76 100644
--- a/src/function/aggregate/distributive/count.cpp
+++ b/src/function/aggregate/distributive/count.cpp
@@ -36,7 +36,7 @@ struct CountStarFunction : public BaseCountFunction {
 	template <typename RESULT_TYPE>
 	static void Window(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition, const_data_ptr_t,
 	                   data_ptr_t l_state, const SubFrames &frames, Vector &result, idx_t rid) {
-		D_ASSERT(partition.inputs == nullptr);
+		D_ASSERT(partition.column_ids.empty());
 
 		auto data = FlatVector::GetData<RESULT_TYPE>(result);
 		RESULT_TYPE total = 0;
diff --git a/src/include/duckdb/execution/window_executor.hpp b/src/include/duckdb/execution/window_executor.hpp
index b64c8ad74cca..1622f5c64c60 100644
--- a/src/include/duckdb/execution/window_executor.hpp
+++ b/src/include/duckdb/execution/window_executor.hpp
@@ -20,6 +20,7 @@ class WindowCollection {
 public:
 	using ColumnDataCollectionPtr = unique_ptr<ColumnDataCollection>;
 	using ColumnDataCollectionSpec = pair<idx_t, optional_ptr<ColumnDataCollection>>;
+	using ColumnSet = unordered_set<column_t>;
 
 	WindowCollection(BufferManager &buffer_manager, idx_t count, const vector<LogicalType> &types);
 
@@ -38,21 +39,20 @@ class WindowCollection {
 	//! Update a thread-local collection for appending data to a given row
 	void GetCollection(idx_t row_idx, ColumnDataCollectionSpec &spec);
 	//! Single-threaded, idempotent ordered combining of all the appended data.
-	void Combine(bool build_validity);
+	void Combine(const ColumnSet &build_validity);
 
 	//! The collection data. May be null if the column count is 0.
 	ColumnDataCollectionPtr inputs;
 	//! Global validity mask
-	atomic<bool> all_valid;
+	vector<atomic<bool>> all_valids;
 	//! Optional validity mask for the entire collection
-	ValidityMask validity;
+	vector<ValidityMask> validities;
 
 	//! The collection columns
 	const vector<LogicalType> types;
 	//! The collection rows
 	const idx_t count;
 
-private:
 	//! Guard for range updates
 	mutex lock;
 	//! The paging buffer manager to use
@@ -84,7 +84,8 @@ class WindowBuilder {
 
 class WindowCursor {
 public:
-	explicit WindowCursor(const WindowCollection &paged);
+	WindowCursor(const WindowCollection &paged, column_t col_idx);
+	WindowCursor(const WindowCollection &paged, vector<column_t> column_ids);
 
 	//! Is the scan in range?
 	inline bool RowIsVisible(idx_t row_idx) const {
@@ -131,6 +132,10 @@ class WindowCursor {
 		VectorOperations::Copy(source, target, index + 1, index, target_offset);
 	}
 
+	unique_ptr<WindowCursor> Copy() const {
+		return make_uniq<WindowCursor>(paged, state.column_ids);
+	}
+
 	//! The pageable data
 	const WindowCollection &paged;
 	//! The state used for reading the collection
@@ -139,66 +144,67 @@ class WindowCursor {
 	DataChunk chunk;
 };
 
-struct WindowInputExpression {
-	static void PrepareInputExpression(Expression &expr, ExpressionExecutor &executor, DataChunk &chunk) {
-		vector<LogicalType> types;
-		types.push_back(expr.return_type);
-		executor.AddExpression(expr);
-
-		auto &allocator = executor.GetAllocator();
-		chunk.Initialize(allocator, types);
-	}
+//	Column indexes of the bounds chunk
+enum WindowBounds : uint8_t { PARTITION_BEGIN, PARTITION_END, PEER_BEGIN, PEER_END, WINDOW_BEGIN, WINDOW_END };
 
-	WindowInputExpression(optional_ptr<Expression> expr_p, ClientContext &context)
-	    : expr(expr_p), ptype(PhysicalType::INVALID), scalar(true), executor(context) {
-		if (expr) {
-			PrepareInputExpression(*expr, executor, chunk);
-			ptype = expr->return_type.InternalType();
-			scalar = expr->IsScalar();
+//! A shared set of expressions
+struct WindowSharedExpressions {
+	struct Shared {
+		column_t size = 0;
+		expression_map_t<vector<column_t>> columns;
+	};
+
+	//! Register a shared expression in a shared set
+	static column_t RegisterExpr(const unique_ptr<Expression> &expr, Shared &shared);
+
+	//! Register a shared collection expression
+	column_t RegisterCollection(const unique_ptr<Expression> &expr, bool build_validity) {
+		auto result = RegisterExpr(expr, coll_shared);
+		if (build_validity) {
+			coll_validity.insert(result);
 		}
+		return result;
 	}
-
-	void Execute(DataChunk &input_chunk) {
-		if (expr) {
-			chunk.Reset();
-			executor.Execute(input_chunk, chunk);
-			chunk.Verify();
-			chunk.Flatten();
-		}
+	//! Register a shared collection expression
+	inline column_t RegisterSink(const unique_ptr<Expression> &expr) {
+		return RegisterExpr(expr, sink_shared);
+	}
+	//! Register a shared evaluation expression
+	inline column_t RegisterEvaluate(const unique_ptr<Expression> &expr) {
+		return RegisterExpr(expr, eval_shared);
 	}
 
-	template <typename T>
-	inline T GetCell(idx_t i) const {
-		D_ASSERT(!chunk.data.empty());
-		const auto data = FlatVector::GetData<T>(chunk.data[0]);
-		return data[scalar ? 0 : i];
+	//! Expression layout
+	static vector<const Expression *> GetSortedExpressions(Shared &shared);
+
+	//! Expression execution utility
+	static void PrepareExecutors(Shared &shared, ExpressionExecutor &exec, DataChunk &chunk);
+
+	//! Prepare collection expressions
+	inline void PrepareCollection(ExpressionExecutor &exec, DataChunk &chunk) {
+		PrepareExecutors(coll_shared, exec, chunk);
 	}
 
-	inline bool CellIsNull(idx_t i) const {
-		D_ASSERT(!chunk.data.empty());
-		if (chunk.data[0].GetVectorType() == VectorType::CONSTANT_VECTOR) {
-			return ConstantVector::IsNull(chunk.data[0]);
-		}
-		return FlatVector::IsNull(chunk.data[0], i);
+	//! Prepare collection expressions
+	inline void PrepareSink(ExpressionExecutor &exec, DataChunk &chunk) {
+		PrepareExecutors(sink_shared, exec, chunk);
 	}
 
-	inline void CopyCell(Vector &target, idx_t target_offset, idx_t width = 1) const {
-		D_ASSERT(!chunk.data.empty());
-		auto &source = chunk.data[0];
-		auto source_offset = scalar ? 0 : target_offset;
-		VectorOperations::Copy(source, target, source_offset + width, source_offset, target_offset);
+	//! Prepare collection expressions
+	inline void PrepareEvaluate(ExpressionExecutor &exec, DataChunk &chunk) {
+		PrepareExecutors(eval_shared, exec, chunk);
 	}
 
-	optional_ptr<Expression> expr;
-	PhysicalType ptype;
-	bool scalar;
-	ExpressionExecutor executor;
-	DataChunk chunk;
+	//! Fully materialised shared expressions
+	Shared coll_shared;
+	//! Sink shared expressions
+	Shared sink_shared;
+	//! Evaluate shared expressions
+	Shared eval_shared;
+	//! Requested collection validity masks
+	unordered_set<column_t> coll_validity;
 };
 
-//	Column indexes of the bounds chunk
-enum WindowBounds : uint8_t { PARTITION_BEGIN, PARTITION_END, PEER_BEGIN, PEER_END, WINDOW_BEGIN, WINDOW_END };
-
 class WindowExecutorState {
 public:
 	WindowExecutorState() {};
@@ -221,6 +227,8 @@ class WindowExecutor;
 
 class WindowExecutorGlobalState : public WindowExecutorState {
 public:
+	using CollectionPtr = optional_ptr<WindowCollection>;
+
 	WindowExecutorGlobalState(const WindowExecutor &executor, const idx_t payload_count,
 	                          const ValidityMask &partition_mask, const ValidityMask &order_mask);
 
@@ -230,35 +238,26 @@ class WindowExecutorGlobalState : public WindowExecutorState {
 	const ValidityMask &partition_mask;
 	const ValidityMask &order_mask;
 	vector<LogicalType> arg_types;
-
-	// evaluate RANGE expressions, if needed
-	optional_ptr<Expression> range_expr;
-	unique_ptr<WindowCollection> range;
 };
 
 class WindowExecutorLocalState : public WindowExecutorState {
 public:
-	explicit WindowExecutorLocalState(const WindowExecutorGlobalState &gstate);
+	using CollectionPtr = optional_ptr<WindowCollection>;
 
-	void Sink(WindowExecutorGlobalState &gstate, DataChunk &input_chunk, idx_t input_idx);
-	virtual void Finalize(WindowExecutorGlobalState &gstate);
+	explicit WindowExecutorLocalState(const WindowExecutorGlobalState &gstate);
 
-	// Argument evaluation
-	ExpressionExecutor payload_executor;
-	DataChunk payload_chunk;
+	void Sink(WindowExecutorGlobalState &gstate, DataChunk &sink_chunk, DataChunk &coll_chunk, idx_t input_idx);
+	virtual void Finalize(WindowExecutorGlobalState &gstate, CollectionPtr collection);
 
-	//! Range evaluation
-	ExpressionExecutor range_executor;
-	DataChunk range_chunk;
-	//! The state used for building the range collection
-	unique_ptr<WindowBuilder> range_builder;
 	//! The state used for reading the range collection
 	unique_ptr<WindowCursor> range_cursor;
 };
 
 class WindowExecutor {
 public:
-	WindowExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	using CollectionPtr = optional_ptr<WindowCollection>;
+
+	WindowExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 	virtual ~WindowExecutor() {
 	}
 
@@ -266,30 +265,45 @@ class WindowExecutor {
 	GetGlobalState(const idx_t payload_count, const ValidityMask &partition_mask, const ValidityMask &order_mask) const;
 	virtual unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const;
 
-	virtual void Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
+	virtual void Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, const idx_t input_idx,
 	                  WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const;
 
-	virtual void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const;
+	virtual void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+	                      CollectionPtr collection) const;
 
-	void Evaluate(idx_t row_idx, DataChunk &input_chunk, Vector &result, WindowExecutorLocalState &lstate,
+	void Evaluate(idx_t row_idx, DataChunk &eval_chunk, Vector &result, WindowExecutorLocalState &lstate,
 	              WindowExecutorGlobalState &gstate) const;
 
 	// The function
 	const BoundWindowExpression &wexpr;
 	ClientContext &context;
 
+	// evaluate frame expressions, if needed
+	column_t boundary_start_idx = DConstants::INVALID_INDEX;
+	column_t boundary_end_idx = DConstants::INVALID_INDEX;
+
+	// evaluate RANGE expressions, if needed
+	optional_ptr<Expression> range_expr;
+	column_t range_idx = DConstants::INVALID_INDEX;
+
 protected:
-	virtual void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                              idx_t count, idx_t row_idx) const = 0;
+	virtual void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+	                              DataChunk &eval_chunk, Vector &result, idx_t count, idx_t row_idx) const = 0;
 };
 
 class WindowAggregateExecutor : public WindowExecutor {
 public:
-	WindowAggregateExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowAggregationMode mode);
+	WindowAggregateExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared,
+	                        WindowAggregationMode mode);
+
+	bool IsConstantAggregate();
+	bool IsCustomAggregate();
+	bool IsDistinctAggregate();
 
-	void Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count, WindowExecutorGlobalState &gstate,
+	void Sink(DataChunk &sink_chunk, DataChunk &coll_chunk, const idx_t input_idx, WindowExecutorGlobalState &gstate,
 	          WindowExecutorLocalState &lstate) const override;
-	void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const override;
+	void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+	              CollectionPtr collection) const override;
 
 	unique_ptr<WindowExecutorGlobalState> GetGlobalState(const idx_t payload_count, const ValidityMask &partition_mask,
 	                                                     const ValidityMask &order_mask) const override;
@@ -297,121 +311,136 @@ class WindowAggregateExecutor : public WindowExecutor {
 
 	const WindowAggregationMode mode;
 
+	// aggregate computation algorithm
+	unique_ptr<WindowAggregator> aggregator;
+
+	// FILTER reference expression in sink_chunk
+	unique_ptr<Expression> filter_ref;
+
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowRowNumberExecutor : public WindowExecutor {
 public:
-	WindowRowNumberExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowRowNumberExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 //	Base class for non-aggregate functions that use peer boundaries
 class WindowRankExecutor : public WindowExecutor {
 public:
-	WindowRankExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowRankExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 	unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const override;
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowDenseRankExecutor : public WindowExecutor {
 public:
-	WindowDenseRankExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowDenseRankExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 	unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const override;
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowPercentRankExecutor : public WindowExecutor {
 public:
-	WindowPercentRankExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowPercentRankExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 	unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const override;
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowCumeDistExecutor : public WindowExecutor {
 public:
-	WindowCumeDistExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowCumeDistExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 // Base class for non-aggregate functions that have a payload
 class WindowValueExecutor : public WindowExecutor {
 public:
-	WindowValueExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowValueExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
-	void Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count, WindowExecutorGlobalState &gstate,
-	          WindowExecutorLocalState &lstate) const override;
+	void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
+	              CollectionPtr collection) const override;
 
 	unique_ptr<WindowExecutorGlobalState> GetGlobalState(const idx_t payload_count, const ValidityMask &partition_mask,
 	                                                     const ValidityMask &order_mask) const override;
 	unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const override;
+
+	//! The column index of the value column
+	column_t child_idx = DConstants::INVALID_INDEX;
+	//! The column index of the Nth column
+	column_t nth_idx = DConstants::INVALID_INDEX;
+	//! The column index of the offset column
+	column_t offset_idx = DConstants::INVALID_INDEX;
+	//! The column index of the default value column
+	column_t default_idx = DConstants::INVALID_INDEX;
 };
 
 //
 class WindowNtileExecutor : public WindowValueExecutor {
 public:
-	WindowNtileExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowNtileExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 class WindowLeadLagExecutor : public WindowValueExecutor {
 public:
-	WindowLeadLagExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowLeadLagExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 	unique_ptr<WindowExecutorLocalState> GetLocalState(const WindowExecutorGlobalState &gstate) const override;
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowFirstValueExecutor : public WindowValueExecutor {
 public:
-	WindowFirstValueExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowFirstValueExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowLastValueExecutor : public WindowValueExecutor {
 public:
-	WindowLastValueExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowLastValueExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 class WindowNthValueExecutor : public WindowValueExecutor {
 public:
-	WindowNthValueExecutor(BoundWindowExpression &wexpr, ClientContext &context);
+	WindowNthValueExecutor(BoundWindowExpression &wexpr, ClientContext &context, WindowSharedExpressions &shared);
 
 protected:
-	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, Vector &result,
-	                      idx_t count, idx_t row_idx) const override;
+	void EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate, DataChunk &eval_chunk,
+	                      Vector &result, idx_t count, idx_t row_idx) const override;
 };
 
 } // namespace duckdb
diff --git a/src/include/duckdb/execution/window_segment_tree.hpp b/src/include/duckdb/execution/window_segment_tree.hpp
index b9b41954f7a8..7df2a77342e9 100644
--- a/src/include/duckdb/execution/window_segment_tree.hpp
+++ b/src/include/duckdb/execution/window_segment_tree.hpp
@@ -18,6 +18,9 @@
 
 namespace duckdb {
 
+struct WindowSharedExpressions;
+class WindowCollection;
+
 class WindowAggregatorState {
 public:
 	WindowAggregatorState();
@@ -41,8 +44,11 @@ class WindowAggregatorState {
 
 class WindowAggregator {
 public:
-	WindowAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types_p, const LogicalType &result_type_p,
-	                 const WindowExcludeMode exclude_mode_p);
+	using CollectionPtr = optional_ptr<WindowCollection>;
+
+	WindowAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode_p);
+	WindowAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode_p,
+	                 WindowSharedExpressions &shared);
 	virtual ~WindowAggregator();
 
 	//	Threading states
@@ -51,31 +57,36 @@ class WindowAggregator {
 	virtual unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const = 0;
 
 	//	Build
-	virtual void Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &arg_chunk,
-	                  idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered);
-	virtual void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats);
+	virtual void Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &sink_chunk,
+	                  DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered);
+	virtual void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+	                      const FrameStats &stats);
 
 	//	Probe
 	virtual void Evaluate(const WindowAggregatorState &gsink, WindowAggregatorState &lstate, const DataChunk &bounds,
 	                      Vector &result, idx_t count, idx_t row_idx) const = 0;
 
+	//! The window function
+	const BoundWindowExpression &wexpr;
 	//! A description of the aggregator
 	const AggregateObject aggr;
 	//! The argument types for the function
-	const vector<LogicalType> arg_types;
+	vector<LogicalType> arg_types;
 	//! The result type of the window function
 	const LogicalType result_type;
 	//! The size of a single aggregate state
 	const idx_t state_size;
 	//! The window exclusion clause
 	const WindowExcludeMode exclude_mode;
+	//! Partition collection column indicies
+	vector<column_t> child_idx;
 };
 
 // Used for validation
 class WindowNaiveAggregator : public WindowAggregator {
 public:
-	WindowNaiveAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types_p,
-	                      const LogicalType &result_type_p, const WindowExcludeMode exclude_mode);
+	WindowNaiveAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode,
+	                      WindowSharedExpressions &shared);
 	~WindowNaiveAggregator() override;
 
 	unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const override;
@@ -85,16 +96,18 @@ class WindowNaiveAggregator : public WindowAggregator {
 
 class WindowConstantAggregator : public WindowAggregator {
 public:
-	WindowConstantAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types_p,
-	                         const LogicalType &result_type_p, WindowExcludeMode exclude_mode_p);
+	WindowConstantAggregator(const BoundWindowExpression &wexpr, WindowExcludeMode exclude_mode_p,
+	                         WindowSharedExpressions &shared);
 	~WindowConstantAggregator() override {
 	}
 
 	unique_ptr<WindowAggregatorState> GetGlobalState(ClientContext &context, idx_t group_count,
 	                                                 const ValidityMask &partition_mask) const override;
-	void Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &arg_chunk, idx_t input_idx,
-	          optional_ptr<SelectionVector> filter_sel, idx_t filtered) override;
-	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) override;
+	void Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &sink_chunk,
+	          DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
+	          idx_t filtered) override;
+	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+	              const FrameStats &stats) override;
 
 	unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const override;
 	void Evaluate(const WindowAggregatorState &gsink, WindowAggregatorState &lstate, const DataChunk &bounds,
@@ -103,13 +116,14 @@ class WindowConstantAggregator : public WindowAggregator {
 
 class WindowCustomAggregator : public WindowAggregator {
 public:
-	WindowCustomAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types_p,
-	                       const LogicalType &result_type_p, const WindowExcludeMode exclude_mode);
+	WindowCustomAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode,
+	                       WindowSharedExpressions &shared);
 	~WindowCustomAggregator() override;
 
 	unique_ptr<WindowAggregatorState> GetGlobalState(ClientContext &context, idx_t group_count,
 	                                                 const ValidityMask &partition_mask) const override;
-	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) override;
+	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+	              const FrameStats &stats) override;
 
 	unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const override;
 	void Evaluate(const WindowAggregatorState &gsink, WindowAggregatorState &lstate, const DataChunk &bounds,
@@ -119,13 +133,14 @@ class WindowCustomAggregator : public WindowAggregator {
 class WindowSegmentTree : public WindowAggregator {
 
 public:
-	WindowSegmentTree(AggregateObject aggr, const vector<LogicalType> &arg_types_p, const LogicalType &result_type_p,
-	                  WindowAggregationMode mode_p, const WindowExcludeMode exclude_mode);
+	WindowSegmentTree(const BoundWindowExpression &wexpr, WindowAggregationMode mode_p,
+	                  const WindowExcludeMode exclude_mode, WindowSharedExpressions &shared);
 
 	unique_ptr<WindowAggregatorState> GetGlobalState(ClientContext &context, idx_t group_count,
 	                                                 const ValidityMask &partition_mask) const override;
 	unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const override;
-	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) override;
+	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+	              const FrameStats &stats) override;
 
 	void Evaluate(const WindowAggregatorState &gstate, WindowAggregatorState &lstate, const DataChunk &bounds,
 	              Vector &result, idx_t count, idx_t row_idx) const override;
@@ -142,16 +157,16 @@ class WindowSegmentTree : public WindowAggregator {
 
 class WindowDistinctAggregator : public WindowAggregator {
 public:
-	WindowDistinctAggregator(AggregateObject aggr, const vector<LogicalType> &arg_types_p,
-	                         const LogicalType &result_type_p, const WindowExcludeMode exclude_mode_p,
-	                         ClientContext &context);
+	WindowDistinctAggregator(const BoundWindowExpression &wexpr, const WindowExcludeMode exclude_mode_p,
+	                         WindowSharedExpressions &shared, ClientContext &context);
 
 	//	Build
 	unique_ptr<WindowAggregatorState> GetGlobalState(ClientContext &context, idx_t group_count,
 	                                                 const ValidityMask &partition_mask) const override;
-	void Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk, idx_t input_idx,
-	          optional_ptr<SelectionVector> filter_sel, idx_t filtered) override;
-	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) override;
+	void Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &sink_chunk, DataChunk &coll_chunk,
+	          idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered) override;
+	void Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
+	              const FrameStats &stats) override;
 
 	//	Evaluate
 	unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const override;
diff --git a/src/include/duckdb/function/aggregate_function.hpp b/src/include/duckdb/function/aggregate_function.hpp
index ab674cdb4987..c054040e6fa5 100644
--- a/src/include/duckdb/function/aggregate_function.hpp
+++ b/src/include/duckdb/function/aggregate_function.hpp
@@ -35,14 +35,17 @@ using FrameStats = array<FrameDelta, 2>;
 //! but the row count will still be valid
 class ColumnDataCollection;
 struct WindowPartitionInput {
-	WindowPartitionInput(ClientContext &context, const ColumnDataCollection *inputs, idx_t count, bool all_valid,
-	                     const ValidityMask &filter_mask, const FrameStats &stats)
-	    : context(context), inputs(inputs), count(count), all_valid(all_valid), filter_mask(filter_mask), stats(stats) {
+	WindowPartitionInput(ClientContext &context, const ColumnDataCollection *inputs, idx_t count,
+	                     vector<column_t> &column_ids, vector<bool> &all_valid, const ValidityMask &filter_mask,
+	                     const FrameStats &stats)
+	    : context(context), inputs(inputs), count(count), column_ids(column_ids), all_valid(all_valid),
+	      filter_mask(filter_mask), stats(stats) {
 	}
 	ClientContext &context;
 	const ColumnDataCollection *inputs;
 	idx_t count;
-	bool all_valid;
+	vector<column_t> column_ids;
+	vector<bool> all_valid;
 	const ValidityMask &filter_mask;
 	const FrameStats stats;
 };


Test patch:
diff --git a/test/fuzzer/sqlsmith/window-leadlag-overflow.test b/test/fuzzer/sqlsmith/window-leadlag-overflow.test
index b50646169201..2328f6627246 100644
--- a/test/fuzzer/sqlsmith/window-leadlag-overflow.test
+++ b/test/fuzzer/sqlsmith/window-leadlag-overflow.test
@@ -9,7 +9,7 @@ WITH all_types AS (
 SELECT lag(c16, COLUMNS(*)) OVER (ROWS BETWEEN 1768 FOLLOWING AND UNBOUNDED FOLLOWING) 
 FROM all_types AS t42(c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14, c15, c16, c17, c18, c19, c20, c21, c22, c23, c24, c25, c26, c27, c28, c29, c30, c31, c32, c33, c34, c35, c36, c37, c38, c39, c40, c41)
 ----
-Overflow in subtraction
+the value is out of range for the destination type
 
 statement ok
 SELECT lead('1e668c84-6cbc-4d41-843e-970c17446f9e', 8479, 3087) OVER (PARTITION BY 9136)
diff --git a/test/sql/window/test_ignore_nulls.test b/test/sql/window/test_ignore_nulls.test
index b879a7a4e3ec..27f4d6b704fe 100644
--- a/test/sql/window/test_ignore_nulls.test
+++ b/test/sql/window/test_ignore_nulls.test
@@ -303,6 +303,28 @@ ORDER BY id
 ----
 40000 values hashing to c302c8b0f3c10c1e5cc7211c4af7a8d6
 
+# Independent INGORE/RESPECT NULLS with shared input.
+query III
+SELECT 
+	v, 
+	lead(v) OVER (ORDER BY id), 
+	lead(v IGNORE NULLS) OVER (ORDER BY id) 
+FROM (VALUES 
+	(1, 1), 
+	(2, NULL), 
+	(3, 2), 
+	(4, NULL), 
+	(5, 3), 
+	(6, NULL)
+) tbl(id, v);
+----
+1	NULL	2
+NULL	2	2
+2	NULL	3
+NULL	3	3
+3	NULL	NULL
+NULL	NULL	NULL
+
 #
 # Unsupported
 #
diff --git a/test/sql/window/test_volatile_independence.test b/test/sql/window/test_volatile_independence.test
new file mode 100644
index 000000000000..faa6fa5bf1b3
--- /dev/null
+++ b/test/sql/window/test_volatile_independence.test
@@ -0,0 +1,18 @@
+# name: test/sql/window/test_volatile_independence.test
+# description: Window Sharing should distiguish identical volatile expressions
+# group: [window]
+
+require skip_reload
+
+statement ok
+SELECT SETSEED(0.8675309);
+
+query II
+SELECT 
+	list(random()) OVER (ORDER BY id), 
+	max(random()) OVER (ORDER BY id) 
+FROM range(3) t(id);
+----
+[0.23450047366512405]	0.772592377754351
+[0.23450047366512405, 0.4512114749041355]	0.772592377754351
+[0.23450047366512405, 0.4512114749041355, 0.5199990716061366]	0.9123614504583814