GitHub user Subodh-17 added a comment to the discussion: [RFC/Discussion] Partition-at-a-Time Streaming Evaluation in WindowAggExec
### Design Consultation: Partition-by-Partition Streaming for WindowAggExec Hello DataFusion maintainers and community, We have been investigating memory efficiency and time-to-first-batch latency in `WindowAggExec` for partitioned window queries (e.g. `OVER (PARTITION BY ...)` where the input physical plan already provides ordering on the partition keys). #### 1. Context & Motivation In DataFusion, queries with bounded window frames can slide incrementally via `BoundedWindowAggExec`. However, queries planned with `WindowAggExec` (such as unbounded window frames or whole-partition aggregations) currently accumulate the entire input stream into an in-memory `Vec<RecordBatch>` before evaluating window expressions and emitting results upon EOF. When queries contain many distinct partitions (e.g., millions of user sessions or sensor IDs) and the input stream is ordered by those partition keys, buffering the entire dataset introduces unnecessary peak memory accumulation and delays the emission of completed partition results. We have developed an isolated prototype of **partition-by-partition streaming** directly within `WindowAggExec` and would welcome maintainer guidance on whether this direction fits the project's roadmap. #### 2. Mechanism & Conservative Eligibility The prototype introduces incremental partition boundary detection: - **Eligibility Check:** Streaming is enabled only when the physical input plan's declared properties (`input.equivalence_properties()`) satisfy all `PARTITION BY` expressions, and all window expressions in the operator share identical `PARTITION BY` expressions. The operator relies on declared plan ordering and does not perform runtime sort validation. - **Incremental Emission:** Incoming batches are inspected for partition boundary transitions. Completed interior partitions are evaluated with `compute_window_aggregates()` and yielded downstream immediately without waiting for global EOF. - **Cross-Batch Partitions:** Partitions that span across batch boundaries are buffered in `active_partition_batches` until the partition boundary arrives or input EOF occurs. - **Conservative Fallback:** If the physical plan does not declare an ordering matching all partition keys, if partition expressions vary across window functions, or if the query is unpartitioned, the stream safely falls back to the existing bulk whole-stream accumulation path. #### 3. Verification & Semantic Equivalence The prototype is covered by 30 focused unit and regression tests in `windows::window_agg_exec::tests`: - **Semantic Equivalence:** Exact column-by-column, schema, and row count equality between the streaming path and the bulk fallback path on identical multi-batch datasets (`test_t8_streaming_bulk_equivalence`). - **Edge Cases Tested:** Null/repeated partition keys, dictionary-encoded keys, composite partition keys, rank ties, interspersed empty batches, and large multi-batch macro-partitions. - **Memory Lifecycle & Fusing:** Tested monotonic `MemoryReservation` growth during buffering, immediate quota shrinking upon downstream yield, clean `ResourcesExhausted` handling when memory pool limits are exceeded, and strict stream fusing to `None` across tested error routes (source errors, schema mismatch, and pool exhaustion). #### 4. Disclosed Architectural Boundaries To ensure clear alignment with DataFusion's design principles, we want to highlight three current engine boundaries: 1. **Logical Accounting vs. Physical RSS:** `MemoryReservation` tracks estimated batch buffer byte sizes against the `MemoryPool` (cooperative quota). Because partition slicing uses zero-copy slicing (`batch.slice()`), underlying parent Arrow buffers remain held in physical RAM until all referencing slices from that batch are dropped downstream. 2. **Concatenation Memory Overhead:** While `concat_batches()` executes (in the bulk fallback path or when assembling partitions that span multiple batches), the operator's existing `MemoryReservation` remains active and charged for the buffered input data. However, `concat_batches()` allocates new contiguous Arrow buffers for the combined output before the source batch slices are dropped. During this step, both the input slices and the new concatenated batch concurrently reside in heap memory, creating a workload-dependent transient allocation peak. This additional temporary allocation is not accounted for by an incremental reservation against the memory pool. 3. **No External Spilling & Single-Partition Limits:** Partition streaming reduces operator memory accumulation by emitting completed partitions as boundaries arrive rather than buffering the entire multi-partition stream. However, it does not guarantee a hard physical-RAM bound. A single partition whose size exceeds the available memory pool budget will fail with `ResourcesExhausted`, as out-of-core disk spilling is not implemented. #### 5. Questions for Maintainers: 1. **Operator Architecture:** Do maintainers prefer evolving `WindowAggExec` in place to support partition streaming with conservative bulk fallback, or would you favor introducing a distinct physical plan node (e.g. `PartitionedWindowAggExec`) or an optimizer rule? 2. **Arrow Memory Contract:** When slicing batches across partition boundaries, is proportional logical reservation over zero-copy slices acceptable to the engine, or would maintainers require explicit deep buffer copying (allocating fresh arrays) to guarantee the immediate physical release of parent batch allocations? 3. **Spilling Prerequisites:** Is partition-by-partition streaming valuable as a standalone in-memory streaming optimization, or would maintainers consider disk spilling a prerequisite before accepting changes in this area? 4. **GSoC 2027 Alignment:** Does the community consider partitioned window streaming an impactful topic for formal contribution / GSoC project proposals? We appreciate your feedback and architectural guidance! GitHub link: https://github.com/apache/datafusion/discussions/26425#discussioncomment-19035368 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
