laminar_sql/planner/streaming_optimizer/
mod.rs

1//! Physical optimizer rule for streaming plan validation.
2//!
3//! Detects pipeline-breaking operators (Sort, Final Aggregate) on unbounded
4//! inputs and rejects or warns at plan-creation time, before any execution
5//! begins.
6
7use std::fmt::Debug;
8use std::sync::Arc;
9
10use datafusion::physical_optimizer::PhysicalOptimizerRule;
11use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode};
12use datafusion::physical_plan::execution_plan::Boundedness;
13use datafusion::physical_plan::sorts::sort::SortExec;
14use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties};
15use datafusion_common::config::ConfigOptions;
16use datafusion_common::DataFusionError;
17
18/// How the validator handles streaming plan violations.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum StreamingValidatorMode {
21    /// Return an error, preventing plan execution. Default.
22    Reject,
23    /// Log a warning but allow execution.
24    Warn,
25    /// Disable validation entirely.
26    Off,
27}
28
29/// A streaming plan violation detected during validation.
30#[derive(Debug)]
31struct StreamingViolation {
32    operator: String,
33    reason: String,
34    plan_path: String,
35}
36
37/// Validates that a physical plan is safe for streaming execution.
38///
39/// Detects pipeline-breaking operators (Sort, Final Aggregate) on
40/// unbounded inputs and rejects or warns depending on configuration.
41#[derive(Debug)]
42pub struct StreamingPhysicalValidator {
43    mode: StreamingValidatorMode,
44}
45
46impl StreamingPhysicalValidator {
47    /// Creates a new validator with the given mode.
48    #[must_use]
49    pub fn new(mode: StreamingValidatorMode) -> Self {
50        Self { mode }
51    }
52}
53
54impl PhysicalOptimizerRule for StreamingPhysicalValidator {
55    fn optimize(
56        &self,
57        plan: Arc<dyn ExecutionPlan>,
58        _config: &ConfigOptions,
59    ) -> datafusion_common::Result<Arc<dyn ExecutionPlan>> {
60        if matches!(self.mode, StreamingValidatorMode::Off) {
61            return Ok(plan);
62        }
63
64        let violations = find_streaming_violations(&plan);
65        if violations.is_empty() {
66            return Ok(plan);
67        }
68
69        match self.mode {
70            StreamingValidatorMode::Reject => {
71                Err(DataFusionError::Plan(format_violations(&violations)))
72            }
73            StreamingValidatorMode::Warn => {
74                for v in &violations {
75                    tracing::warn!(
76                        operator = %v.operator,
77                        path = %v.plan_path,
78                        "Streaming plan violation: {}", v.reason
79                    );
80                }
81                Ok(plan)
82            }
83            StreamingValidatorMode::Off => unreachable!(),
84        }
85    }
86
87    #[allow(clippy::unnecessary_literal_bound)]
88    fn name(&self) -> &str {
89        "streaming_physical_validator"
90    }
91
92    fn schema_check(&self) -> bool {
93        true
94    }
95}
96
97fn find_streaming_violations(plan: &Arc<dyn ExecutionPlan>) -> Vec<StreamingViolation> {
98    let mut violations = Vec::new();
99    walk_plan(plan, &mut violations, "");
100    violations
101}
102
103fn walk_plan(plan: &Arc<dyn ExecutionPlan>, violations: &mut Vec<StreamingViolation>, path: &str) {
104    let name = plan.name();
105    let current_path = if path.is_empty() {
106        name.to_string()
107    } else {
108        format!("{path} -> {name}")
109    };
110
111    // Check 1: SortExec on unbounded input
112    if plan.as_any().downcast_ref::<SortExec>().is_some() && has_unbounded_child(plan) {
113        violations.push(StreamingViolation {
114            operator: name.to_string(),
115            reason: "Sort requires buffering all input; unbounded source will \
116                     buffer forever. Remove ORDER BY or add a window."
117                .to_string(),
118            plan_path: current_path.clone(),
119        });
120    }
121
122    // Check 2: Final AggregateExec on unbounded input
123    if let Some(agg) = plan.as_any().downcast_ref::<AggregateExec>() {
124        if matches!(
125            agg.mode(),
126            &AggregateMode::Final | &AggregateMode::FinalPartitioned
127        ) && has_unbounded_child(plan)
128        {
129            violations.push(StreamingViolation {
130                operator: name.to_string(),
131                reason: "Final aggregation on unbounded input will never emit \
132                         results. Use a window function (TUMBLE/HOP/SESSION) or \
133                         add an EMIT clause."
134                    .to_string(),
135                plan_path: current_path.clone(),
136            });
137        }
138    }
139
140    for child in plan.children() {
141        walk_plan(child, violations, &current_path);
142    }
143}
144
145fn has_unbounded_child(plan: &Arc<dyn ExecutionPlan>) -> bool {
146    plan.children()
147        .iter()
148        .any(|c| matches!(c.boundedness(), Boundedness::Unbounded { .. }))
149}
150
151fn format_violations(violations: &[StreamingViolation]) -> String {
152    use std::fmt::Write;
153
154    let mut msg = String::from("Streaming plan validation failed:\n");
155    for (i, v) in violations.iter().enumerate() {
156        let _ = writeln!(
157            msg,
158            "  {}. [{}] {} (at: {})",
159            i + 1,
160            v.operator,
161            v.reason,
162            v.plan_path
163        );
164    }
165    msg
166}
167
168#[cfg(test)]
169mod tests;
laminar_sql/planner/streaming_optimizer/mod.rs

laminar_sql/planner/streaming_optimizer/
mod.rs
