Adding GroupedWindowAggStream

Cargo.toml

@@ -55,4 +55,4 @@ chrono = { version = "0.4.38", default-features = false }
 itertools = "0.13"
 
 [patch.crates-io]
-datafusion = { git = "https://github.com/probably-nothing-labs/arrow-datafusion" }
+datafusion = { git = "https://github.com/probably-nothing-labs/arrow-datafusion", rev="fc67038" }

clippy.toml

@@ -10,4 +10,4 @@ disallowed-types = [
 # Lowering the threshold to help prevent stack overflows (default is 16384)
 # See: https://rust-lang.github.io/rust-clippy/master/index.html#/large_futures
 future-size-threshold = 10000
-too-many-arguments-threshold = 10
+too-many-arguments-threshold = 15

crates/core/Cargo.toml

@@ -28,3 +28,5 @@ rocksdb = "0.22.0"
 bincode = "1.3.3"
 half = "2.4.1"
 delegate = "0.12.0"
+ahash = "0.8.11"
+hashbrown = "0.14.5"

crates/core/src/datastream.rs

@@ -132,14 +132,11 @@ impl DataStream {
         loop {
             match stream.next().await.transpose() {
                 Ok(Some(batch)) => {
-                    for i in 0..batch.num_rows() {
-                        let row = batch.slice(i, 1);
-                        println!(
-                            "{}",
-                            datafusion::common::arrow::util::pretty::pretty_format_batches(&[row])
-                                .unwrap()
-                        );
-                    }
+                    println!(
+                        "{}",
+                        datafusion::common::arrow::util::pretty::pretty_format_batches(&[batch])
+                            .unwrap()
+                    );
                 }
                 Ok(None) => {
                     log::warn!("No RecordBatch in stream");

crates/core/src/physical_optimizer/coalesce_before_streaming_window_aggregate.rs

@@ -8,7 +8,7 @@ use datafusion::physical_plan::ExecutionPlanProperties;
 use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion::error::Result;
 
-use crate::physical_plan::streaming_window::FranzStreamingWindowExec;
+use crate::physical_plan::continuous::streaming_window::FranzStreamingWindowExec;
 
 pub struct CoaslesceBeforeStreamingAggregate {}