Repository: duckdb/duckdb (language: python)

Issue:
selecting struct column using parquet_scan returns varchar instead of struct
#### What happens?
If reading a nested column from a parquet file using the `parquet_scan` function, the data is returned as a varchar. But if I read the parquet file into an arrow table, and then query against the arrow table, the proper struct is returned.

#### To Reproduce
```
import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
from os.path import exists

def create_parquet():
    sch = pa.schema([
        pa.field("col1", pa.string()),
        pa.field("col2", pa.struct([
            pa.field("nested", pa.string()),
        ]))
    ])

    data={
        "col1": ["A", "B", "C"],
        "col2": [{"nested": "A"}, {"nested": "B"}, {"nested": "C"}]
    }

    tbl = pa.table(data, schema=sch)
    pq.write_table(table=tbl, where="testdata.parquet")


if __name__ == '__main__':
    if not exists("testdata.parquet"):
        create_parquet()
    con = duckdb.connect()
    print(con.execute("select col2 from parquet_scan('testdata.parquet')").fetchall())


    arrow = duckdb.arrow(pq.read_table("testdata.parquet"))
    print(arrow.query('arrow', "select col2 from arrow").fetchall())
```

#### Environment (please complete the following information):
 - OS: Ubuntu 20.04
 - DuckDB Version: 0.3.1
 - DuckDB Client: python-3.8, arrow-6.0.1

#### Before Submitting

