Add StreamingWindowExec to DataFusion physical plan to support aggregations over unbounded data #11366

Activity

1. ameyc

   added

   enhancementNew feature or requestNew feature or request

   on Jul 9, 2024
2. mustafasrepo commented on Jul 10, 2024

   

   mustafasrepo

   on Jul 10, 2024

   Contributor

   More actions

   Actually existing AggregateExec supports streaming. It has a member input_order_mode which is following enum:

   pub enum InputOrderMode {
       /// There is no partial permutation of the expressions satisfying the
       /// existing ordering.
       Linear,
       /// There is a partial permutation of the expressions satisfying the
       /// existing ordering. Indices describing the longest partial permutation
       /// are stored in the vector.
       PartiallySorted(Vec<usize>),
       /// There is a (full) permutation of the expressions satisfying the
       /// existing ordering.
       Sorted,
   }

   When mode is either Sorted or PartiallySorted (decided by planner according to ordering at the input) operation is streamable. However, to trigger this modes at least one of the group by expression should be ordered.

   👍React with 👍1Reacted by Amey Chaugule
3. ozankabak commented on Jul 10, 2024

   

   ozankabak

   on Jul 10, 2024

   Contributor

   More actions

   To add to the comment above -- BoundedWindowAggExec also supports streaming (when possible).

   👍React with 👍1Reacted by Amey Chaugule
4. ameyc commented on Jul 11, 2024

   

   ameyc

   on Jul 11, 2024

   ContributorAuthor

   More actions

   @mustafasrepo & @ozankabak thanks for the feedback. the target usecases we were going for are flink style workloads, with data read from kafka that is generally not be ordered and thus needs to be watermarked. we tried the vanilla aggregates and ran into PipelineBreaking panics.

   An example workload we're trying to compute is of the nature, lmk if this can already be expressed with current operators as is --

       let windowed_df = df
           .clone()
           .streaming_window(
               vec![],
               vec![
                   max(col("imu_measurement").field("gps").field("speed")),
                   min(col("imu_measurement").field("gps").field("altitude")),
                   count(col("imu_measurement")).alias("count"),
               ],
               Duration::from_millis(5_000), // 5 second window
               Some(Duration::from_millis(1_000)), // 1 second slide
           )
           .unwrap();
   
5. ozankabak commented on Jul 12, 2024

   

   ozankabak

   on Jul 12, 2024

   Contributor

   More actions

   To help as best as I can, let me first reiterate my understanding of your use case: You have a streaming source, which has some columns like speed and altitude, but your ts column may be out-of-order and is not monotonic. Or, maybe you don't have such a column at all. Still, you want to do streaming computations based on processing time on such data.

   In such a case, what you can do is to use a projection to add an order defining column based on processing time, and use BoundedWindowAggExec to do streaming/incremental windowing based on that. So basically two built-in operators compose together to give you what you want (or I should say, what I think you want 🙂 ). Let me know if this helps.
6. alamb commented on Aug 1, 2026

   

   alamb

   on Aug 1, 2026

   Contributor

   More actions

   Per the analysis above by @ozankabak and @mustafasrepo -- i think BoundedWindowAggExec satistifies this usecase so closing this ticket as completed. Please reopen it if that is not right.
7. alamb

   closed this as completedon Aug 1, 2026