Adding serde utils for Arrow Arrays

crates/core/Cargo.toml

@@ -37,3 +37,4 @@ half = "2.4.1"
 delegate = "0.12.0"
 ahash = "0.8.11"
 hashbrown = "0.14.5"
+flatbuffers = "24.3.25"

crates/core/src/context.rs

@@ -8,7 +8,7 @@ use datafusion::execution::{
 
 use crate::datasource::kafka::TopicReader;
 use crate::datastream::DataStream;
-use crate::physical_optimizer::CoaslesceBeforeStreamingAggregate;
+use crate::physical_optimizer::EnsureHashPartititionOnGroupByForStreamingAggregates;
 use crate::query_planner::StreamingQueryPlanner;
 use crate::utils::get_default_optimizer_rules;
 
@@ -41,7 +41,9 @@ impl Context {
             .with_runtime_env(runtime)
             .with_query_planner(Arc::new(StreamingQueryPlanner {}))
             .with_optimizer_rules(get_default_optimizer_rules())
-            .with_physical_optimizer_rule(Arc::new(CoaslesceBeforeStreamingAggregate::new()))
+            .with_physical_optimizer_rule(Arc::new(
+                EnsureHashPartititionOnGroupByForStreamingAggregates::new(),
+            ))
             .build();
 
         Ok(Self {

crates/core/src/logical_plan/mod.rs

@@ -45,7 +45,7 @@ impl StreamingLogicalPlanBuilder for LogicalPlanBuilder {
         let plan = self.plan().clone();
 
         Aggregate::try_new(Arc::new(plan.clone()), group_expr, aggr_expr)
-            .map(|new_aggr| {
+            .map(|new_aggr: Aggregate| {
                 LogicalPlan::Extension(Extension {
                     node: Arc::new(StreamingWindowPlanNode {
                         window_type: window,

crates/core/src/physical_optimizer/coalesce_before_streaming_window_aggregate.rs

@@ -10,15 +10,15 @@ use datafusion::error::Result;
 
 use crate::physical_plan::continuous::streaming_window::StreamingWindowExec;
 
-pub struct CoaslesceBeforeStreamingAggregate {}
+pub struct EnsureHashPartititionOnGroupByForStreamingAggregates {}
 
-impl Default for CoaslesceBeforeStreamingAggregate {
+impl Default for EnsureHashPartititionOnGroupByForStreamingAggregates {
     fn default() -> Self {
         Self::new()
     }
 }
 
-impl CoaslesceBeforeStreamingAggregate {
+impl EnsureHashPartititionOnGroupByForStreamingAggregates {
     #[allow(missing_docs)]
     pub fn new() -> Self {
         Self {}
@@ -29,7 +29,7 @@ impl CoaslesceBeforeStreamingAggregate {
 // Franz optimizer rule, added to ensure coalescing of partitions before a global aggregate
 // window. This rule may be removed once we have support for two stage partial and final
 // aggregates a la vanilla Datafusion.
-impl PhysicalOptimizerRule for CoaslesceBeforeStreamingAggregate {
+impl PhysicalOptimizerRule for EnsureHashPartititionOnGroupByForStreamingAggregates {
     fn optimize(
         &self,
         plan: Arc<dyn datafusion::physical_plan::ExecutionPlan>,

crates/core/src/physical_optimizer/mod.rs

@@ -1,3 +1,3 @@
 pub mod coalesce_before_streaming_window_aggregate;
 
-pub use coalesce_before_streaming_window_aggregate::CoaslesceBeforeStreamingAggregate;
+pub use coalesce_before_streaming_window_aggregate::EnsureHashPartititionOnGroupByForStreamingAggregates;

crates/core/src/utils/mod.rs

@@ -2,5 +2,6 @@
 pub mod arrow_helpers;
 mod default_optimizer_rules;
 pub mod row_encoder;
+pub mod serialization;
 
 pub use default_optimizer_rules::get_default_optimizer_rules;