- [ ] **Have you tried this on the latest `master` branch?** : **No**
* **Python**: `pip install duckdb --upgrade --pre`
* **R**: `install.packages("https://github.com/duckdb/duckdb/releases/download/master-builds/duckdb_r_src.tar.gz", repos = NULL)`
* **Other Platforms**: You can find binaries [here](https://github.com/duckdb/duckdb/releases/tag/master-builds) or compile from source.

- [x ] **Have you tried the steps to reproduce? Do they include all relevant data and configuration? Does the issue you report still appear there?**



Discussion / hints:
This appears to have been fixed in the latest master already. Running this on the latest master gives me the following results:

```python
print(con.execute("select col2 from parquet_scan('testdata.parquet')").fetchall())
# [({'nested': 'A'},), ({'nested': 'B'},), ({'nested': 'C'},)]
print(arrow.query('arrow', "select col2 from arrow").fetchall())
# [({'nested': 'A'},), ({'nested': 'B'},), ({'nested': 'C'},)]
```

Could you verify that these are the expected result and that this has been fixed already by installing the pre-release and testing again?
Verified that this is working in the pre-release. Thanks!

Fix patch:
diff --git a/extension/core_functions/aggregate/holistic/mad.cpp b/extension/core_functions/aggregate/holistic/mad.cpp
index 6aa518dac963..94b7d7097f4a 100644
--- a/extension/core_functions/aggregate/holistic/mad.cpp
+++ b/extension/core_functions/aggregate/holistic/mad.cpp
@@ -190,12 +190,21 @@ struct MedianAbsoluteDeviationOperation : QuantileOperation {
 	}
 
 	template <class STATE, class INPUT_TYPE, class RESULT_TYPE>
-	static void Window(const INPUT_TYPE *data, const ValidityMask &fmask, const ValidityMask &dmask,
-	                   AggregateInputData &aggr_input_data, STATE &state, const SubFrames &frames, Vector &result,
-	                   idx_t ridx, const STATE *gstate) {
+	static void Window(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
+	                   const_data_ptr_t g_state, data_ptr_t l_state, const SubFrames &frames, Vector &result,
+	                   idx_t ridx) {
+		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);
+		const auto &fmask = partition.filter_mask;
+
 		auto rdata = FlatVector::GetData<RESULT_TYPE>(result);
 
-		QuantileIncluded included(fmask, dmask);
+		QuantileIncluded<INPUT_TYPE> included(fmask, data);
 		const auto n = FrameSize(included, frames);
 
 		if (!n) {
@@ -263,7 +272,7 @@ AggregateFunction GetTypedMedianAbsoluteDeviationAggregateFunction(const Logical
 	fun.bind = BindMAD;
 	fun.order_dependent = AggregateOrderDependent::NOT_ORDER_DEPENDENT;
 #ifndef DUCKDB_SMALLER_BINARY
-	fun.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, TARGET_TYPE, OP>;
+	fun.window = OP::template Window<STATE, INPUT_TYPE, TARGET_TYPE>;
 	fun.window_init = OP::template WindowInit<STATE, INPUT_TYPE>;
 #endif
 	return fun;
diff --git a/extension/core_functions/aggregate/holistic/mode.cpp b/extension/core_functions/aggregate/holistic/mode.cpp
index cc2ae9ad2d38..2c0de366c1f1 100644
--- a/extension/core_functions/aggregate/holistic/mode.cpp
+++ b/extension/core_functions/aggregate/holistic/mode.cpp
@@ -2,6 +2,7 @@
 #include "duckdb/common/uhugeint.hpp"
 #include "duckdb/common/vector_operations/vector_operations.hpp"
 #include "duckdb/common/operator/comparison_operators.hpp"
+#include "duckdb/common/types/column/column_data_collection.hpp"
 #include "core_functions/aggregate/distributive_functions.hpp"
 #include "core_functions/aggregate/holistic_functions.hpp"
 #include "duckdb/planner/expression/bound_aggregate_expression.hpp"
@@ -100,6 +101,17 @@ struct ModeState {
 	bool valid = false;
 	size_t count = 0;
 
+	//! The collection being read
+	const ColumnDataCollection *inputs;
+	//! The state used for reading the collection on this thread
+	ColumnDataScanState scan;
+	//! The data chunk paged into into
+	DataChunk page;
+	//! The data pointer
+	const KEY_TYPE *data = nullptr;
+	//! The validity mask
+	const ValidityMask *validity = nullptr;
+
 	~ModeState() {
 		if (frequency_map) {
 			delete frequency_map;
@@ -109,6 +121,43 @@ struct ModeState {
 		}
 	}
 
+	void InitializePage(const ColumnDataCollection &inputs) {
+		if (page.ColumnCount() == 0) {
+			this->inputs = &inputs;
+			inputs.InitializeScan(scan);
+			inputs.InitializeScanChunk(scan, page);
+		}
+	}
+
+	inline sel_t RowOffset(idx_t row_idx) const {
+		D_ASSERT(RowIsVisible(row_idx));
+		return UnsafeNumericCast<sel_t>(row_idx - scan.current_row_index);
+	}
+
+	inline bool RowIsVisible(idx_t row_idx) const {
+		return (row_idx < scan.next_row_index && scan.current_row_index <= row_idx);
+	}
+
+	inline idx_t Seek(idx_t row_idx) {
+		if (!RowIsVisible(row_idx)) {
+			D_ASSERT(inputs);
+			inputs->Seek(row_idx, scan, page);
+			data = FlatVector::GetData<KEY_TYPE>(page.data[0]);
+			validity = &FlatVector::Validity(page.data[0]);
+		}
+		return RowOffset(row_idx);
+	}
+
+	inline const KEY_TYPE &GetCell(idx_t row_idx) {
+		const auto offset = Seek(row_idx);
+		return data[offset];
+	}
+
+	inline bool RowIsValid(idx_t row_idx) {
+		const auto offset = Seek(row_idx);
+		return validity->RowIsValid(offset);
+	}
+
 	void Reset() {
 		if (frequency_map) {
 			frequency_map->clear();
@@ -118,7 +167,8 @@ struct ModeState {
 		valid = false;
 	}
 
-	void ModeAdd(const KEY_TYPE &key, idx_t row) {
+	void ModeAdd(idx_t row) {
+		const auto &key = GetCell(row);
 		auto &attr = (*frequency_map)[key];
 		auto new_count = (attr.count += 1);
 		if (new_count == 1) {
@@ -138,7 +188,8 @@ struct ModeState {
 		}
 	}
 
-	void ModeRm(const KEY_TYPE &key, idx_t frame) {
+	void ModeRm(idx_t frame) {
+		const auto &key = GetCell(frame);
 		auto &attr = (*frequency_map)[key];
 		auto old_count = attr.count;
 		nonzero -= size_t(old_count == 1);
@@ -164,16 +215,16 @@ struct ModeState {
 	}
 };
 
+template <typename STATE>
 struct ModeIncluded {
-	inline explicit ModeIncluded(const ValidityMask &fmask_p, const ValidityMask &dmask_p)
-	    : fmask(fmask_p), dmask(dmask_p) {
+	inline explicit ModeIncluded(const ValidityMask &fmask_p, STATE &state) : fmask(fmask_p), state(state) {
 	}
 
 	inline bool operator()(const idx_t &idx) const {
-		return fmask.RowIsValid(idx) && dmask.RowIsValid(idx);
+		return fmask.RowIsValid(idx) && state.RowIsValid(idx);
 	}
 	const ValidityMask &fmask;
-	const ValidityMask &dmask;
+	STATE &state;
 };
 
 template <typename TYPE_OP>
@@ -261,11 +312,9 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 	template <typename STATE, typename INPUT_TYPE>
 	struct UpdateWindowState {
 		STATE &state;
-		const INPUT_TYPE *data;
-		ModeIncluded &included;
+		ModeIncluded<STATE> &included;
 
-		inline UpdateWindowState(STATE &state, const INPUT_TYPE *data, ModeIncluded &included)
-		    : state(state), data(data), included(included) {
+		inline UpdateWindowState(STATE &state, ModeIncluded<STATE> &included) : state(state), included(included) {
 		}
 
 		inline void Neither(idx_t begin, idx_t end) {
@@ -274,7 +323,7 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 		inline void Left(idx_t begin, idx_t end) {
 			for (; begin < end; ++begin) {
 				if (included(begin)) {
-					state.ModeRm(data[begin], begin);
+					state.ModeRm(begin);
 				}
 			}
 		}
@@ -282,7 +331,7 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 		inline void Right(idx_t begin, idx_t end) {
 			for (; begin < end; ++begin) {
 				if (included(begin)) {
-					state.ModeAdd(data[begin], begin);
+					state.ModeAdd(begin);
 				}
 			}
 		}
@@ -292,9 +341,17 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 	};
 
 	template <class STATE, class INPUT_TYPE, class RESULT_TYPE>
-	static void Window(const INPUT_TYPE *data, const ValidityMask &fmask, const ValidityMask &dmask,
-	                   AggregateInputData &aggr_input_data, STATE &state, const SubFrames &frames, Vector &result,
-	                   idx_t rid, const STATE *gstate) {
+	static void Window(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
+	                   const_data_ptr_t g_state, data_ptr_t l_state, const SubFrames &frames, Vector &result,
+	                   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);
+		const auto &fmask = partition.filter_mask;
+
 		auto rdata = FlatVector::GetData<RESULT_TYPE>(result);
 		auto &rmask = FlatVector::Validity(result);
 		auto &prevs = state.prevs;
@@ -302,7 +359,7 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 			prevs.resize(1);
 		}
 
-		ModeIncluded included(fmask, dmask);
+		ModeIncluded<STATE> included(fmask, state);
 
 		if (!state.frequency_map) {
 			state.frequency_map = TYPE_OP::CreateEmpty(Allocator::DefaultAllocator());
@@ -315,13 +372,13 @@ struct ModeFunction : TypedModeFunction<TYPE_OP> {
 			for (const auto &frame : frames) {
 				for (auto i = frame.start; i < frame.end; ++i) {
 					if (included(i)) {
-						state.ModeAdd(data[i], i);
+						state.ModeAdd(i);
 					}
 				}
 			}
 		} else {
 			using Updater = UpdateWindowState<STATE, INPUT_TYPE>;
-			Updater updater(state, data, included);
+			Updater updater(state, included);
 			AggregateExecutor::IntersectFrames(prevs, frames, updater);
 		}
 
@@ -380,7 +437,7 @@ AggregateFunction GetTypedModeFunction(const LogicalType &type) {
 	using STATE = ModeState<INPUT_TYPE, TYPE_OP>;
 	using OP = ModeFunction<TYPE_OP>;
 	auto func = AggregateFunction::UnaryAggregateDestructor<STATE, INPUT_TYPE, INPUT_TYPE, OP>(type, type);
-	func.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, INPUT_TYPE, OP>;
+	func.window = OP::template Window<STATE, INPUT_TYPE, INPUT_TYPE>;
 	return func;
 }
 
diff --git a/extension/core_functions/aggregate/holistic/quantile.cpp b/extension/core_functions/aggregate/holistic/quantile.cpp
index 0dff6e7bec38..24caec942363 100644
--- a/extension/core_functions/aggregate/holistic/quantile.cpp
+++ b/extension/core_functions/aggregate/holistic/quantile.cpp
@@ -208,10 +208,19 @@ struct QuantileScalarOperation : public QuantileOperation {
 	}
 
 	template <class STATE, class INPUT_TYPE, class RESULT_TYPE>
-	static void Window(const INPUT_TYPE *data, const ValidityMask &fmask, const ValidityMask &dmask,
-	                   AggregateInputData &aggr_input_data, STATE &state, const SubFrames &frames, Vector &result,
-	                   idx_t ridx, const STATE *gstate) {
-		QuantileIncluded included(fmask, dmask);
+	static void Window(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
+	                   const_data_ptr_t g_state, data_ptr_t l_state, const SubFrames &frames, Vector &result,
+	                   idx_t ridx) {
+		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);
+		const auto &fmask = partition.filter_mask;
+
+		QuantileIncluded<INPUT_TYPE> included(fmask, data);
 		const auto n = FrameSize(included, frames);
 
 		D_ASSERT(aggr_input_data.bind_data);
@@ -305,13 +314,22 @@ struct QuantileListOperation : QuantileOperation {
 	}
 
 	template <class STATE, class INPUT_TYPE, class RESULT_TYPE>
-	static void Window(const INPUT_TYPE *data, const ValidityMask &fmask, const ValidityMask &dmask,
-	                   AggregateInputData &aggr_input_data, STATE &state, const SubFrames &frames, Vector &list,
-	                   idx_t lidx, const STATE *gstate) {
+	static void Window(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
+	                   const_data_ptr_t g_state, data_ptr_t l_state, const SubFrames &frames, Vector &list,
+	                   idx_t lidx) {
+		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);
+		const auto &fmask = partition.filter_mask;
+
 		D_ASSERT(aggr_input_data.bind_data);
 		auto &bind_data = aggr_input_data.bind_data->Cast<QuantileBindData>();
 
-		QuantileIncluded included(fmask, dmask);
+		QuantileIncluded<INPUT_TYPE> included(fmask, data);
 		const auto n = FrameSize(included, frames);
 
 		// Result is a constant LIST<RESULT_TYPE> with a fixed length
@@ -410,7 +428,7 @@ struct ScalarDiscreteQuantile {
 		using OP = QuantileScalarOperation<true>;
 		auto fun = AggregateFunction::UnaryAggregateDestructor<STATE, INPUT_TYPE, INPUT_TYPE, OP>(type, type);
 #ifndef DUCKDB_SMALLER_BINARY
-		fun.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, INPUT_TYPE, OP>;
+		fun.window = OP::Window<STATE, INPUT_TYPE, INPUT_TYPE>;
 		fun.window_init = OP::WindowInit<STATE, INPUT_TYPE>;
 #endif
 		return fun;
@@ -447,7 +465,7 @@ struct ListDiscreteQuantile {
 		auto fun = QuantileListAggregate<STATE, INPUT_TYPE, list_entry_t, OP>(type, type);
 		fun.order_dependent = AggregateOrderDependent::NOT_ORDER_DEPENDENT;
 #ifndef DUCKDB_SMALLER_BINARY
-		fun.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, list_entry_t, OP>;
+		fun.window = OP::template Window<STATE, INPUT_TYPE, list_entry_t>;
 		fun.window_init = OP::template WindowInit<STATE, INPUT_TYPE>;
 #endif
 		return fun;
@@ -538,7 +556,7 @@ struct ScalarContinuousQuantile {
 		    AggregateFunction::UnaryAggregateDestructor<STATE, INPUT_TYPE, TARGET_TYPE, OP>(input_type, target_type);
 		fun.order_dependent = AggregateOrderDependent::NOT_ORDER_DEPENDENT;
 #ifndef DUCKDB_SMALLER_BINARY
-		fun.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, TARGET_TYPE, OP>;
+		fun.window = OP::template Window<STATE, INPUT_TYPE, TARGET_TYPE>;
 		fun.window_init = OP::template WindowInit<STATE, INPUT_TYPE>;
 #endif
 		return fun;
@@ -553,7 +571,7 @@ struct ListContinuousQuantile {
 		auto fun = QuantileListAggregate<STATE, INPUT_TYPE, list_entry_t, OP>(input_type, target_type);
 		fun.order_dependent = AggregateOrderDependent::NOT_ORDER_DEPENDENT;
 #ifndef DUCKDB_SMALLER_BINARY
-		fun.window = AggregateFunction::UnaryWindow<STATE, INPUT_TYPE, list_entry_t, OP>;
+		fun.window = OP::template Window<STATE, INPUT_TYPE, list_entry_t>;
 		fun.window_init = OP::template WindowInit<STATE, INPUT_TYPE>;
 #endif
 		return fun;
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 bd8480ecc996..c59ebb642f67 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
@@ -8,10 +8,14 @@
 
 #pragma once
 
+#include "duckdb/common/sort/sort.hpp"
+#include "duckdb/common/types/column/column_data_collection.hpp"
+#include "duckdb/common/types/row/row_layout.hpp"
 #include "core_functions/aggregate/quantile_helpers.hpp"
 #include "duckdb/execution/merge_sort_tree.hpp"
 #include "duckdb/common/operator/cast_operators.hpp"
 #include "duckdb/common/operator/multiply.hpp"
+#include "duckdb/planner/expression/bound_constant_expression.hpp"
 #include <algorithm>
 #include <numeric>
 #include <stdlib.h>
@@ -19,6 +23,60 @@
 
 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);
+		inputs.InitializeScanChunk(scan, page);
+	}
+
+	inline sel_t RowOffset(idx_t row_idx) const {
+		D_ASSERT(RowIsVisible(row_idx));
+		return UnsafeNumericCast<sel_t>(row_idx - scan.current_row_index);
+	}
+
+	inline bool RowIsVisible(idx_t row_idx) const {
+		return (row_idx < scan.next_row_index && scan.current_row_index <= row_idx);
+	}
+
+	inline idx_t Seek(idx_t row_idx) {
+		if (!RowIsVisible(row_idx)) {
+			inputs.Seek(row_idx, scan, page);
+			data = FlatVector::GetData<INPUT_TYPE>(page.data[0]);
+			validity = &FlatVector::Validity(page.data[0]);
+		}
+		return RowOffset(row_idx);
+	}
+
+	inline const INPUT_TYPE &operator[](idx_t row_idx) {
+		const auto offset = Seek(row_idx);
+		return data[offset];
+	}
+
+	inline bool RowIsValid(idx_t row_idx) {
+		const auto offset = Seek(row_idx);
+		return validity->RowIsValid(offset);
+	}
+
+	inline bool AllValid() {
+		return all_valid;
+	}
+
+	//! Windowed paging
+	const ColumnDataCollection &inputs;
+	//! The state used for reading the collection on this thread
+	ColumnDataScanState scan;
+	//! The data chunk paged into into
+	DataChunk page;
+	//! The data pointer
+	const INPUT_TYPE *data = nullptr;
+	//! The validity mask
+	const ValidityMask *validity = nullptr;
+	//! Paged chunks do not track this but it is really necessary for performance
+	bool all_valid;
+};
+
 // Direct access
 template <typename T>
 struct QuantileDirect {
@@ -35,12 +93,13 @@ template <typename T>
 struct QuantileIndirect {
 	using INPUT_TYPE = idx_t;
 	using RESULT_TYPE = T;
-	const RESULT_TYPE *data;
+	using CURSOR = QuantileCursor<RESULT_TYPE>;
+	CURSOR &data;
 
-	explicit QuantileIndirect(const RESULT_TYPE *data_p) : data(data_p) {
+	explicit QuantileIndirect(CURSOR &data_p) : data(data_p) {
 	}
 
-	inline RESULT_TYPE operator()(const idx_t &input) const {
+	inline RESULT_TYPE operator()(const INPUT_TYPE &input) const {
 		return data[input];
 	}
 };
@@ -66,14 +125,23 @@ struct QuantileComposed {
 template <typename ACCESSOR>
 struct QuantileCompare {
 	using INPUT_TYPE = typename ACCESSOR::INPUT_TYPE;
-	const ACCESSOR &accessor;
+	const ACCESSOR &accessor_l;
+	const ACCESSOR &accessor_r;
 	const bool desc;
-	explicit QuantileCompare(const ACCESSOR &accessor_p, bool desc_p) : accessor(accessor_p), desc(desc_p) {
+
+	// Single cursor for linear operations
+	explicit QuantileCompare(const ACCESSOR &accessor, bool desc_p)
+	    : accessor_l(accessor), accessor_r(accessor), desc(desc_p) {
+	}
+
+	// Independent cursors for sorting
+	explicit QuantileCompare(const ACCESSOR &accessor_l, const ACCESSOR &accessor_r, bool desc_p)
+	    : accessor_l(accessor_l), accessor_r(accessor_r), desc(desc_p) {
 	}
 
 	inline bool operator()(const INPUT_TYPE &lhs, const INPUT_TYPE &rhs) const {
-		const auto lval = accessor(lhs);
-		const auto rval = accessor(rhs);
+		const auto lval = accessor_l(lhs);
+		const auto rval = accessor_r(rhs);
 
 		return desc ? (rval < lval) : (lval < rval);
 	}
@@ -143,12 +211,12 @@ struct Interpolator {
 	}
 
 	template <class INPUT_TYPE, class TARGET_TYPE>
-	inline TARGET_TYPE Extract(const INPUT_TYPE **dest, Vector &result) const {
+	inline TARGET_TYPE Extract(const INPUT_TYPE *dest, Vector &result) const {
 		if (CRN == FRN) {
-			return CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(*dest[0], result);
+			return CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(dest[0], result);
 		} else {
-			auto lo = CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(*dest[0], result);
-			auto hi = CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(*dest[1], result);
+			auto lo = CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(dest[0], result);
+			auto hi = CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(dest[1], result);
 			return CastInterpolation::Interpolate<TARGET_TYPE>(lo, RN - FRN, hi);
 		}
 	}
@@ -212,8 +280,8 @@ struct Interpolator<true> {
 	}
 
 	template <class INPUT_TYPE, class TARGET_TYPE>
-	TARGET_TYPE Extract(const INPUT_TYPE **dest, Vector &result) const {
-		return CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(*dest[0], result);
+	TARGET_TYPE Extract(const INPUT_TYPE *dest, Vector &result) const {
+		return CastInterpolation::Cast<INPUT_TYPE, TARGET_TYPE>(dest[0], result);
 	}
 
 	const bool desc;
@@ -224,23 +292,108 @@ struct Interpolator<true> {
 	idx_t end;
 };
 
+template <typename INPUT_TYPE>
 struct QuantileIncluded {
-	inline explicit QuantileIncluded(const ValidityMask &fmask_p, const ValidityMask &dmask_p)
+	using CURSOR_TYPE = QuantileCursor<INPUT_TYPE>;
+
+	inline explicit QuantileIncluded(const ValidityMask &fmask_p, CURSOR_TYPE &dmask_p)
 	    : fmask(fmask_p), dmask(dmask_p) {
 	}
 
-	inline bool operator()(const idx_t &idx) const {
+	inline bool operator()(const idx_t &idx) {
 		return fmask.RowIsValid(idx) && dmask.RowIsValid(idx);
 	}
 
-	inline bool AllValid() const {
+	inline bool AllValid() {
 		return fmask.AllValid() && dmask.AllValid();
 	}
 
 	const ValidityMask &fmask;
-	const ValidityMask &dmask;
+	CURSOR_TYPE &dmask;
 };
 
+// Shared untemplated sort logic
+static unique_ptr<GlobalSortState> SortQuantileIndices(const WindowPartitionInput &partition, // NOLINT
+                                                       const LogicalType &index_type, OrderType order_type) {
+	auto &inputs = *partition.inputs;
+	const auto &filter_mask = partition.filter_mask;
+
+	// Sort the unfiltered indices by the argument values
+	vector<LogicalType> payload_types;
+	payload_types.emplace_back(index_type);
+
+	idx_t capacity = STANDARD_VECTOR_SIZE;
+	DataChunk payload;
+	payload.Initialize(inputs.GetAllocator(), payload_types, capacity);
+	RowLayout payload_layout;
+	payload_layout.Initialize(payload.GetTypes());
+	SelectionVector filtered(capacity);
+
+	// TODO: Two pass parallel sorting using Build
+	auto order_expr = make_uniq<BoundConstantExpression>(Value(inputs.Types()[0]));
+	vector<BoundOrderByNode> orders;
+	orders.emplace_back(BoundOrderByNode(order_type, OrderByNullType::NULLS_LAST, std::move(order_expr)));
+
+	auto &buffer_manager = BufferManager::GetBufferManager(partition.context);
+	auto global_sort = make_uniq<GlobalSortState>(buffer_manager, orders, payload_layout);
+	global_sort->external = ClientConfig::GetConfig(partition.context).force_external;
+	const auto memory_per_thread = PhysicalOperator::GetMaxThreadMemory(partition.context);
+
+	LocalSortState local_sort;
+	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) {
+			payload.Destroy();
+			capacity = sort.size();
+			payload.Initialize(inputs.GetAllocator(), payload_types, capacity);
+			filtered.Initialize(capacity);
+		} else {
+			payload.Reset();
+		}
+		auto &indices = payload.data[0];
+		payload.SetCardinality(sort);
+		indices.Sequence(int64_t(state.current_row_index), 1, payload.size());
+
+		if (!filter_mask.AllValid() || !partition.all_valid) {
+			auto &key = sort.data[0];
+			auto &validity = FlatVector::Validity(key);
+			idx_t valid = 0;
+			for (sel_t i = 0; i < sort.size(); ++i) {
+				if (filter_mask.RowIsValid(i + state.current_row_index) && validity.RowIsValid(i)) {
+					filtered[valid++] = i;
+				}
+			}
+			if (valid < sort.size()) {
+				payload.Slice(filtered, valid);
+				sort.Slice(filtered, valid);
+			}
+		}
+		local_sort.SinkChunk(sort, payload);
+		if (local_sort.SizeInBytes() > memory_per_thread) {
+			local_sort.Sort(*global_sort, true);
+		}
+	}
+	global_sort->AddLocalState(local_sort);
+
+	//	Sort it
+	global_sort->PrepareMergePhase();
+	while (global_sort->sorted_blocks.size() > 1) {
+		global_sort->InitializeMergeRound();
+		MergeSorter merge_sorter(*global_sort, global_sort->buffer_manager);
+		merge_sorter.PerformInMergeRound();
+		global_sort->CompleteMergeRound(false);
+	}
+
+	return global_sort;
+}
+
 template <typename IDX>
 struct QuantileSortTree : public MergeSortTree<IDX, IDX> {
 
@@ -253,31 +406,48 @@ struct QuantileSortTree : public MergeSortTree<IDX, IDX> {
 	}
 
 	template <class INPUT_TYPE>
-	static unique_ptr<QuantileSortTree> WindowInit(const INPUT_TYPE *data, AggregateInputData &aggr_input_data,
-	                                               const ValidityMask &data_mask, const ValidityMask &filter_mask,
-	                                               idx_t count) {
-		//	Build the indirection array
+	static unique_ptr<QuantileSortTree> WindowInit(AggregateInputData &aggr_input_data,
+	                                               const WindowPartitionInput &partition) {
+		auto &inputs = *partition.inputs;
+
+		// Sort the unfiltered indices by the argument values
 		using ElementType = typename QuantileSortTree::ElementType;
-		vector<ElementType> sorted(count);
-		if (filter_mask.AllValid() && data_mask.AllValid()) {
-			std::iota(sorted.begin(), sorted.end(), 0);
-		} else {
-			size_t valid = 0;
-			QuantileIncluded included(filter_mask, data_mask);
-			for (ElementType i = 0; i < count; ++i) {
-				if (included(i)) {
-					sorted[valid++] = i;
-				}
-			}
-			sorted.resize(valid);
+		vector<LogicalType> payload_types;
+		switch (sizeof(ElementType)) {
+		case sizeof(int64_t):
+			payload_types.emplace_back(LogicalType::BIGINT);
+			break;
+		case sizeof(int32_t):
+			payload_types.emplace_back(LogicalType::INTEGER);
+			break;
+		default:
+			throw InternalException("Unsupported Quantile Sort Tree index size");
 		}
 
-		//	Sort it
+		// TODO: Two pass parallel sorting using Build
 		auto &bind_data = aggr_input_data.bind_data->Cast<QuantileBindData>();
-		using Accessor = QuantileIndirect<INPUT_TYPE>;
-		Accessor indirect(data);
-		QuantileCompare<Accessor> cmp(indirect, bind_data.desc);
-		std::sort(sorted.begin(), sorted.end(), cmp);
+		auto order_type = bind_data.desc ? OrderType::DESCENDING : OrderType::ASCENDING;
+		auto global_sort = SortQuantileIndices(partition, payload_types[0], order_type);
+
+		// Now scan the sorted indices into an array we can use as the leaves
+		vector<ElementType> sorted;
+		if (!global_sort->sorted_blocks.empty()) {
+			PayloadScanner scanner(*global_sort);
+			DataChunk payload;
+			payload.Initialize(inputs.GetAllocator(), payload_types);
+			sorted.resize(scanner.Remaining());
+			for (;;) {
+				idx_t row_idx = scanner.Scanned();
+				scanner.Scan(payload);
+				if (payload.size() == 0) {
+					break;
+				}
+				auto &indices = payload.data[0];
+				auto data = FlatVector::GetData<ElementType>(indices);
+
+				std::copy(data, data + payload.size(), sorted.data() + row_idx);
+			}
+		}
 
 		return make_uniq<QuantileSortTree>(std::move(sorted));
 	}
@@ -287,7 +457,7 @@ struct QuantileSortTree : public MergeSortTree<IDX, IDX> {
 	}
 
 	template <typename INPUT_TYPE, typename RESULT_TYPE, bool DISCRETE>
-	RESULT_TYPE WindowScalar(const INPUT_TYPE *data, const SubFrames &frames, const idx_t n, Vector &result,
+	RESULT_TYPE WindowScalar(QuantileCursor<INPUT_TYPE> &data, const SubFrames &frames, const idx_t n, Vector &result,
 	                         const QuantileValue &q) {
 		D_ASSERT(n > 0);
 
@@ -309,8 +479,8 @@ struct QuantileSortTree : public MergeSortTree<IDX, IDX> {
 	}
 
 	template <typename INPUT_TYPE, typename CHILD_TYPE, bool DISCRETE>
-	void WindowList(const INPUT_TYPE *data, const SubFrames &frames, const idx_t n, Vector &list, const idx_t lidx,
-	                const QuantileBindData &bind_data) {
+	void WindowList(QuantileCursor<INPUT_TYPE> &data, const SubFrames &frames, const idx_t n, Vector &list,
+	                const idx_t lidx, const QuantileBindData &bind_data) {
 		D_ASSERT(n > 0);
 
 		//	Thread safe and idempotent.
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 bba4fede6646..3b4c9304792e 100644
--- a/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp
+++ b/extension/core_functions/include/core_functions/aggregate/quantile_state.hpp
@@ -52,11 +52,9 @@ struct QuantileOperation {
 	template <class STATE, class INPUT_TYPE>
 	static void WindowInit(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
 	                       data_ptr_t g_state) {
-		D_ASSERT(partition.input_count == 1);
+		D_ASSERT(partition.inputs);
 
-		auto inputs = partition.inputs;
 		const auto count = partition.count;
-		const auto &filter_mask = partition.filter_mask;
 		const auto &stats = partition.stats;
 
 		//	If frames overlap significantly, then use local skip lists.
@@ -70,22 +68,18 @@ struct QuantileOperation {
 			}
 		}
 
-		const auto data = FlatVector::GetData<const INPUT_TYPE>(inputs[0]);
-		const auto &data_mask = FlatVector::Validity(inputs[0]);
-
 		//	Build the tree
 		auto &state = *reinterpret_cast<STATE *>(g_state);
 		auto &window_state = state.GetOrCreateWindowState();
 		if (count < std::numeric_limits<uint32_t>::max()) {
-			window_state.qst32 = QuantileSortTree<uint32_t>::WindowInit<INPUT_TYPE>(data, aggr_input_data, data_mask,
-			                                                                        filter_mask, count);
+			window_state.qst32 = QuantileSortTree<uint32_t>::WindowInit<INPUT_TYPE>(aggr_input_data, partition);
 		} else {
-			window_state.qst64 = QuantileSortTree<uint64_t>::WindowInit<INPUT_TYPE>(data, aggr_input_data, data_mask,
-			                                                                        filter_mask, count);
+			window_state.qst64 = QuantileSortTree<uint64_t>::WindowInit<INPUT_TYPE>(aggr_input_data, partition);
 		}
 	}
 
-	static idx_t FrameSize(const QuantileIncluded &included, const SubFrames &frames) {
+	template <class INPUT_TYPE>
+	static idx_t FrameSize(QuantileIncluded<INPUT_TYPE> &included, const SubFrames &frames) {
 		//	Count the number of valid values
 		idx_t n = 0;
 		if (included.AllValid()) {
@@ -106,9 +100,9 @@ struct QuantileOperation {
 };
 
 template <class T>
-struct PointerLess {
+struct SkipLess {
 	inline bool operator()(const T &lhi, const T &rhi) const {
-		return *lhi < *rhi;
+		return lhi.second < rhi.second;
 	}
 };
 
@@ -121,16 +115,19 @@ struct WindowQuantileState {
 	unique_ptr<QuantileSortTree64> qst64;
 
 	// Windowed Quantile skip lists
-	using PointerType = const INPUT_TYPE *;
-	using SkipListType = duckdb_skiplistlib::skip_list::HeadNode<PointerType, PointerLess<PointerType>>;
+	using SkipType = pair<idx_t, INPUT_TYPE>;
+	using SkipListType = duckdb_skiplistlib::skip_list::HeadNode<SkipType, SkipLess<SkipType>>;
 	SubFrames prevs;
 	unique_ptr<SkipListType> s;
-	mutable vector<PointerType> dest;
+	mutable vector<SkipType> skips;
 
 	// Windowed MAD indirection
 	idx_t count;
 	vector<idx_t> m;
 
+	using IncludedType = QuantileIncluded<INPUT_TYPE>;
+	using CursorType = QuantileCursor<INPUT_TYPE>;
+
 	WindowQuantileState() : count(0) {
 	}
 
@@ -151,10 +148,10 @@ struct WindowQuantileState {
 
 	struct SkipListUpdater {
 		SkipListType &skip;
-		const INPUT_TYPE *data;
-		const QuantileIncluded &included;
+		CursorType &data;
+		IncludedType &included;
 
-		inline SkipListUpdater(SkipListType &skip, const INPUT_TYPE *data, const QuantileIncluded &included)
+		inline SkipListUpdater(SkipListType &skip, CursorType &data, IncludedType &included)
 		    : skip(skip), data(data), included(included) {
 		}
 
@@ -164,7 +161,7 @@ struct WindowQuantileState {
 		inline void Left(idx_t begin, idx_t end) {
 			for (; begin < end; ++begin) {
 				if (included(begin)) {
-					skip.remove(data + begin);
+					skip.remove(SkipType(begin, data[begin]));
 				}
 			}
 		}
@@ -172,7 +169,7 @@ struct WindowQuantileState {
 		inline void Right(idx_t begin, idx_t end) {
 			for (; begin < end; ++begin) {
 				if (included(begin)) {
-					skip.insert(data + begin);
+					skip.insert(SkipType(begin, data[begin]));
 				}
 			}
 		}
@@ -181,14 +178,14 @@ struct WindowQuantileState {
 		}
 	};
 
-	void UpdateSkip(const INPUT_TYPE *data, const SubFrames &frames, const QuantileIncluded &included) {
+	void UpdateSkip(CursorType &data, const SubFrames &frames, IncludedType &included) {
 		//	No overlap, or no data
 		if (!s || prevs.back().end <= frames.front().start || frames.back().end <= prevs.front().start) {
 			auto &skip = GetSkipList(true);
 			for (const auto &frame : frames) {
 				for (auto i = frame.start; i < frame.end; ++i) {
 					if (included(i)) {
-						skip.insert(data + i);
+						skip.insert(SkipType(i, data[i]));
 					}
 				}
 			}
@@ -204,7 +201,7 @@ struct WindowQuantileState {
 	}
 
 	template <typename RESULT_TYPE, bool DISCRETE>
-	RESULT_TYPE WindowScalar(const INPUT_TYPE *data, const SubFrames &frames, const idx_t n, Vector &result,
+	RESULT_TYPE WindowScalar(CursorType &data, const SubFrames &frames, const idx_t n, Vector &result,
 	                         const QuantileValue &q) const {
 		D_ASSERT(n > 0);
 		if (qst32) {
@@ -215,7 +212,12 @@ struct WindowQuantileState {
 			// Find the position(s) needed
 			try {
 				Interpolator<DISCRETE> interp(q, s->size(), false);
-				s->at(interp.FRN, interp.CRN - interp.FRN + 1, dest);
+				s->at(interp.FRN, interp.CRN - interp.FRN + 1, skips);
+				array<INPUT_TYPE, 2> dest;
+				dest[0] = skips[0].second;
+				if (skips.size() > 1) {
+					dest[1] = skips[1].second;
+				}
 				return interp.template Extract<INPUT_TYPE, RESULT_TYPE>(dest.data(), result);
 			} catch (const duckdb_skiplistlib::skip_list::IndexError &idx_err) {
 				throw InternalException(idx_err.message());
@@ -226,7 +228,7 @@ struct WindowQuantileState {
 	}
 
 	template <typename CHILD_TYPE, bool DISCRETE>
-	void WindowList(const INPUT_TYPE *data, const SubFrames &frames, const idx_t n, Vector &list, const idx_t lidx,
+	void WindowList(CursorType &data, const SubFrames &frames, const idx_t n, Vector &list, const idx_t lidx,
 	                const QuantileBindData &bind_data) const {
 		D_ASSERT(n > 0);
 		// Result is a constant LIST<CHILD_TYPE> with a fixed length
@@ -269,12 +271,14 @@ struct QuantileStringType {
 template <typename INPUT_TYPE, class TYPE_OP>
 struct QuantileState {
 	using InputType = INPUT_TYPE;
+	using CursorType = QuantileCursor<INPUT_TYPE>;
 
 	// Regular aggregation
 	vector<INPUT_TYPE> v;
 
 	// Window Quantile State
 	unique_ptr<WindowQuantileState<INPUT_TYPE>> window_state;
+	unique_ptr<CursorType> window_cursor;
 
 	void AddElement(INPUT_TYPE element, AggregateInputData &aggr_input) {
 		v.emplace_back(TYPE_OP::Operation(element, aggr_input));
@@ -295,6 +299,19 @@ struct QuantileState {
 	const WindowQuantileState<INPUT_TYPE> &GetWindowState() const {
 		return *window_state;
 	}
+
+	CursorType &GetOrCreateWindowCursor(const ColumnDataCollection &inputs, bool all_valid) {
+		if (!window_state) {
+			window_cursor = make_uniq<CursorType>(inputs, all_valid);
+		}
+		return *window_cursor;
+	}
+	CursorType &GetWindowCursor() {
+		return *window_cursor;
+	}
+	const CursorType &GetWindowCursor() const {
+		return *window_cursor;
+	}
 };
 
 } // namespace duckdb
diff --git a/src/common/types/column/column_data_collection.cpp b/src/common/types/column/column_data_collection.cpp
index c3740444aa25..41626d470633 100644
--- a/src/common/types/column/column_data_collection.cpp
+++ b/src/common/types/column/column_data_collection.cpp
@@ -923,6 +923,29 @@ bool ColumnDataCollection::NextScanIndex(ColumnDataScanState &state, idx_t &chun
 	return true;
 }
 
+bool ColumnDataCollection::PrevScanIndex(ColumnDataScanState &state, idx_t &chunk_index, idx_t &segment_index,
+                                         idx_t &row_index) const {
+	// check within the current segment if we still have chunks to scan
+	// Note that state.chunk_index is 1-indexed, with 0 as undefined.
+	while (state.chunk_index <= 1) {
+		if (!state.segment_index) {
+			return false;
+		}
+
+		--state.segment_index;
+		state.chunk_index = segments[state.segment_index]->chunk_data.size() + 1;
+		state.current_chunk_state.handles.clear();
+	}
+
+	--state.chunk_index;
+	segment_index = state.segment_index;
+	chunk_index = state.chunk_index - 1;
+	state.next_row_index = state.current_row_index;
+	state.current_row_index -= segments[state.segment_index]->chunk_data[chunk_index].count;
+	row_index = state.current_row_index;
+	return true;
+}
+
 void ColumnDataCollection::ScanAtIndex(ColumnDataParallelScanState &state, ColumnDataLocalScanState &lstate,
                                        DataChunk &result, idx_t chunk_index, idx_t segment_index,
                                        idx_t row_index) const {
@@ -955,6 +978,38 @@ bool ColumnDataCollection::Scan(ColumnDataScanState &state, DataChunk &result) c
 	return true;
 }
 
+bool ColumnDataCollection::Seek(idx_t seek_idx, ColumnDataScanState &state, DataChunk &result) const {
+	//	Idempotency: Don't change anything if the row is already in range
+	if (state.current_row_index <= seek_idx && seek_idx < state.next_row_index) {
+		return true;
+	}
+
+	result.Reset();
+
+	//	Linear scan for now. We could use a current_row_index => chunk map at some point
+	//	but most use cases should be pretty local
+	idx_t chunk_index;
+	idx_t segment_index;
+	idx_t row_index;
+	while (seek_idx < state.current_row_index) {
+		if (!PrevScanIndex(state, chunk_index, segment_index, row_index)) {
+			return false;
+		}
+	}
+	while (state.next_row_index <= seek_idx) {
+		if (!NextScanIndex(state, chunk_index, segment_index, row_index)) {
+			return false;
+		}
+	}
+
+	// found a chunk to scan -> scan it
+	auto &segment = *segments[segment_index];
+	state.current_chunk_state.properties = state.properties;
+	segment.ReadChunk(chunk_index, state.current_chunk_state, result, state.column_ids);
+	result.Verify();
+	return true;
+}
+
 ColumnDataRowCollection ColumnDataCollection::GetRows() const {
 	return ColumnDataRowCollection(*this);
 }
diff --git a/src/execution/window_executor.cpp b/src/execution/window_executor.cpp
index 56397f7a7a5b..0b7a0a48cf2b 100644
--- a/src/execution/window_executor.cpp
+++ b/src/execution/window_executor.cpp
@@ -8,74 +8,98 @@
 namespace duckdb {
 
 //===--------------------------------------------------------------------===//
-// WindowDataChunk
+// WindowCollection
 //===--------------------------------------------------------------------===//
-bool WindowDataChunk::IsSimple(const Vector &v) {
-	switch (v.GetType().InternalType()) {
-	case PhysicalType::BOOL:
-	case PhysicalType::UINT8:
-	case PhysicalType::INT8:
-	case PhysicalType::UINT16:
-	case PhysicalType::INT16:
-	case PhysicalType::UINT32:
-	case PhysicalType::INT32:
-	case PhysicalType::UINT64:
-	case PhysicalType::INT64:
-	case PhysicalType::FLOAT:
-	case PhysicalType::DOUBLE:
-	case PhysicalType::INTERVAL:
-	case PhysicalType::UINT128:
-	case PhysicalType::INT128:
-		return true;
-	case PhysicalType::LIST:
-	case PhysicalType::STRUCT:
-	case PhysicalType::ARRAY:
-	case PhysicalType::VARCHAR:
-	case PhysicalType::BIT:
-		return false;
-	default:
-		break;
+WindowCollection::WindowCollection(BufferManager &buffer_manager, idx_t count, const vector<LogicalType> &types)
+    : all_valid(true), types(types), count(count), buffer_manager(buffer_manager) {
+	if (!types.empty()) {
+		inputs = make_uniq<ColumnDataCollection>(buffer_manager, types);
 	}
-
-	throw InternalException("Unsupported type for WindowDataChunk");
 }
 
-WindowDataChunk::WindowDataChunk(DataChunk &chunk) : chunk(chunk) {
+void WindowCollection::GetCollection(idx_t row_idx, ColumnDataCollectionSpec &spec) {
+	if (spec.second && row_idx == spec.first + spec.second->Count()) {
+		return;
+	}
+
+	lock_guard<mutex> collection_guard(lock);
+
+	auto collection = make_uniq<ColumnDataCollection>(buffer_manager, types);
+	spec = {row_idx, collection.get()};
+	Range probe {row_idx, collections.size()};
+	auto i = std::upper_bound(ranges.begin(), ranges.end(), probe);
+	ranges.insert(i, probe);
+	collections.emplace_back(std::move(collection));
 }
 
-void WindowDataChunk::Initialize(Allocator &allocator, const vector<LogicalType> &types, idx_t capacity) {
-	vector<mutex> new_locks(types.size());
-	locks.swap(new_locks);
-	chunk.Initialize(allocator, types, capacity);
-	chunk.SetCardinality(capacity);
+void WindowCollection::Combine(bool build_validity) {
+	lock_guard<mutex> collection_guard(lock);
+
+	if (collections.empty()) {
+		return;
+	}
 
-	is_simple.clear();
-	for (const auto &v : chunk.data) {
-		is_simple.push_back(IsSimple(v));
+	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;
+		}
 	}
 }
 
-void WindowDataChunk::Copy(DataChunk &input, idx_t begin) {
-	const auto source_count = input.size();
-	const idx_t end = begin + source_count;
-	const idx_t count = chunk.size();
-	D_ASSERT(end <= count);
-	// Can we overwrite the validity mask in parallel?
-	bool aligned = IsMaskAligned(begin, end, count);
-	for (column_t i = 0; i < chunk.data.size(); ++i) {
-		auto &src = input.data[i];
-		auto &dst = chunk.data[i];
-		UnifiedVectorFormat sdata;
-		src.ToUnifiedFormat(count, sdata);
-		if (is_simple[i] && aligned && sdata.validity.AllValid()) {
-			VectorOperations::Copy(src, dst, source_count, 0, begin);
-		} else {
-			lock_guard<mutex> column_guard(locks[i]);
-			VectorOperations::Copy(src, dst, source_count, 0, begin);
+WindowBuilder::WindowBuilder(WindowCollection &collection) : collection(collection) {
+}
+
+void WindowBuilder::Sink(DataChunk &chunk, idx_t input_idx) {
+	// Check whether we need a a new collection
+	if (!sink.second || input_idx < sink.first || sink.first + sink.second->Count() < input_idx) {
+		collection.GetCollection(input_idx, sink);
+		D_ASSERT(sink.second);
+		sink.second->InitializeAppend(appender);
+	}
+	sink.second->Append(appender, chunk);
+
+	// Record NULLs
+	if (all_valid) {
+		UnifiedVectorFormat data;
+		chunk.data[0].ToUnifiedFormat(chunk.size(), data);
+		all_valid = data.validity.AllValid();
+		if (!all_valid) {
+			collection.all_valid = false;
 		}
 	}
 }
 
+WindowCursor::WindowCursor(const WindowCollection &paged) : paged(paged) {
+	if (paged.GetTypes().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.SetCardinality(state.next_row_index);
+		return;
+	} else if (chunk.data.empty()) {
+		auto &inputs = paged.inputs;
+		D_ASSERT(inputs.get());
+		inputs->InitializeScan(state);
+		inputs->InitializeScanChunk(state, chunk);
+	}
+}
+
 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);
@@ -142,46 +166,6 @@ static idx_t FindPrevStart(const ValidityMask &mask, const idx_t l, idx_t r, idx
 	return l;
 }
 
-template <typename T>
-static T GetCell(const DataChunk &chunk, idx_t column, idx_t index) {
-	D_ASSERT(chunk.ColumnCount() > column);
-	auto &source = chunk.data[column];
-	const auto data = FlatVector::GetData<T>(source);
-	return data[index];
-}
-
-static bool CellIsNull(const DataChunk &chunk, idx_t column, idx_t index) {
-	D_ASSERT(chunk.ColumnCount() > column);
-	auto &source = chunk.data[column];
-	return FlatVector::IsNull(source, index);
-}
-
-static void CopyCell(const DataChunk &chunk, idx_t column, idx_t index, Vector &target, idx_t target_offset) {
-	D_ASSERT(chunk.ColumnCount() > column);
-	auto &source = chunk.data[column];
-	VectorOperations::Copy(source, target, index + 1, index, target_offset);
-}
-
-//===--------------------------------------------------------------------===//
-// WindowInputColumn
-//===--------------------------------------------------------------------===//
-WindowInputColumn::WindowInputColumn(optional_ptr<Expression> expr_p, ClientContext &context, idx_t count)
-    : expr(expr_p), scalar(expr ? expr->IsScalar() : true), count(count), wtarget(target) {
-
-	if (expr) {
-		vector<LogicalType> types;
-		types.emplace_back(expr->return_type);
-		wtarget.Initialize(Allocator::Get(context), types, count);
-		ptype = expr->return_type.InternalType();
-	}
-}
-
-void WindowInputColumn::Copy(DataChunk &input_chunk, idx_t input_idx) {
-	if (expr && (!input_idx || !scalar)) {
-		wtarget.Copy(input_chunk, input_idx);
-	}
-}
-
 //===--------------------------------------------------------------------===//
 // WindowColumnIterator
 //===--------------------------------------------------------------------===//
@@ -194,12 +178,12 @@ struct WindowColumnIterator {
 	using reference = T;
 	using pointer = idx_t;
 
-	explicit WindowColumnIterator(const WindowInputColumn &coll_p, pointer pos_p = 0) : coll(&coll_p), pos(pos_p) {
+	explicit WindowColumnIterator(WindowCursor &coll, pointer pos = 0) : coll(&coll), pos(pos) {
 	}
 
 	//	Forward iterator
 	inline reference operator*() const {
-		return coll->GetCell<T>(pos);
+		return coll->GetCell<T>(0, pos);
 	}
 	inline explicit operator pointer() const {
 		return pos;
@@ -237,7 +221,7 @@ struct WindowColumnIterator {
 	}
 
 	inline reference operator[](difference_type m) const {
-		return coll->GetCell<T>(pos + m);
+		return coll->GetCell<T>(0, pos + m);
 	}
 
 	friend inline iterator &operator+(const iterator &a, difference_type n) {
@@ -275,7 +259,8 @@ struct WindowColumnIterator {
 	}
 
 private:
-	optional_ptr<const WindowInputColumn> coll;
+	// optional_ptr does not allow us to modify this, but the constructor enforces it.
+	WindowCursor *coll;
 	pointer pos;
 };
 
@@ -287,7 +272,7 @@ struct OperationCompare : public std::function<bool(T, T)> {
 };
 
 template <typename T, typename OP, bool FROM>
-static idx_t FindTypedRangeBound(const WindowInputColumn &over, const idx_t order_begin, const idx_t order_end,
+static idx_t FindTypedRangeBound(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.CellIsNull(chunk_idx));
@@ -298,14 +283,14 @@ static idx_t FindTypedRangeBound(const WindowInputColumn &over, const idx_t orde
 	// Check that the value we are searching for is in range.
 	if (range == WindowBoundary::EXPR_PRECEDING_RANGE) {
 		//	Preceding but value past the current value
-		const auto cur_val = over.GetCell<T>(order_end - 1);
+		const auto cur_val = over.GetCell<T>(0, order_end - 1);
 		if (comp(cur_val, val)) {
 			throw OutOfRangeException("Invalid RANGE PRECEDING value");
 		}
 	} else {
 		//	Following but value before the current value
 		D_ASSERT(range == WindowBoundary::EXPR_FOLLOWING_RANGE);
-		const auto cur_val = over.GetCell<T>(order_begin);
+		const auto cur_val = over.GetCell<T>(0, order_begin);
 		if (comp(val, cur_val)) {
 			throw OutOfRangeException("Invalid RANGE FOLLOWING value");
 		}
@@ -318,14 +303,14 @@ static idx_t FindTypedRangeBound(const WindowInputColumn &over, const idx_t orde
 	WindowColumnIterator<T> end(over, order_end);
 	if (prev.start < prev.end) {
 		if (order_begin < prev.start && prev.start < order_end) {
-			const auto first = over.GetCell<T>(prev.start);
+			const auto first = over.GetCell<T>(0, prev.start);
 			if (!comp(val, first)) {
 				//	prev.first <= val, so we can start further forward
 				begin += UnsafeNumericCast<int64_t>(prev.start - order_begin);
 			}
 		}
 		if (order_begin < prev.end && prev.end < order_end) {
-			const auto second = over.GetCell<T>(prev.end - 1);
+			const auto second = over.GetCell<T>(0, prev.end - 1);
 			if (!comp(second, val)) {
 				//	val <= prev.second, so we can end further back
 				// (prev.second is the largest peer)
@@ -342,13 +327,12 @@ static idx_t FindTypedRangeBound(const WindowInputColumn &over, const idx_t orde
 }
 
 template <typename OP, bool FROM>
-static idx_t FindRangeBound(const WindowInputColumn &over, const idx_t order_begin, const idx_t order_end,
+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);
-	D_ASSERT(boundary.chunk.data[0].GetType().InternalType() == over.ptype);
 
-	switch (over.ptype) {
+	switch (boundary.chunk.data[0].GetType().InternalType()) {
 	case PhysicalType::INT8:
 		return FindTypedRangeBound<int8_t, OP, FROM>(over, order_begin, order_end, range, boundary, chunk_idx, prev);
 	case PhysicalType::INT16:
@@ -383,7 +367,7 @@ static idx_t FindRangeBound(const WindowInputColumn &over, const idx_t order_beg
 }
 
 template <bool FROM>
-static idx_t FindOrderedRangeBound(const WindowInputColumn &over, const OrderType range_sense, const idx_t order_begin,
+static idx_t FindOrderedRangeBound(WindowCursor &over, const OrderType range_sense, const idx_t order_begin,
                                    const idx_t order_end, const WindowBoundary range, WindowInputExpression &boundary,
                                    const idx_t chunk_idx, const FrameBounds &prev) {
 	switch (range_sense) {
@@ -426,11 +410,11 @@ struct WindowBoundariesState {
 
 	WindowBoundariesState(const BoundWindowExpression &wexpr, const idx_t input_size);
 
-	void Update(const idx_t row_idx, const WindowInputColumn &range_collection, const idx_t chunk_idx,
+	void Update(const idx_t row_idx, optional_ptr<WindowCursor> range_collection, const idx_t chunk_idx,
 	            WindowInputExpression &boundary_start, WindowInputExpression &boundary_end,
 	            const ValidityMask &partition_mask, const ValidityMask &order_mask);
 
-	void Bounds(DataChunk &bounds, idx_t row_idx, const WindowInputColumn &range, const idx_t count,
+	void Bounds(DataChunk &bounds, idx_t row_idx, optional_ptr<WindowCursor> range, const idx_t count,
 	            WindowInputExpression &boundary_start, WindowInputExpression &boundary_end,
 	            const ValidityMask &partition_mask, const ValidityMask &order_mask);
 
@@ -461,10 +445,9 @@ struct WindowBoundariesState {
 //===--------------------------------------------------------------------===//
 // WindowBoundariesState
 //===--------------------------------------------------------------------===//
-void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn &range_collection,
-                                   const idx_t chunk_idx, WindowInputExpression &boundary_start,
-                                   WindowInputExpression &boundary_end, const ValidityMask &partition_mask,
-                                   const ValidityMask &order_mask) {
+void WindowBoundariesState::Update(const idx_t row_idx, optional_ptr<WindowCursor> range, const idx_t chunk_idx,
+                                   WindowInputExpression &boundary_start, WindowInputExpression &boundary_end,
+                                   const ValidityMask &partition_mask, const ValidityMask &order_mask) {
 
 	if (partition_count + order_count > 0) {
 
@@ -499,7 +482,7 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 
 			if ((valid_start < valid_end) && has_preceding_range) {
 				// Exclude any leading NULLs
-				if (range_collection.CellIsNull(valid_start)) {
+				if (range->CellIsNull(0, valid_start)) {
 					idx_t n = 1;
 					valid_start = FindNextStart(order_mask, valid_start + 1, valid_end, n);
 				}
@@ -507,7 +490,7 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 
 			if ((valid_start < valid_end) && has_following_range) {
 				// Exclude any trailing NULLs
-				if (range_collection.CellIsNull(valid_end - 1)) {
+				if (range->CellIsNull(0, valid_end - 1)) {
 					idx_t n = 1;
 					valid_end = FindPrevStart(order_mask, valid_start, valid_end, n);
 				}
@@ -570,8 +553,8 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 		if (boundary_start.CellIsNull(chunk_idx)) {
 			window_start = peer_start;
 		} else {
-			prev.start = FindOrderedRangeBound<true>(range_collection, range_sense, valid_start, row_idx + 1,
-			                                         start_boundary, boundary_start, chunk_idx, prev);
+			prev.start = FindOrderedRangeBound<true>(*range, range_sense, valid_start, row_idx + 1, start_boundary,
+			                                         boundary_start, chunk_idx, prev);
 			window_start = prev.start;
 		}
 		break;
@@ -580,7 +563,7 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 		if (boundary_start.CellIsNull(chunk_idx)) {
 			window_start = peer_start;
 		} else {
-			prev.start = FindOrderedRangeBound<true>(range_collection, range_sense, row_idx, valid_end, start_boundary,
+			prev.start = FindOrderedRangeBound<true>(*range, range_sense, row_idx, valid_end, start_boundary,
 			                                         boundary_start, chunk_idx, prev);
 			window_start = prev.start;
 		}
@@ -624,8 +607,8 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 		if (boundary_end.CellIsNull(chunk_idx)) {
 			window_end = peer_end;
 		} else {
-			prev.end = FindOrderedRangeBound<false>(range_collection, range_sense, valid_start, row_idx + 1,
-			                                        end_boundary, boundary_end, chunk_idx, prev);
+			prev.end = FindOrderedRangeBound<false>(*range, range_sense, valid_start, row_idx + 1, end_boundary,
+			                                        boundary_end, chunk_idx, prev);
 			window_end = prev.end;
 		}
 		break;
@@ -634,8 +617,8 @@ void WindowBoundariesState::Update(const idx_t row_idx, const WindowInputColumn
 		if (boundary_end.CellIsNull(chunk_idx)) {
 			window_end = peer_end;
 		} else {
-			prev.end = FindOrderedRangeBound<false>(range_collection, range_sense, row_idx, valid_end, end_boundary,
-			                                        boundary_end, chunk_idx, prev);
+			prev.end = FindOrderedRangeBound<false>(*range, range_sense, row_idx, valid_end, end_boundary, boundary_end,
+			                                        chunk_idx, prev);
 			window_end = prev.end;
 		}
 		break;
@@ -677,9 +660,10 @@ WindowBoundariesState::WindowBoundariesState(const BoundWindowExpression &wexpr,
                  wexpr.exclude_clause >= WindowExcludeMode::GROUP) {
 }
 
-void WindowBoundariesState::Bounds(DataChunk &bounds, idx_t row_idx, const WindowInputColumn &range, const idx_t count,
-                                   WindowInputExpression &boundary_start, WindowInputExpression &boundary_end,
-                                   const ValidityMask &partition_mask, const ValidityMask &order_mask) {
+void WindowBoundariesState::Bounds(DataChunk &bounds, idx_t row_idx, optional_ptr<WindowCursor> range,
+                                   const idx_t count, WindowInputExpression &boundary_start,
+                                   WindowInputExpression &boundary_end, const ValidityMask &partition_mask,
+                                   const ValidityMask &order_mask) {
 	bounds.Reset();
 	D_ASSERT(bounds.ColumnCount() == 6);
 	auto partition_begin_data = FlatVector::GetData<idx_t>(bounds.data[PARTITION_BEGIN]);
@@ -711,7 +695,7 @@ class WindowExecutorBoundsState : public WindowExecutorLocalState {
 	~WindowExecutorBoundsState() override {
 	}
 
-	virtual void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, const WindowInputColumn &range);
+	virtual void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range);
 
 	// Frame management
 	const ValidityMask &partition_mask;
@@ -733,7 +717,7 @@ WindowExecutorBoundsState::WindowExecutorBoundsState(const WindowExecutorGlobalS
 	bounds.Initialize(Allocator::Get(gstate.executor.context), bounds_types);
 }
 
-void WindowExecutorBoundsState::UpdateBounds(idx_t row_idx, DataChunk &input_chunk, const WindowInputColumn &range) {
+void WindowExecutorBoundsState::UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range) {
 	// Evaluate the row-level arguments
 	boundary_start.Execute(input_chunk);
 	boundary_end.Execute(input_chunk);
@@ -778,8 +762,6 @@ class ExclusionFilter {
 	ValidityMask mask;
 	//! The validity mask upon which mask is based
 	const ValidityMask &mask_src;
-	//! A validity mask consisting of only one entries (needed if no ignore_nulls mask is supplied)
-	ValidityMask all_ones_mask;
 };
 
 void ExclusionFilter::FetchFromSource(idx_t begin, idx_t end) {
@@ -869,13 +851,18 @@ WindowExecutor::WindowExecutor(BoundWindowExpression &wexpr, ClientContext &cont
 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((HasPrecedingRange(executor.wexpr) || HasFollowingRange(executor.wexpr))
-                ? executor.wexpr.orders[0].expression.get()
-                : nullptr,
-            executor.context, payload_count) {
+      range_expr((HasPrecedingRange(executor.wexpr) || HasFollowingRange(executor.wexpr))
+                     ? executor.wexpr.orders[0].expression.get()
+                     : nullptr) {
 	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)
@@ -885,20 +872,28 @@ WindowExecutorLocalState::WindowExecutorLocalState(const WindowExecutorGlobalSta
 	// evaluate inner expressions of window functions, could be more complex
 	PrepareInputExpressions(gstate.executor.wexpr.children, payload_executor, payload_chunk);
 
-	if (gstate.range.expr) {
-		vector<LogicalType> types;
-		types.emplace_back(gstate.range.expr->return_type);
-		range_executor.AddExpression(*gstate.range.expr);
+	if (gstate.range_expr) {
+		range_executor.AddExpression(*gstate.range_expr);
 
+		D_ASSERT(gstate.range.get());
 		auto &allocator = range_executor.GetAllocator();
-		range_chunk.Initialize(allocator, types);
+		range_chunk.Initialize(allocator, gstate.range->GetTypes());
+		range_builder = make_uniq<WindowBuilder>(*gstate.range);
 	}
 }
 
 void WindowExecutorLocalState::Sink(WindowExecutorGlobalState &gstate, DataChunk &input_chunk, idx_t input_idx) {
-	if (gstate.range.expr && (!input_idx || !gstate.range.scalar)) {
+	if (gstate.range_expr) {
 		range_executor.Execute(input_chunk, range_chunk);
-		gstate.range.Copy(range_chunk, input_idx);
+		range_builder->Sink(range_chunk, input_idx);
+	}
+}
+
+void WindowExecutorLocalState::Finalize(WindowExecutorGlobalState &gstate) {
+	if (gstate.range) {
+		if (!range_cursor) {
+			range_cursor = make_uniq<WindowCursor>(*gstate.range);
+		}
 	}
 }
 
@@ -917,6 +912,14 @@ void WindowExecutor::Sink(DataChunk &input_chunk, const idx_t input_idx, const i
 	lstate.Sink(gstate, input_chunk, input_idx);
 }
 
+void WindowExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
+	if (gstate.range) {
+		gstate.range->Combine(false);
+	}
+
+	lstate.Finalize(gstate);
+}
+
 //===--------------------------------------------------------------------===//
 // WindowAggregateExecutor
 //===--------------------------------------------------------------------===//
@@ -1023,7 +1026,7 @@ bool WindowAggregateExecutorGlobalState::IsCustomAggregate() {
 void WindowExecutor::Evaluate(idx_t row_idx, DataChunk &input_chunk, Vector &result, WindowExecutorLocalState &lstate,
                               WindowExecutorGlobalState &gstate) const {
 	auto &lbstate = lstate.Cast<WindowExecutorBoundsState>();
-	lbstate.UpdateBounds(row_idx, input_chunk, gstate.range);
+	lbstate.UpdateBounds(row_idx, input_chunk, lstate.range_cursor);
 
 	const auto count = input_chunk.size();
 	EvaluateInternal(gstate, lstate, result, count, row_idx);
@@ -1066,7 +1069,7 @@ WindowAggregateExecutorGlobalState::WindowAggregateExecutorGlobalState(const Win
 		aggregator = make_uniq<WindowSegmentTree>(aggr, arg_types, return_type, mode, wexpr.exclude_clause);
 	}
 
-	gsink = aggregator->GetGlobalState(group_count, partition_mask);
+	gsink = aggregator->GetGlobalState(context, group_count, partition_mask);
 }
 
 unique_ptr<WindowExecutorGlobalState> WindowAggregateExecutor::GetGlobalState(const idx_t payload_count,
@@ -1200,6 +1203,8 @@ static void ApplyWindowStats(const WindowBoundary &boundary, FrameDelta &delta,
 }
 
 void WindowAggregateExecutor::Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
+	WindowExecutor::Finalize(gstate, lstate);
+
 	auto &gastate = gstate.Cast<WindowAggregateExecutorGlobalState>();
 	auto &aggregator = gastate.aggregator;
 	auto &gsink = gastate.gsink;
@@ -1427,38 +1432,21 @@ void WindowCumeDistExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 
 class WindowValueGlobalState : public WindowExecutorGlobalState {
 public:
+	using WindowDataChunkPtr = unique_ptr<WindowCollection>;
 	WindowValueGlobalState(const WindowExecutor &executor, const idx_t payload_count,
 	                       const ValidityMask &partition_mask, const ValidityMask &order_mask)
-	    : WindowExecutorGlobalState(executor, payload_count, partition_mask, order_mask),
-	      payload_collection(payload_data), ignore_nulls(&no_nulls)
+	    : WindowExecutorGlobalState(executor, payload_count, partition_mask, order_mask)
 
 	{
 		if (!arg_types.empty()) {
-			payload_collection.Initialize(Allocator::Get(executor.context), arg_types, payload_count);
-		}
-
-		auto &wexpr = executor.wexpr;
-		if (wexpr.ignore_nulls) {
-			switch (wexpr.type) {
-			case ExpressionType::WINDOW_LEAD:
-			case ExpressionType::WINDOW_LAG:
-			case ExpressionType::WINDOW_FIRST_VALUE:
-			case ExpressionType::WINDOW_LAST_VALUE:
-			case ExpressionType::WINDOW_NTH_VALUE:
-				ignore_nulls = &FlatVector::Validity(payload_collection.chunk.data[0]);
-				break;
-			default:
-				break;
-			}
+			auto &buffer_manager = BufferManager::GetBufferManager(executor.context);
+			payload_collection = make_uniq<WindowCollection>(buffer_manager, payload_count, arg_types);
+			ignore_nulls = &payload_collection->validity;
 		}
 	}
 
 	// The partition values
-	DataChunk payload_data;
-	// The partition values
-	WindowDataChunk payload_collection;
-	// Mask to use for exclusion if we are not ignoring NULLs
-	ValidityMask no_nulls;
+	WindowDataChunkPtr payload_collection;
 	// IGNORE NULLS
 	optional_ptr<ValidityMask> ignore_nulls;
 };
@@ -1471,26 +1459,46 @@ class WindowValueGlobalState : public WindowExecutorGlobalState {
 class WindowValueLocalState : public WindowExecutorBoundsState {
 public:
 	explicit WindowValueLocalState(const WindowValueGlobalState &gvstate)
-	    : WindowExecutorBoundsState(gvstate), gvstate(gvstate) {
+	    : WindowExecutorBoundsState(gvstate), gvstate(gvstate), builder(*gvstate.payload_collection) {
 	}
 
-	//! Lazily initialize for value Execute
-	void Initialize();
+	//! 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;
 
 	//! The corresponding global value state
 	const WindowValueGlobalState &gvstate;
-	//! Lazy initialization flag
-	bool initialized = false;
 	//! The exclusion filter handler
 	unique_ptr<ExclusionFilter> exclusion_filter;
 	//! 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::Initialize() {
-	if (initialized) {
-		return;
-	}
+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);
+
+	// Set up the IGNORE NULLS state
 	auto ignore_nulls = gvstate.ignore_nulls;
 	if (gvstate.executor.wexpr.exclude_clause == WindowExcludeMode::NO_OTHER) {
 		exclusion_filter = nullptr;
@@ -1502,7 +1510,10 @@ void WindowValueLocalState::Initialize() {
 		ignore_nulls_exclude = &exclusion_filter->mask;
 	}
 
-	initialized = true;
+	// Prepare to scan
+	if (!cursor) {
+		cursor = make_uniq<WindowCursor>(*payload_collection);
+	}
 }
 
 //===--------------------------------------------------------------------===//
@@ -1524,17 +1535,10 @@ unique_ptr<WindowExecutorGlobalState> WindowValueExecutor::GetGlobalState(const
 
 void WindowValueExecutor::Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
                                WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
-	auto &payload_chunk = lvstate.payload_chunk;
-	auto &payload_executor = lvstate.payload_executor;
-	auto &payload_collection = gvstate.payload_collection;
 
 	if (!wexpr.children.empty()) {
-		payload_chunk.Reset();
-		payload_executor.Execute(input_chunk, payload_chunk);
-		payload_chunk.Verify();
-		payload_collection.Copy(payload_chunk, input_idx);
+		lvstate.Sink(input_chunk, input_idx);
 	}
 
 	WindowExecutor::Sink(input_chunk, input_idx, total_count, gstate, lstate);
@@ -1547,18 +1551,16 @@ unique_ptr<WindowExecutorLocalState> WindowValueExecutor::GetLocalState(const Wi
 
 void WindowNtileExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
                                            Vector &result, idx_t count, idx_t row_idx) const {
-	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
-	auto &payload_collection = gvstate.payload_collection.chunk;
-	D_ASSERT(payload_collection.ColumnCount() == 1);
-	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]);
+	auto &lvstate = lstate.Cast<WindowValueLocalState>();
+	auto &cursor = *lvstate.cursor;
+	auto partition_begin = FlatVector::GetData<const idx_t>(lvstate.bounds.data[PARTITION_BEGIN]);
+	auto partition_end = FlatVector::GetData<const idx_t>(lvstate.bounds.data[PARTITION_END]);
 	auto rdata = FlatVector::GetData<int64_t>(result);
 	for (idx_t i = 0; i < count; ++i, ++row_idx) {
-		if (CellIsNull(payload_collection, 0, row_idx)) {
+		if (cursor.CellIsNull(0, row_idx)) {
 			FlatVector::SetNull(result, i, true);
 		} else {
-			auto n_param = GetCell<int64_t>(payload_collection, 0, row_idx);
+			auto n_param = cursor.GetCell<int64_t>(0, row_idx);
 			if (n_param < 1) {
 				throw InvalidInputException("Argument for ntile must be greater than zero");
 			}
@@ -1603,7 +1605,7 @@ class WindowLeadLagLocalState : public WindowValueLocalState {
 	      leadlag_default(gstate.executor.wexpr.default_expr.get(), gstate.executor.context) {
 	}
 
-	void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, const WindowInputColumn &range) override;
+	void UpdateBounds(idx_t row_idx, DataChunk &input_chunk, optional_ptr<WindowCursor> range) override;
 
 public:
 	// LEAD/LAG Evaluation
@@ -1611,7 +1613,7 @@ class WindowLeadLagLocalState : public WindowValueLocalState {
 	WindowInputExpression leadlag_default;
 };
 
-void WindowLeadLagLocalState::UpdateBounds(idx_t row_idx, DataChunk &input_chunk, const WindowInputColumn &range) {
+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);
@@ -1632,9 +1634,9 @@ WindowLeadLagExecutor::GetLocalState(const WindowExecutorGlobalState &gstate) co
 void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
                                              Vector &result, idx_t count, idx_t row_idx) const {
 	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
-	auto &payload_collection = gvstate.payload_collection.chunk;
 	auto &ignore_nulls = gvstate.ignore_nulls;
 	auto &llstate = lstate.Cast<WindowLeadLagLocalState>();
+	auto &cursor = *llstate.cursor;
 
 	bool can_shift = ignore_nulls->AllValid();
 	if (wexpr.offset_expr) {
@@ -1673,14 +1675,22 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 		if (can_shift) {
 			if (!delta) {
 				//	Copy source[index:index+width] => result[i:]
-				const auto index = NumericCast<idx_t>(val_idx);
+				auto index = NumericCast<idx_t>(val_idx);
 				const auto source_limit = partition_end[i] - index;
 				const auto target_limit = MinValue(partition_end[i], row_end) - row_idx;
-				const auto width = MinValue(source_limit, target_limit);
-				auto &source = payload_collection.data[0];
-				VectorOperations::Copy(source, result, index + width, index, i);
-				i += width;
-				row_idx += width;
+				auto width = MinValue(source_limit, target_limit);
+				// We may have to scan multiple blocks here, so loop until we have copied everything
+				const idx_t col_idx = 0;
+				while (width) {
+					const auto source_offset = cursor.Seek(index);
+					auto &source = cursor.chunk.data[col_idx];
+					const auto copied = MinValue<idx_t>(cursor.chunk.size() - source_offset, width);
+					VectorOperations::Copy(source, result, source_offset + copied, source_offset, i);
+					i += copied;
+					row_idx += copied;
+					index += copied;
+					width -= copied;
+				}
 			} else if (wexpr.default_expr) {
 				const auto width = MinValue(delta, count - i);
 				llstate.leadlag_default.CopyCell(result, i, width);
@@ -1693,7 +1703,7 @@ void WindowLeadLagExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate,
 			}
 		} else {
 			if (!delta) {
-				CopyCell(payload_collection, 0, NumericCast<idx_t>(val_idx), result, i);
+				cursor.CopyCell(0, NumericCast<idx_t>(val_idx), result, i);
 			} else if (wexpr.default_expr) {
 				llstate.leadlag_default.CopyCell(result, i);
 			} else {
@@ -1711,10 +1721,8 @@ WindowFirstValueExecutor::WindowFirstValueExecutor(BoundWindowExpression &wexpr,
 
 void WindowFirstValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
                                                 Vector &result, idx_t count, idx_t row_idx) const {
-	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
-	auto &payload_collection = gvstate.payload_collection.chunk;
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
-	lvstate.Initialize();
+	auto &cursor = *lvstate.cursor;
 	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]);
 	for (idx_t i = 0; i < count; ++i, ++row_idx) {
@@ -1731,7 +1739,7 @@ void WindowFirstValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstat
 		idx_t n = 1;
 		const auto first_idx = FindNextStart(*lvstate.ignore_nulls_exclude, window_begin[i], window_end[i], n);
 		if (!n) {
-			CopyCell(payload_collection, 0, first_idx, result, i);
+			cursor.CopyCell(0, first_idx, result, i);
 		} else {
 			FlatVector::SetNull(result, i, true);
 		}
@@ -1748,10 +1756,8 @@ WindowLastValueExecutor::WindowLastValueExecutor(BoundWindowExpression &wexpr, C
 
 void WindowLastValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
                                                Vector &result, idx_t count, idx_t row_idx) const {
-	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
-	auto &payload_collection = gvstate.payload_collection.chunk;
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
-	lvstate.Initialize();
+	auto &cursor = *lvstate.cursor;
 	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]);
 	for (idx_t i = 0; i < count; ++i, ++row_idx) {
@@ -1767,7 +1773,7 @@ void WindowLastValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate
 		idx_t n = 1;
 		const auto last_idx = FindPrevStart(*lvstate.ignore_nulls_exclude, window_begin[i], window_end[i], n);
 		if (!n) {
-			CopyCell(payload_collection, 0, last_idx, result, i);
+			cursor.CopyCell(0, last_idx, result, i);
 		} else {
 			FlatVector::SetNull(result, i, true);
 		}
@@ -1784,12 +1790,9 @@ WindowNthValueExecutor::WindowNthValueExecutor(BoundWindowExpression &wexpr, Cli
 
 void WindowNthValueExecutor::EvaluateInternal(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate,
                                               Vector &result, idx_t count, idx_t row_idx) const {
-	auto &gvstate = gstate.Cast<WindowValueGlobalState>();
-	auto &payload_collection = gvstate.payload_collection.chunk;
-	D_ASSERT(payload_collection.ColumnCount() == 2);
-
 	auto &lvstate = lstate.Cast<WindowValueLocalState>();
-	lvstate.Initialize();
+	auto &cursor = *lvstate.cursor;
+	D_ASSERT(cursor.chunk.ColumnCount() == 2);
 	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]);
 	for (idx_t i = 0; i < count; ++i, ++row_idx) {
@@ -1804,17 +1807,17 @@ 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 (CellIsNull(payload_collection, 1, row_idx)) {
+		if (cursor.CellIsNull(1, row_idx)) {
 			FlatVector::SetNull(result, i, true);
 		} else {
-			auto n_param = GetCell<int64_t>(payload_collection, 1, row_idx);
+			auto n_param = cursor.GetCell<int64_t>(1, row_idx);
 			if (n_param < 1) {
 				FlatVector::SetNull(result, i, true);
 			} else {
 				auto n = idx_t(n_param);
 				const auto nth_index = FindNextStart(*lvstate.ignore_nulls_exclude, window_begin[i], window_end[i], n);
 				if (!n) {
-					CopyCell(payload_collection, 0, nth_index, result, i);
+					cursor.CopyCell(0, nth_index, result, i);
 				} else {
 					FlatVector::SetNull(result, i, true);
 				}
diff --git a/src/execution/window_segment_tree.cpp b/src/execution/window_segment_tree.cpp
index be71e85c78b1..930b526083ae 100644
--- a/src/execution/window_segment_tree.cpp
+++ b/src/execution/window_segment_tree.cpp
@@ -22,12 +22,11 @@ WindowAggregatorState::WindowAggregatorState() : allocator(Allocator::DefaultAll
 
 class WindowAggregatorGlobalState : public WindowAggregatorState {
 public:
-	WindowAggregatorGlobalState(const WindowAggregator &aggregator_p, idx_t group_count)
-	    : aggregator(aggregator_p), winputs(inputs), locals(0), finalized(0) {
+	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) {
 
-		if (!aggregator.arg_types.empty()) {
-			winputs.Initialize(Allocator::DefaultAllocator(), aggregator.arg_types, group_count);
-		}
 		if (aggregator.aggr.filter) {
 			// 	Start with all invalid and set the ones that pass
 			filter_mask.Initialize(group_count, false);
@@ -38,8 +37,10 @@ class WindowAggregatorGlobalState : public WindowAggregatorState {
 	const WindowAggregator &aggregator;
 
 	//! Partition data chunk
-	DataChunk inputs;
-	WindowDataChunk winputs;
+	WindowCollection winputs;
+
+	//! Validity flag for the entire collection
+	atomic<bool> all_valid;
 
 	//! The filtered rows in inputs.
 	ValidityArray filter_mask;
@@ -54,6 +55,20 @@ class WindowAggregatorGlobalState : public WindowAggregatorState {
 	std::atomic<idx_t> finalized;
 };
 
+class WindowAggregatorLocalState : public WindowAggregatorState {
+public:
+	WindowAggregatorLocalState() {
+	}
+
+	void Sink(WindowAggregatorGlobalState &gastate, DataChunk &input, idx_t row_idx);
+	virtual void Finalize(WindowAggregatorGlobalState &gastate);
+
+	//! 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),
@@ -63,26 +78,50 @@ WindowAggregator::WindowAggregator(AggregateObject aggr_p, const vector<LogicalT
 WindowAggregator::~WindowAggregator() {
 }
 
-unique_ptr<WindowAggregatorState> WindowAggregator::GetGlobalState(idx_t group_count, const ValidityMask &) const {
-	return make_uniq<WindowAggregatorGlobalState>(*this, group_count);
+unique_ptr<WindowAggregatorState> WindowAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
+                                                                   const ValidityMask &) const {
+	return make_uniq<WindowAggregatorGlobalState>(context, *this, group_count);
 }
 
-void WindowAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk,
-                            idx_t input_idx, optional_ptr<SelectionVector> filter_sel, idx_t filtered) {
-	auto &gasink = gsink.Cast<WindowAggregatorGlobalState>();
-	auto &winputs = gasink.winputs;
-	auto &filter_mask = gasink.filter_mask;
-	if (winputs.chunk.ColumnCount()) {
-		winputs.Copy(arg_chunk, input_idx);
+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 WindowAggregator::Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &arg_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);
 	if (filter_sel) {
+		auto &filter_mask = gastate.filter_mask;
 		for (idx_t f = 0; f < filtered; ++f) {
 			filter_mask.SetValid(input_idx + filter_sel->get_index(f));
 		}
 	}
 }
 
+void WindowAggregatorLocalState::Finalize(WindowAggregatorGlobalState &gastate) {
+	auto &winputs = gastate.winputs;
+
+	// Thread-safe and idempotent
+	winputs.Combine(false);
+
+	// Prepare to scan
+	if (!cursor) {
+		cursor = make_uniq<WindowCursor>(winputs);
+	}
+}
+
 void WindowAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, const FrameStats &stats) {
+	auto &gasink = gstate.Cast<WindowAggregatorGlobalState>();
+	auto &lastate = lstate.Cast<WindowAggregatorLocalState>();
+	lastate.Finalize(gasink);
 }
 
 //===--------------------------------------------------------------------===//
@@ -174,7 +213,7 @@ void WindowAggregateStates::Destroy() {
 
 class WindowConstantAggregatorGlobalState : public WindowAggregatorGlobalState {
 public:
-	WindowConstantAggregatorGlobalState(const WindowConstantAggregator &aggregator, idx_t count,
+	WindowConstantAggregatorGlobalState(ClientContext &context, const WindowConstantAggregator &aggregator, idx_t count,
 	                                    const ValidityMask &partition_mask);
 
 	void Finalize(const FrameStats &stats);
@@ -187,7 +226,7 @@ class WindowConstantAggregatorGlobalState : public WindowAggregatorGlobalState {
 	unique_ptr<Vector> results;
 };
 
-class WindowConstantAggregatorLocalState : public WindowAggregatorState {
+class WindowConstantAggregatorLocalState : public WindowAggregatorLocalState {
 public:
 	explicit WindowConstantAggregatorLocalState(const WindowConstantAggregatorGlobalState &gstate);
 	~WindowConstantAggregatorLocalState() override {
@@ -211,10 +250,11 @@ class WindowConstantAggregatorLocalState : public WindowAggregatorState {
 	SelectionVector matches;
 };
 
-WindowConstantAggregatorGlobalState::WindowConstantAggregatorGlobalState(const WindowConstantAggregator &aggregator,
+WindowConstantAggregatorGlobalState::WindowConstantAggregatorGlobalState(ClientContext &context,
+                                                                         const WindowConstantAggregator &aggregator,
                                                                          idx_t group_count,
                                                                          const ValidityMask &partition_mask)
-    : WindowAggregatorGlobalState(aggregator, STANDARD_VECTOR_SIZE), statef(aggregator.aggr) {
+    : WindowAggregatorGlobalState(context, aggregator, STANDARD_VECTOR_SIZE), statef(aggregator.aggr) {
 
 	// Locate the partition boundaries
 	if (partition_mask.AllValid()) {
@@ -274,9 +314,9 @@ WindowConstantAggregator::WindowConstantAggregator(AggregateObject aggr, const v
     : WindowAggregator(std::move(aggr), arg_types, result_type, exclude_mode_p) {
 }
 
-unique_ptr<WindowAggregatorState> WindowConstantAggregator::GetGlobalState(idx_t group_count,
+unique_ptr<WindowAggregatorState> WindowConstantAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
                                                                            const ValidityMask &partition_mask) const {
-	return make_uniq<WindowConstantAggregatorGlobalState>(*this, group_count, partition_mask);
+	return make_uniq<WindowConstantAggregatorGlobalState>(context, *this, group_count, partition_mask);
 }
 
 void WindowConstantAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk,
@@ -437,7 +477,7 @@ WindowCustomAggregator::WindowCustomAggregator(AggregateObject aggr, const vecto
 WindowCustomAggregator::~WindowCustomAggregator() {
 }
 
-class WindowCustomAggregatorState : public WindowAggregatorState {
+class WindowCustomAggregatorState : public WindowAggregatorLocalState {
 public:
 	WindowCustomAggregatorState(const AggregateObject &aggr, const WindowExcludeMode exclude_mode);
 	~WindowCustomAggregatorState() override;
@@ -472,12 +512,15 @@ static void InitSubFrames(SubFrames &frames, const WindowExcludeMode exclude_mod
 
 class WindowCustomAggregatorGlobalState : public WindowAggregatorGlobalState {
 public:
-	explicit WindowCustomAggregatorGlobalState(const WindowCustomAggregator &aggregator, idx_t group_count)
-	    : WindowAggregatorGlobalState(aggregator, group_count) {
+	explicit WindowCustomAggregatorGlobalState(ClientContext &context, const WindowCustomAggregator &aggregator,
+	                                           idx_t group_count)
+	    : WindowAggregatorGlobalState(context, aggregator, group_count), context(context) {
 
 		gcstate = make_uniq<WindowCustomAggregatorState>(aggregator.aggr, aggregator.exclude_mode);
 	}
 
+	//! Buffer manager for paging custom accelerator data
+	ClientContext &context;
 	//! Traditional packed filter mask for API
 	ValidityMask filter_packed;
 	//! Data pointer that contains a single local state, used for global custom window execution state
@@ -503,9 +546,9 @@ WindowCustomAggregatorState::~WindowCustomAggregatorState() {
 	}
 }
 
-unique_ptr<WindowAggregatorState> WindowCustomAggregator::GetGlobalState(idx_t group_count,
+unique_ptr<WindowAggregatorState> WindowCustomAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
                                                                          const ValidityMask &) const {
-	return make_uniq<WindowCustomAggregatorGlobalState>(*this, group_count);
+	return make_uniq<WindowCustomAggregatorGlobalState>(context, *this, group_count);
 }
 
 void WindowCustomAggregator::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate,
@@ -519,13 +562,15 @@ void WindowCustomAggregator::Finalize(WindowAggregatorState &gsink, WindowAggreg
 
 	WindowAggregator::Finalize(gsink, lstate, stats);
 
-	auto &inputs = gcsink.inputs;
+	auto inputs = gcsink.winputs.inputs.get();
+	const auto count = gcsink.winputs.size();
+	const auto all_valid = gcsink.winputs.all_valid.load();
 	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>(inputs.data.data(), inputs.ColumnCount(), inputs.size(), filter_packed, stats);
+	    make_uniq<WindowPartitionInput>(gcsink.context, inputs, count, all_valid, filter_packed, stats);
 
 	if (aggr.function.window_init) {
 		auto &gcstate = *gcsink.gcstate;
@@ -627,36 +672,38 @@ WindowNaiveAggregator::WindowNaiveAggregator(AggregateObject aggr, const vector<
 WindowNaiveAggregator::~WindowNaiveAggregator() {
 }
 
-class WindowNaiveState : public WindowAggregatorState {
+class WindowNaiveState : public WindowAggregatorLocalState {
 public:
 	struct HashRow {
-		HashRow(WindowNaiveState &state, const DataChunk &inputs) : state(state), inputs(inputs) {
+		explicit HashRow(WindowNaiveState &state, const WindowCollection &winputs) : state(state), winputs(winputs) {
 		}
 
-		size_t operator()(const idx_t &i) const {
-			return state.Hash(inputs, i);
+		inline size_t operator()(const idx_t &i) const {
+			return state.Hash(winputs, i);
 		}
 
 		WindowNaiveState &state;
-		const DataChunk &inputs;
+		const WindowCollection &winputs;
 	};
 
 	struct EqualRow {
-		EqualRow(WindowNaiveState &state, const DataChunk &inputs) : state(state), inputs(inputs) {
+		EqualRow(WindowNaiveState &state, const WindowCollection &winputs) : state(state), winputs(winputs) {
 		}
 
-		bool operator()(const idx_t &lhs, const idx_t &rhs) const {
-			return state.KeyEqual(inputs, lhs, rhs);
+		inline bool operator()(const idx_t &lhs, const idx_t &rhs) const {
+			return state.KeyEqual(winputs, lhs, rhs);
 		}
 
 		WindowNaiveState &state;
-		const DataChunk &inputs;
+		const WindowCollection &winputs;
 	};
 
 	using RowSet = std::unordered_set<idx_t, HashRow, EqualRow>;
 
 	explicit WindowNaiveState(const WindowNaiveAggregator &gsink);
 
+	void Finalize(WindowAggregatorGlobalState &gastate) override;
+
 	void Evaluate(const WindowAggregatorGlobalState &gsink, const DataChunk &bounds, Vector &result, idx_t count,
 	              idx_t row_idx);
 
@@ -665,9 +712,9 @@ class WindowNaiveState : public WindowAggregatorState {
 	void FlushStates(const WindowAggregatorGlobalState &gsink);
 
 	//! Hashes a value for the hash table
-	size_t Hash(const DataChunk &inputs, idx_t rid);
+	size_t Hash(const WindowCollection &inputs, idx_t rid);
 	//! Compares two values for the hash table
-	bool KeyEqual(const DataChunk &inputs, const idx_t &lhs, const idx_t &rhs);
+	bool KeyEqual(const WindowCollection &inputs, const idx_t &lhs, const idx_t &rhs);
 
 	//! The global state
 	const WindowNaiveAggregator &aggregator;
@@ -687,6 +734,8 @@ class WindowNaiveState : public WindowAggregatorState {
 	SubFrames frames;
 	//! The optional hash table used for DISTINCT
 	Vector hashes;
+	//! The state used for comparing the collection across chunk boundaries
+	unique_ptr<WindowCursor> comparer;
 };
 
 WindowNaiveState::WindowNaiveState(const WindowNaiveAggregator &aggregator_p)
@@ -708,13 +757,22 @@ WindowNaiveState::WindowNaiveState(const WindowNaiveAggregator &aggregator_p)
 	}
 }
 
+void WindowNaiveState::Finalize(WindowAggregatorGlobalState &gastate) {
+	WindowAggregatorLocalState::Finalize(gastate);
+
+	//	Set up the comparison scanner just in case
+	if (!comparer) {
+		comparer = make_uniq<WindowCursor>(gastate.winputs);
+	}
+}
+
 void WindowNaiveState::FlushStates(const WindowAggregatorGlobalState &gsink) {
 	if (!flush_count) {
 		return;
 	}
 
-	auto &inputs = gsink.inputs;
-	leaves.Slice(inputs, update_sel, flush_count);
+	auto &scanned = cursor->chunk;
+	leaves.Slice(scanned, update_sel, flush_count);
 
 	auto &aggr = aggregator.aggr;
 	AggregateInputData aggr_input_data(aggr.GetFunctionData(), allocator);
@@ -723,28 +781,46 @@ void WindowNaiveState::FlushStates(const WindowAggregatorGlobalState &gsink) {
 	flush_count = 0;
 }
 
-size_t WindowNaiveState::Hash(const DataChunk &inputs, idx_t rid) {
-	auto s = UnsafeNumericCast<sel_t>(rid);
+size_t WindowNaiveState::Hash(const WindowCollection &winputs, idx_t rid) {
+	D_ASSERT(cursor->RowIsVisible(rid));
+	auto s = cursor->RowOffset(rid);
+	auto &scanned = cursor->chunk;
 	SelectionVector sel(&s);
-	leaves.Slice(inputs, sel, 1);
+	leaves.Slice(scanned, sel, 1);
 	leaves.Hash(hashes);
 
 	return *FlatVector::GetData<hash_t>(hashes);
 }
 
-bool WindowNaiveState::KeyEqual(const DataChunk &inputs, const idx_t &lhs, const idx_t &rhs) {
-	auto l = UnsafeNumericCast<sel_t>(lhs);
+bool WindowNaiveState::KeyEqual(const WindowCollection &winputs, 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;
+	if (!cursor->RowIsVisible(lhs)) {
+		std::swap(lhs, rhs);
+		D_ASSERT(cursor->RowIsVisible(lhs));
+	}
+
+	auto &scanned = cursor->chunk;
+	auto l = cursor->RowOffset(lhs);
 	SelectionVector lsel(&l);
 
-	auto r = UnsafeNumericCast<sel_t>(rhs);
+	auto rreader = cursor.get();
+	if (!cursor->RowIsVisible(rhs)) {
+		//	Values on different pages!
+		rreader = comparer.get();
+		rreader->Seek(rhs);
+	}
+	auto rscanned = &rreader->chunk;
+	auto r = rreader->RowOffset(rhs);
 	SelectionVector rsel(&r);
 
 	sel_t f = 0;
 	SelectionVector fsel(&f);
 
-	for (auto &input : inputs.data) {
-		Vector left(input, lsel, 1);
-		Vector right(input, rsel, 1);
+	for (column_t c = 0; c < scanned.ColumnCount(); ++c) {
+		Vector left(scanned.data[c], lsel, 1);
+		Vector right(rscanned->data[c], rsel, 1);
 		if (!VectorOperations::NotDistinctFrom(left, right, nullptr, 1, nullptr, &fsel)) {
 			return false;
 		}
@@ -757,17 +833,18 @@ void WindowNaiveState::Evaluate(const WindowAggregatorGlobalState &gsink, const
                                 idx_t count, idx_t row_idx) {
 	auto &aggr = aggregator.aggr;
 	auto &filter_mask = gsink.filter_mask;
-	auto &inputs = gsink.inputs;
+	auto &winputs = gsink.winputs;
+	auto &types = winputs.GetTypes();
 
-	if (leaves.ColumnCount() == 0 && inputs.ColumnCount() > 0) {
-		leaves.Initialize(Allocator::DefaultAllocator(), inputs.GetTypes());
+	if (leaves.ColumnCount() == 0 && !types.empty()) {
+		leaves.Initialize(Allocator::DefaultAllocator(), types);
 	}
 
 	auto fdata = FlatVector::GetData<data_ptr_t>(statef);
 	auto pdata = FlatVector::GetData<data_ptr_t>(statep);
 
-	HashRow hash_row(*this, inputs);
-	EqualRow equal_row(*this, inputs);
+	HashRow hash_row(*this, winputs);
+	EqualRow equal_row(*this, winputs);
 	RowSet row_set(STANDARD_VECTOR_SIZE, hash_row, equal_row);
 
 	EvaluateSubFrames(bounds, aggregator.exclude_mode, count, row_idx, frames, [&](idx_t rid) {
@@ -782,13 +859,20 @@ void WindowNaiveState::Evaluate(const WindowAggregatorGlobalState &gsink, const
 					continue;
 				}
 
+				//	Seek to the current position
+				if (!cursor->RowIsVisible(f)) {
+					//	We need to flush when we cross a chunk boundary
+					FlushStates(gsink);
+					cursor->Seek(f);
+				}
+
 				//	Filter out duplicates
 				if (aggr.IsDistinct() && !row_set.insert(f).second) {
 					continue;
 				}
 
 				pdata[flush_count] = agg_state;
-				update_sel[flush_count++] = UnsafeNumericCast<sel_t>(f);
+				update_sel[flush_count++] = cursor->RowOffset(f);
 				if (flush_count >= STANDARD_VECTOR_SIZE) {
 					FlushStates(gsink);
 				}
@@ -827,7 +911,7 @@ class WindowSegmentTreeGlobalState : public WindowAggregatorGlobalState {
 public:
 	using AtomicCounters = vector<std::atomic<idx_t>>;
 
-	WindowSegmentTreeGlobalState(const WindowSegmentTree &aggregator, idx_t group_count);
+	WindowSegmentTreeGlobalState(ClientContext &context, const WindowSegmentTree &aggregator, idx_t group_count);
 
 	ArenaAllocator &CreateTreeAllocator() {
 		lock_guard<mutex> tree_lock(lock);
@@ -869,7 +953,7 @@ class WindowSegmentTreePart {
 
 	enum FramePart : uint8_t { FULL = 0, LEFT = 1, RIGHT = 2 };
 
-	WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr, const DataChunk &inputs,
+	WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr, const WindowCollection &inputs,
 	                      const ValidityArray &filter_mask);
 	~WindowSegmentTreePart();
 
@@ -906,13 +990,15 @@ class WindowSegmentTreePart {
 	//! Order insensitive aggregate (we can optimise internal combines)
 	const bool order_insensitive;
 	//! The partition arguments
-	const DataChunk &inputs;
+	const WindowCollection &inputs;
 	//! The filtered rows in inputs
 	const ValidityArray &filter_mask;
 	//! The size of a single aggregate state
 	const idx_t state_size;
 	//! Data pointer that contains a vector of states, used for intermediate window segment aggregation
 	vector<data_t> state;
+	//! Scanned data state
+	unique_ptr<WindowCursor> cursor;
 	//! Input data chunk, used for leaf segment aggregation
 	DataChunk leaves;
 	//! The filtered rows in inputs.
@@ -929,12 +1015,12 @@ class WindowSegmentTreePart {
 	vector<RightEntry> right_stack;
 };
 
-class WindowSegmentTreeState : public WindowAggregatorState {
+class WindowSegmentTreeState : public WindowAggregatorLocalState {
 public:
 	WindowSegmentTreeState() {
 	}
 
-	void Finalize(WindowSegmentTreeGlobalState &gstate);
+	void Finalize(WindowAggregatorGlobalState &gastate) override;
 	void Evaluate(const WindowSegmentTreeGlobalState &gsink, const DataChunk &bounds, Vector &result, idx_t count,
 	              idx_t row_idx);
 	//! The left (default) segment tree part
@@ -946,11 +1032,11 @@ class WindowSegmentTreeState : public WindowAggregatorState {
 void WindowSegmentTree::Finalize(WindowAggregatorState &gsink, WindowAggregatorState &lstate, const FrameStats &stats) {
 
 	auto &gasink = gsink.Cast<WindowSegmentTreeGlobalState>();
-	auto &inputs = gasink.inputs;
+	auto &types = gasink.winputs.GetTypes();
 
 	WindowAggregator::Finalize(gsink, lstate, stats);
 
-	if (inputs.ColumnCount() > 0) {
+	if (!types.empty()) {
 		if (aggr.function.combine && UseCombineAPI()) {
 			lstate.Cast<WindowSegmentTreeState>().Finalize(gasink);
 		}
@@ -960,7 +1046,7 @@ void WindowSegmentTree::Finalize(WindowAggregatorState &gsink, WindowAggregatorS
 }
 
 WindowSegmentTreePart::WindowSegmentTreePart(ArenaAllocator &allocator, const AggregateObject &aggr,
-                                             const DataChunk &inputs, const ValidityArray &filter_mask)
+                                             const WindowCollection &inputs, const ValidityArray &filter_mask)
     : allocator(allocator), aggr(aggr),
       order_insensitive(aggr.function.order_dependent == AggregateOrderDependent::NOT_ORDER_DEPENDENT), inputs(inputs),
       filter_mask(filter_mask), state_size(aggr.function.state_size(aggr.function)),
@@ -981,14 +1067,19 @@ 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() {
 }
 
-unique_ptr<WindowAggregatorState> WindowSegmentTree::GetGlobalState(idx_t group_count,
+unique_ptr<WindowAggregatorState> WindowSegmentTree::GetGlobalState(ClientContext &context, idx_t group_count,
                                                                     const ValidityMask &partition_mask) const {
-	return make_uniq<WindowSegmentTreeGlobalState>(*this, group_count);
+	return make_uniq<WindowSegmentTreeGlobalState>(context, *this, group_count);
 }
 
 unique_ptr<WindowAggregatorState> WindowSegmentTree::GetLocalState(const WindowAggregatorState &gstate) const {
@@ -1005,7 +1096,8 @@ void WindowSegmentTreePart::FlushStates(bool combining) {
 		statel.Verify(flush_count);
 		aggr.function.combine(statel, statep, aggr_input_data, flush_count);
 	} else {
-		leaves.Slice(inputs, filter_sel, flush_count);
+		auto &scanned = cursor->chunk;
+		leaves.Slice(scanned, filter_sel, flush_count);
 		aggr.function.update(&leaves.data[0], aggr_input_data, leaves.ColumnCount(), statep, flush_count);
 	}
 
@@ -1025,9 +1117,10 @@ void WindowSegmentTreePart::ExtractFrame(idx_t begin, idx_t end, data_ptr_t stat
 	//	Otherwise set it to the input rows that pass the filter
 	auto states = FlatVector::GetData<data_ptr_t>(statep);
 	if (filter_mask.AllValid()) {
+		const auto offset = cursor->RowOffset(begin);
 		for (idx_t i = 0; i < count; ++i) {
 			states[flush_count] = state_ptr;
-			filter_sel.set_index(flush_count++, begin + i);
+			filter_sel.set_index(flush_count++, offset + i);
 			if (flush_count >= STANDARD_VECTOR_SIZE) {
 				FlushStates(false);
 			}
@@ -1036,7 +1129,7 @@ void WindowSegmentTreePart::ExtractFrame(idx_t begin, idx_t end, data_ptr_t stat
 		for (idx_t i = begin; i < end; ++i) {
 			if (filter_mask.RowIsValid(i)) {
 				states[flush_count] = state_ptr;
-				filter_sel.set_index(flush_count++, i);
+				filter_sel.set_index(flush_count++, cursor->RowOffset(i));
 				if (flush_count >= STANDARD_VECTOR_SIZE) {
 					FlushStates(false);
 				}
@@ -1054,7 +1147,16 @@ void WindowSegmentTreePart::WindowSegmentValue(const WindowSegmentTreeGlobalStat
 
 	const auto count = end - begin;
 	if (l_idx == 0) {
-		ExtractFrame(begin, end, state_ptr);
+		//	Check the leaves when they cross chunk boundaries
+		while (begin < end) {
+			if (!cursor->RowIsVisible(begin)) {
+				FlushStates(false);
+				cursor->Seek(begin);
+			}
+			auto next = MinValue(end, cursor->state.next_row_index);
+			ExtractFrame(begin, next, state_ptr);
+			begin = next;
+		}
 	} else {
 		// find out where the states begin
 		auto begin_ptr = tree.levels_flat_native.GetStatePtr(begin + tree.levels_flat_start[l_idx - 1]);
@@ -1082,10 +1184,12 @@ void WindowSegmentTreePart::Finalize(Vector &result, idx_t count) {
 	}
 }
 
-WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(const WindowSegmentTree &aggregator, idx_t group_count)
-    : WindowAggregatorGlobalState(aggregator, group_count), tree(aggregator), levels_flat_native(aggregator.aggr) {
+WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(ClientContext &context, const WindowSegmentTree &aggregator,
+                                                           idx_t group_count)
+    : WindowAggregatorGlobalState(context, aggregator, group_count), tree(aggregator),
+      levels_flat_native(aggregator.aggr) {
 
-	D_ASSERT(inputs.ColumnCount() > 0);
+	D_ASSERT(!winputs.GetTypes().empty());
 
 	// compute space required to store internal nodes of segment tree
 	levels_flat_start.push_back(0);
@@ -1096,7 +1200,7 @@ WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(const WindowSegmentTr
 	idx_t level_size;
 	// iterate over the levels of the segment tree
 	while ((level_size =
-	            (level_current == 0 ? inputs.size() : levels_flat_offset - levels_flat_start[level_current - 1])) > 1) {
+	            (level_current == 0 ? group_count : levels_flat_offset - levels_flat_start[level_current - 1])) > 1) {
 		for (idx_t pos = 0; pos < level_size; pos += TREE_FANOUT) {
 			levels_flat_offset++;
 		}
@@ -1126,9 +1230,12 @@ WindowSegmentTreeGlobalState::WindowSegmentTreeGlobalState(const WindowSegmentTr
 	}
 }
 
-void WindowSegmentTreeState::Finalize(WindowSegmentTreeGlobalState &gstate) {
+void WindowSegmentTreeState::Finalize(WindowAggregatorGlobalState &gastate) {
+	WindowAggregatorLocalState::Finalize(gastate);
+
 	//	Single part for constructing the tree
-	auto &inputs = gstate.inputs;
+	auto &gstate = gastate.Cast<WindowSegmentTreeGlobalState>();
+	auto &inputs = gstate.winputs;
 	auto &tree = gstate.tree;
 	auto &filter_mask = gstate.filter_mask;
 	WindowSegmentTreePart gtstate(gstate.CreateTreeAllocator(), tree.aggr, inputs, filter_mask);
@@ -1194,8 +1301,8 @@ void WindowSegmentTreeState::Evaluate(const WindowSegmentTreeGlobalState &gtstat
 	auto peer_end = FlatVector::GetData<const idx_t>(bounds.data[PEER_END]);
 
 	if (!part) {
-		part =
-		    make_uniq<WindowSegmentTreePart>(allocator, gtstate.aggregator.aggr, gtstate.inputs, gtstate.filter_mask);
+		auto &inputs = gtstate.winputs;
+		part = make_uniq<WindowSegmentTreePart>(allocator, gtstate.aggregator.aggr, inputs, gtstate.filter_mask);
 	}
 
 	if (gtstate.aggregator.exclude_mode != WindowExcludeMode::NO_OTHER) {
@@ -1433,7 +1540,8 @@ class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 	using ZippedTuple = WindowDistinctSortTree::ZippedTuple;
 	using ZippedElements = WindowDistinctSortTree::ZippedElements;
 
-	WindowDistinctAggregatorGlobalState(const WindowDistinctAggregator &aggregator, idx_t group_count);
+	WindowDistinctAggregatorGlobalState(ClientContext &context, const WindowDistinctAggregator &aggregator,
+	                                    idx_t group_count);
 
 	//! Compute the block starts
 	void MeasurePayloadBlocks();
@@ -1479,9 +1587,10 @@ class WindowDistinctAggregatorGlobalState : public WindowAggregatorGlobalState {
 	vector<idx_t> levels_flat_start;
 };
 
-WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(const WindowDistinctAggregator &aggregator,
+WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(ClientContext &context,
+                                                                         const WindowDistinctAggregator &aggregator,
                                                                          idx_t group_count)
-    : WindowAggregatorGlobalState(aggregator, group_count), context(aggregator.context),
+    : 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) {
 	payload_types.emplace_back(LogicalType::UBIGINT);
@@ -1538,7 +1647,7 @@ WindowDistinctAggregatorGlobalState::WindowDistinctAggregatorGlobalState(const W
 	}
 }
 
-class WindowDistinctAggregatorLocalState : public WindowAggregatorState {
+class WindowDistinctAggregatorLocalState : public WindowAggregatorLocalState {
 public:
 	explicit WindowDistinctAggregatorLocalState(const WindowDistinctAggregatorGlobalState &aggregator);
 
@@ -1595,15 +1704,16 @@ WindowDistinctAggregatorLocalState::WindowDistinctAggregatorLocalState(
 	sort_chunk.data.back().Reference(payload_chunk.data[0]);
 
 	//! Input data chunk, used for leaf segment aggregation
-	leaves.Initialize(Allocator::DefaultAllocator(), gastate.inputs.GetTypes());
+	auto &inputs = gastate.winputs;
+	leaves.Initialize(Allocator::DefaultAllocator(), inputs.GetTypes());
 	sel.Initialize();
 
 	gastate.locals++;
 }
 
-unique_ptr<WindowAggregatorState> WindowDistinctAggregator::GetGlobalState(idx_t group_count,
+unique_ptr<WindowAggregatorState> WindowDistinctAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
                                                                            const ValidityMask &partition_mask) const {
-	return make_uniq<WindowDistinctAggregatorGlobalState>(*this, group_count);
+	return make_uniq<WindowDistinctAggregatorGlobalState>(context, *this, group_count);
 }
 
 void WindowDistinctAggregator::Sink(WindowAggregatorState &gsink, WindowAggregatorState &lstate, DataChunk &arg_chunk,
@@ -1757,6 +1867,7 @@ void WindowDistinctAggregator::Finalize(WindowAggregatorState &gsink, WindowAggr
                                         const FrameStats &stats) {
 	auto &gdsink = gsink.Cast<WindowDistinctAggregatorGlobalState>();
 	auto &ldstate = lstate.Cast<WindowDistinctAggregatorLocalState>();
+	ldstate.Finalize(gdsink);
 
 	//	5: Sort sorted lexicographically increasing
 	ldstate.ExecuteTask();
@@ -1922,7 +2033,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 &allocator = gdastate.allocator;
-	auto &inputs = gdastate.inputs;
+	auto &inputs = ldastate.cursor->chunk;
 	auto &levels_flat_native = gdastate.levels_flat_native;
 
 	//! Input data chunk, used for leaf segment aggregation
@@ -1959,9 +2070,26 @@ void WindowDistinctSortTree::BuildRun(idx_t level_nr, idx_t run_idx, WindowDisti
 		const auto prev_idx = std::get<0>(zipped_level[j]);
 		level[j] = prev_idx;
 		if (prev_idx < i + 1) {
+			const auto update_idx = std::get<1>(zipped_level[j]);
+			if (!ldastate.cursor->RowIsVisible(update_idx)) {
+				// 	Flush if we have to move the cursor
+				//	Push the updates first so they propagate
+				leaves.Reference(inputs);
+				leaves.Slice(sel, nupdate);
+				aggr.function.update(leaves.data.data(), aggr_input_data, leaves.ColumnCount(), update_v, nupdate);
+				nupdate = 0;
+
+				//	Combine the states sequentially
+				aggr.function.combine(source_v, target_v, aggr_input_data, ncombine);
+				ncombine = 0;
+
+				// Move the update into range.
+				ldastate.cursor->Seek(update_idx);
+			}
+
 			updates[nupdate] = curr_state;
 			//	input_idx
-			sel[nupdate] = UnsafeNumericCast<sel_t>(std::get<1>(zipped_level[j]));
+			sel[nupdate] = ldastate.cursor->RowOffset(update_idx);
 			++nupdate;
 		}
 
diff --git a/src/function/aggregate/distributive/count.cpp b/src/function/aggregate/distributive/count.cpp
index 8b5c3be2b730..d4e837aaa00d 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.input_count == 0);
+		D_ASSERT(partition.inputs == nullptr);
 
 		auto data = FlatVector::GetData<RESULT_TYPE>(result);
 		RESULT_TYPE total = 0;
diff --git a/src/include/duckdb/common/types/column/column_data_collection.hpp b/src/include/duckdb/common/types/column/column_data_collection.hpp
index 1a7473236a5e..571826617a3d 100644
--- a/src/include/duckdb/common/types/column/column_data_collection.hpp
+++ b/src/include/duckdb/common/types/column/column_data_collection.hpp
@@ -141,10 +141,17 @@ class ColumnDataCollection {
 
 	//! Obtains the next scan index to scan from
 	bool NextScanIndex(ColumnDataScanState &state, idx_t &chunk_index, idx_t &segment_index, idx_t &row_index) const;
+	//! Obtains the previous scan index to scan from
+	bool PrevScanIndex(ColumnDataScanState &state, idx_t &chunk_index, idx_t &segment_index, idx_t &row_index) const;
 	//! Scans at the indices (obtained from NextScanIndex)
 	void ScanAtIndex(ColumnDataParallelScanState &state, ColumnDataLocalScanState &lstate, DataChunk &result,
 	                 idx_t chunk_index, idx_t segment_index, idx_t row_index) const;
 
+	//! Seeks to the chunk _containing_ the row. Returns false if it is past the end.
+	//! Note that the returned chunk will likely not be aligned to the given row
+	//! but the scan state will provide the actual range
+	bool Seek(idx_t row_idx, ColumnDataScanState &state, DataChunk &result) const;
+
 	//! Initialize the column data collection
 	void Initialize(vector<LogicalType> types);
 
@@ -190,8 +197,8 @@ class ColumnDataRowCollection {
 
 public:
 	// container API
-	bool empty() const { // NOLINT: match stl API
-		return rows.empty();
+	bool empty() const {     // NOLINT: match stl API
+		return rows.empty(); // NOLINT
 	}
 	idx_t size() const { // NOLINT: match stl API
 		return rows.size();
diff --git a/src/include/duckdb/common/vector_operations/aggregate_executor.hpp b/src/include/duckdb/common/vector_operations/aggregate_executor.hpp
index 239018599a58..6abe7ff1e8d7 100644
--- a/src/include/duckdb/common/vector_operations/aggregate_executor.hpp
+++ b/src/include/duckdb/common/vector_operations/aggregate_executor.hpp
@@ -399,19 +399,6 @@ class AggregateExecutor {
 		}
 	}
 
-	template <class STATE, class INPUT_TYPE, class RESULT_TYPE, class OP>
-	static void UnaryWindow(const Vector &input, const ValidityMask &ifilter, AggregateInputData &aggr_input_data,
-	                        data_ptr_t state_p, const SubFrames &frames, Vector &result, idx_t ridx,
-	                        const_data_ptr_t gstate_p) {
-
-		auto idata = FlatVector::GetData<const INPUT_TYPE>(input);
-		const auto &ivalid = FlatVector::Validity(input);
-		auto &state = *reinterpret_cast<STATE *>(state_p);
-		auto gstate = reinterpret_cast<const STATE *>(gstate_p);
-		OP::template Window<STATE, INPUT_TYPE, RESULT_TYPE>(idata, ifilter, ivalid, aggr_input_data, state, frames,
-		                                                    result, ridx, gstate);
-	}
-
 	template <typename OP>
 	static void IntersectFrames(const SubFrames &lefts, const SubFrames &rights, OP &op) {
 		const auto cover_start = MinValue(rights[0].start, lefts[0].start);
diff --git a/src/include/duckdb/execution/window_executor.hpp b/src/include/duckdb/execution/window_executor.hpp
index 0c9a7c340707..b64c8ad74cca 100644
--- a/src/include/duckdb/execution/window_executor.hpp
+++ b/src/include/duckdb/execution/window_executor.hpp
@@ -15,30 +15,128 @@
 
 namespace duckdb {
 
-// A wrapper for building DataChunks in parallel
-class WindowDataChunk {
+// A wrapper for building ColumnDataCollections in parallel
+class WindowCollection {
 public:
-	// True if the vector data can just be copied to
-	static bool IsSimple(const Vector &v);
+	using ColumnDataCollectionPtr = unique_ptr<ColumnDataCollection>;
+	using ColumnDataCollectionSpec = pair<idx_t, optional_ptr<ColumnDataCollection>>;
 
-	static inline bool IsMaskAligned(idx_t begin, idx_t end, idx_t count) {
-		return ValidityMask::IsAligned(begin) && (ValidityMask::IsAligned(end) || (end == count));
-	}
+	WindowCollection(BufferManager &buffer_manager, idx_t count, const vector<LogicalType> &types);
 
-	explicit WindowDataChunk(DataChunk &chunk);
+	idx_t ColumnCount() const {
+		return types.size();
+	}
 
-	void Initialize(Allocator &allocator, const vector<LogicalType> &types, idx_t capacity);
+	idx_t size() const { // NOLINT
+		return count;
+	}
 
-	void Copy(DataChunk &src, idx_t begin);
+	const vector<LogicalType> &GetTypes() const {
+		return types;
+	}
 
-	//! The wrapped chunk
-	DataChunk &chunk;
+	//! 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);
+
+	//! The collection data. May be null if the column count is 0.
+	ColumnDataCollectionPtr inputs;
+	//! Global validity mask
+	atomic<bool> all_valid;
+	//! Optional validity mask for the entire collection
+	ValidityMask validity;
+
+	//! The collection columns
+	const vector<LogicalType> types;
+	//! The collection rows
+	const idx_t count;
 
 private:
-	//! True if the column is a scalar only value
-	vector<bool> is_simple;
-	//! Exclusive lock for each column
-	vector<mutex> locks;
+	//! Guard for range updates
+	mutex lock;
+	//! The paging buffer manager to use
+	BufferManager &buffer_manager;
+	//! The component column data collections
+	vector<ColumnDataCollectionPtr> collections;
+	//! The (sorted) collection ranges
+	using Range = pair<idx_t, idx_t>;
+	vector<Range> ranges;
+};
+
+class WindowBuilder {
+public:
+	explicit WindowBuilder(WindowCollection &collection);
+
+	//! Add a new chunk at the given index
+	void Sink(DataChunk &chunk, idx_t input_idx);
+
+	//! The collection we are helping to build
+	WindowCollection &collection;
+	//! The thread's current input collection
+	using ColumnDataCollectionSpec = WindowCollection::ColumnDataCollectionSpec;
+	ColumnDataCollectionSpec sink;
+	//! The state used for appending to the collection
+	ColumnDataAppendState appender;
+	//! Are all the sunk rows valid?
+	bool all_valid = true;
+};
+
+class WindowCursor {
+public:
+	explicit WindowCursor(const WindowCollection &paged);
+
+	//! Is the scan in range?
+	inline bool RowIsVisible(idx_t row_idx) const {
+		return (row_idx < state.next_row_index && state.current_row_index <= row_idx);
+	}
+	//! The offset of the row in the given state
+	inline sel_t RowOffset(idx_t row_idx) const {
+		D_ASSERT(RowIsVisible(row_idx));
+		return UnsafeNumericCast<sel_t>(row_idx - state.current_row_index);
+	}
+	//! Scan the next chunk
+	inline bool Scan() {
+		return paged.inputs->Scan(state, chunk);
+	}
+	//! Seek to the given row
+	inline idx_t Seek(idx_t row_idx) {
+		if (!RowIsVisible(row_idx)) {
+			D_ASSERT(paged.inputs.get());
+			paged.inputs->Seek(row_idx, state, chunk);
+		}
+		return RowOffset(row_idx);
+	}
+	//! Check a collection cell for nullity
+	bool CellIsNull(idx_t col_idx, idx_t row_idx) {
+		D_ASSERT(chunk.ColumnCount() > col_idx);
+		auto index = Seek(row_idx);
+		auto &source = chunk.data[col_idx];
+		return FlatVector::IsNull(source, index);
+	}
+	//! Read a typed cell
+	template <typename T>
+	T GetCell(idx_t col_idx, idx_t row_idx) {
+		D_ASSERT(chunk.ColumnCount() > col_idx);
+		auto index = Seek(row_idx);
+		auto &source = chunk.data[col_idx];
+		const auto data = FlatVector::GetData<T>(source);
+		return data[index];
+	}
+	//! Copy a single value
+	void CopyCell(idx_t col_idx, idx_t row_idx, Vector &target, idx_t target_offset) {
+		D_ASSERT(chunk.ColumnCount() > col_idx);
+		auto index = Seek(row_idx);
+		auto &source = chunk.data[col_idx];
+		VectorOperations::Copy(source, target, index + 1, index, target_offset);
+	}
+
+	//! The pageable data
+	const WindowCollection &paged;
+	//! The state used for reading the collection
+	ColumnDataScanState state;
+	//! The data chunk read into
+	DataChunk chunk;
 };
 
 struct WindowInputExpression {
@@ -98,35 +196,6 @@ struct WindowInputExpression {
 	DataChunk chunk;
 };
 
-struct WindowInputColumn {
-	WindowInputColumn(optional_ptr<Expression> expr_p, ClientContext &context, idx_t count);
-
-	void Copy(DataChunk &input_chunk, idx_t input_idx);
-
-	inline bool CellIsNull(idx_t i) const {
-		D_ASSERT(!target.data.empty());
-		D_ASSERT(i < count);
-		return FlatVector::IsNull((target.data[0]), scalar ? 0 : i);
-	}
-
-	template <typename T>
-	inline T GetCell(idx_t i) const {
-		D_ASSERT(!target.data.empty());
-		D_ASSERT(i < count);
-		const auto data = FlatVector::GetData<T>(target.data[0]);
-		return data[scalar ? 0 : i];
-	}
-
-	optional_ptr<Expression> expr;
-	PhysicalType ptype;
-	const bool scalar;
-	const idx_t count;
-
-private:
-	DataChunk target;
-	WindowDataChunk wtarget;
-};
-
 //	Column indexes of the bounds chunk
 enum WindowBounds : uint8_t { PARTITION_BEGIN, PARTITION_END, PEER_BEGIN, PEER_END, WINDOW_BEGIN, WINDOW_END };
 
@@ -163,7 +232,8 @@ class WindowExecutorGlobalState : public WindowExecutorState {
 	vector<LogicalType> arg_types;
 
 	// evaluate RANGE expressions, if needed
-	WindowInputColumn range;
+	optional_ptr<Expression> range_expr;
+	unique_ptr<WindowCollection> range;
 };
 
 class WindowExecutorLocalState : public WindowExecutorState {
@@ -171,6 +241,7 @@ class WindowExecutorLocalState : public WindowExecutorState {
 	explicit WindowExecutorLocalState(const WindowExecutorGlobalState &gstate);
 
 	void Sink(WindowExecutorGlobalState &gstate, DataChunk &input_chunk, idx_t input_idx);
+	virtual void Finalize(WindowExecutorGlobalState &gstate);
 
 	// Argument evaluation
 	ExpressionExecutor payload_executor;
@@ -179,6 +250,10 @@ class WindowExecutorLocalState : public WindowExecutorState {
 	//! 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 {
@@ -194,8 +269,7 @@ class WindowExecutor {
 	virtual void Sink(DataChunk &input_chunk, const idx_t input_idx, const idx_t total_count,
 	                  WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const;
 
-	virtual void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const {
-	}
+	virtual void Finalize(WindowExecutorGlobalState &gstate, WindowExecutorLocalState &lstate) const;
 
 	void Evaluate(idx_t row_idx, DataChunk &input_chunk, Vector &result, WindowExecutorLocalState &lstate,
 	              WindowExecutorGlobalState &gstate) const;
diff --git a/src/include/duckdb/execution/window_segment_tree.hpp b/src/include/duckdb/execution/window_segment_tree.hpp
index d70b478981a0..b9b41954f7a8 100644
--- a/src/include/duckdb/execution/window_segment_tree.hpp
+++ b/src/include/duckdb/execution/window_segment_tree.hpp
@@ -46,7 +46,7 @@ class WindowAggregator {
 	virtual ~WindowAggregator();
 
 	//	Threading states
-	virtual unique_ptr<WindowAggregatorState> GetGlobalState(idx_t group_count,
+	virtual unique_ptr<WindowAggregatorState> GetGlobalState(ClientContext &context, idx_t group_count,
 	                                                         const ValidityMask &partition_mask) const;
 	virtual unique_ptr<WindowAggregatorState> GetLocalState(const WindowAggregatorState &gstate) const = 0;
 
@@ -90,7 +90,7 @@ class WindowConstantAggregator : public WindowAggregator {
 	~WindowConstantAggregator() override {
 	}
 
-	unique_ptr<WindowAggregatorState> GetGlobalState(idx_t group_count,
+	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;
@@ -107,7 +107,7 @@ class WindowCustomAggregator : public WindowAggregator {
 	                       const LogicalType &result_type_p, const WindowExcludeMode exclude_mode);
 	~WindowCustomAggregator() override;
 
-	unique_ptr<WindowAggregatorState> GetGlobalState(idx_t group_count,
+	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;
 
@@ -122,7 +122,7 @@ class WindowSegmentTree : public WindowAggregator {
 	WindowSegmentTree(AggregateObject aggr, const vector<LogicalType> &arg_types_p, const LogicalType &result_type_p,
 	                  WindowAggregationMode mode_p, const WindowExcludeMode exclude_mode);
 
-	unique_ptr<WindowAggregatorState> GetGlobalState(idx_t group_count,
+	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;
@@ -147,7 +147,7 @@ class WindowDistinctAggregator : public WindowAggregator {
 	                         ClientContext &context);
 
 	//	Build
-	unique_ptr<WindowAggregatorState> GetGlobalState(idx_t group_count,
+	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;
diff --git a/src/include/duckdb/function/aggregate_function.hpp b/src/include/duckdb/function/aggregate_function.hpp
index 4d4ef63ff0d0..ab674cdb4987 100644
--- a/src/include/duckdb/function/aggregate_function.hpp
+++ b/src/include/duckdb/function/aggregate_function.hpp
@@ -16,6 +16,8 @@
 
 namespace duckdb {
 
+class BufferManager;
+
 //! A half-open range of frame boundary values _relative to the current row_
 //! This is why they are signed values.
 struct FrameDelta {
@@ -29,14 +31,18 @@ struct FrameDelta {
 using FrameStats = array<FrameDelta, 2>;
 
 //! The partition data for custom window functions
+//! Note that if the inputs is nullptr then the column count is 0,
+//! but the row count will still be valid
+class ColumnDataCollection;
 struct WindowPartitionInput {
-	WindowPartitionInput(const Vector inputs[], idx_t input_count, idx_t count, const ValidityMask &filter_mask,
-	                     const FrameStats &stats)
-	    : inputs(inputs), input_count(input_count), count(count), filter_mask(filter_mask), stats(stats) {
+	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) {
 	}
-	const Vector *inputs;
-	idx_t input_count;
+	ClientContext &context;
+	const ColumnDataCollection *inputs;
 	idx_t count;
+	bool all_valid;
 	const ValidityMask &filter_mask;
 	const FrameStats stats;
 };
@@ -265,16 +271,6 @@ class AggregateFunction : public BaseScalarFunction { // NOLINT: work-around bug
 		AggregateExecutor::UnaryUpdate<STATE, INPUT_TYPE, OP>(inputs[0], aggr_input_data, state, count);
 	}
 
-	template <class STATE, class INPUT_TYPE, class RESULT_TYPE, class OP>
-	static void UnaryWindow(AggregateInputData &aggr_input_data, const WindowPartitionInput &partition,
-	                        const_data_ptr_t g_state, data_ptr_t l_state, const SubFrames &subframes, Vector &result,
-	                        idx_t rid) {
-
-		D_ASSERT(partition.input_count == 1);
-		AggregateExecutor::UnaryWindow<STATE, INPUT_TYPE, RESULT_TYPE, OP>(
-		    partition.inputs[0], partition.filter_mask, aggr_input_data, l_state, subframes, result, rid, g_state);
-	}
-
 	template <class STATE, class A_TYPE, class B_TYPE, class OP>
 	static void BinaryScatterUpdate(Vector inputs[], AggregateInputData &aggr_input_data, idx_t input_count,
 	                                Vector &states, idx_t count) {


Test patch:
diff --git a/test/sql/window/test_custom_spooling.test_slow b/test/sql/window/test_custom_spooling.test_slow
new file mode 100644
index 000000000000..f6d2fd9e0fd9
--- /dev/null
+++ b/test/sql/window/test_custom_spooling.test_slow
@@ -0,0 +1,37 @@
+# name: test/sql/window/test_custom_spooling.test_slow
+# description: Validate spooling of custom window functions.
+# group: [window]
+
+require tpch
+
+statement ok
+call dbgen(sf=0.1);
+
+statement ok
+PRAGMA temp_directory='__TEST_DIR__/window_spooling'
+
+# Force paging
+statement ok
+PRAGMA memory_limit='50MB'
+
+statement ok
+PRAGMA enable_verification
+
+# MODE is very unstable on this data
+query IIII
+SELECT SUM(s), SUM(q), SUM(m), ROUND(SUM(d), -4)
+FROM (
+	SELECT
+		SUM(l_extendedprice) OVER w AS s,
+		QUANTILE(l_extendedprice, 0.5) OVER w AS q,
+		MAD(l_extendedprice) OVER w AS m,
+		MODE(l_quantity) OVER w AS d,
+	FROM lineitem
+	WINDOW w AS (
+		PARTITION BY (l_linenumber = 1) 
+		ORDER BY l_orderkey, l_linenumber 
+		ROWS BETWEEN 20 PRECEDING AND 20 FOLLOWING
+	)
+) t;
+----
+886221884732.62	20709502135.61	9898471727.97	15330000
diff --git a/test/sql/window/test_window_distinct.test_slow b/test/sql/window/test_window_distinct.test_slow
index a7e17c4b255d..19cb7a1540f3 100644
--- a/test/sql/window/test_window_distinct.test_slow
+++ b/test/sql/window/test_window_distinct.test_slow
@@ -32,6 +32,8 @@ PRAGMA disable_verification
 statement ok
 PRAGMA threads=4
 
+# The actual memory consumption of this query is 1.5GB
+# So this is a serious level of stress.
 statement ok
 PRAGMA memory_limit='500MB'