alopex_dataframe/physical/
plan.rs

1use std::path::PathBuf;
2
3use crate::lazy::{LogicalPlan, ProjectionKind};
4use crate::ops::{FillNull, JoinKeys, JoinType, SortOptions};
5use crate::physical::budget::StreamOptions;
6use crate::{DataFrame, Expr, Result};
7use arrow::datatypes::SchemaRef;
8
9/// Plan/source subject named by a streaming eligibility result.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum PlanSubject {
12    /// In-memory materialized DataFrame source.
13    DataFrameScan,
14    /// CSV source.
15    CsvScan,
16    /// Parquet source.
17    ParquetScan,
18    /// Provisioned range-addressable Alopex V08 columnar segment source.
19    ColumnarSegmentScan,
20    /// Strict vertical concatenation.
21    Concat,
22    /// Projection or with-columns operator.
23    Projection,
24    /// Filter operator.
25    Filter,
26    /// Aggregate operator.
27    Aggregate,
28    /// Join operator.
29    Join,
30    /// Sort operator.
31    Sort,
32    /// Slice operator.
33    Slice,
34    /// Unique operator.
35    Unique,
36    /// Fill-null operator.
37    FillNull,
38    /// Drop-nulls operator.
39    DropNulls,
40    /// Null-count operator.
41    NullCount,
42    /// Explode operator.
43    Explode,
44    /// Implode operator.
45    Implode,
46}
47
48impl PlanSubject {
49    /// Stable diagnostic name.
50    pub const fn as_str(self) -> &'static str {
51        match self {
52            Self::DataFrameScan => "dataframe_scan",
53            Self::CsvScan => "csv_scan",
54            Self::ParquetScan => "parquet_scan",
55            Self::ColumnarSegmentScan => "columnar_segment_scan",
56            Self::Concat => "concat",
57            Self::Projection => "projection",
58            Self::Filter => "filter",
59            Self::Aggregate => "aggregate",
60            Self::Join => "join",
61            Self::Sort => "sort",
62            Self::Slice => "slice",
63            Self::Unique => "unique",
64            Self::FillNull => "fill_null",
65            Self::DropNulls => "drop_nulls",
66            Self::NullCount => "null_count",
67            Self::Explode => "explode",
68            Self::Implode => "implode",
69        }
70    }
71}
72
73/// Stable reason for a streaming eligibility outcome.
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub enum StreamingReason {
76    /// The v0.8 bounded source reader is not registered yet.
77    SourceReaderUnavailable,
78    /// The legacy source is already materialized and cannot be used as a streaming source.
79    LegacyMaterializedSource,
80    /// The operator needs global state and has no bounded streaming algorithm.
81    GlobalState,
82    /// Tail requires knowledge of the complete input.
83    ReverseSlice,
84    /// The requested bound cannot support an approved algorithm.
85    InsufficientResourceBound,
86    /// The source is available but its required batch operator is not installed yet.
87    OperatorNotInstalled,
88}
89
90impl StreamingReason {
91    /// Stable diagnostic name.
92    pub const fn as_str(self) -> &'static str {
93        match self {
94            Self::SourceReaderUnavailable => "source_reader_unavailable",
95            Self::LegacyMaterializedSource => "legacy_materialized_source",
96            Self::GlobalState => "global_state",
97            Self::ReverseSlice => "reverse_slice_requires_materialization",
98            Self::InsufficientResourceBound => "insufficient_resource_bound",
99            Self::OperatorNotInstalled => "operator_not_installed",
100        }
101    }
102}
103
104/// Source limitation visible before source I/O.
105#[derive(Debug, Clone, PartialEq, Eq)]
106pub struct SourceLimit {
107    /// Source that owns the limitation.
108    pub source: PlanSubject,
109    /// Stable limitation code.
110    pub code: &'static str,
111    /// Short public explanation.
112    pub description: &'static str,
113}
114
115/// Eligibility outcome returned before opening a source.
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub enum StreamingEligibility {
118    /// The compiler can construct a bounded source and per-batch pipeline.
119    Supported { source_limits: Vec<SourceLimit> },
120    /// No streaming implementation exists for this source or operator.
121    Unsupported {
122        /// The rejected source or operator.
123        subject: PlanSubject,
124        /// Stable reason code.
125        reason: StreamingReason,
126    },
127    /// The operation is known but needs an explicit bounded materialization algorithm.
128    RequiresMaterialization {
129        /// The rejected operator.
130        subject: PlanSubject,
131        /// Whether a bounded materialization implementation is available.
132        bounded_possible: bool,
133    },
134}
135
136impl StreamingEligibility {
137    /// Convert a non-supported result to its public preflight error.
138    pub fn into_result(self) -> Result<Vec<SourceLimit>> {
139        match self {
140            Self::Supported { source_limits } => Ok(source_limits),
141            Self::Unsupported { subject, reason } => Err(
142                crate::DataFrameError::streaming_unsupported(subject.as_str(), reason.as_str()),
143            ),
144            Self::RequiresMaterialization {
145                subject,
146                bounded_possible,
147            } => Err(crate::DataFrameError::streaming_requires_materialization(
148                subject.as_str(),
149                if bounded_possible {
150                    "bounded_algorithm_required"
151                } else {
152                    "no_bounded_algorithm"
153                },
154            )),
155        }
156    }
157}
158
159/// A physical source and only the operators approved for batch-at-a-time execution.
160#[derive(Debug, Clone)]
161pub struct StreamingPhysicalPlan {
162    /// Source tree to open through bounded source factories.
163    pub source: StreamingSource,
164    /// Operators in source-to-consumer order.
165    pub operators: Vec<StreamingOperator>,
166    /// Source limitations determined before opening the source.
167    pub source_limits: Vec<SourceLimit>,
168}
169
170/// Streaming source tree preserving concat input order and child operators.
171#[derive(Debug, Clone)]
172pub enum StreamingSource {
173    /// A single scan source with no nested stream boundary.
174    Scan(ScanSource),
175    /// Strict vertical concat of independently compiled child streams.
176    Concat {
177        /// Children in declared output order.
178        inputs: Vec<StreamingPhysicalPlan>,
179        /// Validated common schema, or deferred bounded source preflight.
180        schema: Option<SchemaRef>,
181    },
182}
183
184/// Per-batch operator selected by the streaming compiler.
185#[derive(Debug, Clone)]
186pub enum StreamingOperator {
187    /// Projection/select or with-columns applied to one batch.
188    Projection {
189        /// Expressions in declaration order.
190        exprs: Vec<Expr>,
191        /// Projection behavior.
192        kind: ProjectionKind,
193    },
194    /// Predicate applied to one batch.
195    Filter {
196        /// Predicate expression.
197        predicate: Expr,
198    },
199    /// Forward-only head/slice operation.
200    ForwardSlice {
201        /// Rows to skip from the stream start.
202        offset: usize,
203        /// Maximum rows to publish.
204        len: usize,
205    },
206}
207
208/// Source for a physical scan operator.
209#[derive(Debug, Clone)]
210pub enum ScanSource {
211    /// In-memory scan of a `DataFrame`.
212    DataFrame(DataFrame),
213    /// CSV file scan with optional predicate/projection pushdown.
214    Csv {
215        path: PathBuf,
216        predicate: Option<Expr>,
217        projection: Option<Vec<String>>,
218    },
219    /// Parquet file scan with optional predicate/projection pushdown.
220    Parquet {
221        path: PathBuf,
222        predicate: Option<Expr>,
223        projection: Option<Vec<String>>,
224    },
225}
226
227/// Physical execution plan produced from a `LogicalPlan`.
228#[derive(Debug, Clone)]
229pub enum PhysicalPlan {
230    /// Scan operator.
231    ScanExec { source: ScanSource },
232    /// Strict vertical concat execution preserving input order.
233    ConcatExec {
234        /// Inputs in declared output order.
235        inputs: Vec<PhysicalPlan>,
236        /// Validated common schema, or deferred bounded source preflight.
237        schema: Option<SchemaRef>,
238    },
239    /// Projection operator.
240    ProjectionExec {
241        input: Box<PhysicalPlan>,
242        exprs: Vec<Expr>,
243        kind: ProjectionKind,
244    },
245    /// Filter operator.
246    FilterExec {
247        input: Box<PhysicalPlan>,
248        predicate: Expr,
249    },
250    /// Aggregate operator.
251    AggregateExec {
252        input: Box<PhysicalPlan>,
253        group_by: Vec<Expr>,
254        aggs: Vec<Expr>,
255    },
256    /// Join operator.
257    JoinExec {
258        left: Box<PhysicalPlan>,
259        right: Box<PhysicalPlan>,
260        keys: JoinKeys,
261        how: JoinType,
262    },
263    /// Sort operator.
264    SortExec {
265        input: Box<PhysicalPlan>,
266        options: SortOptions,
267    },
268    /// Slice operator (head/tail).
269    SliceExec {
270        input: Box<PhysicalPlan>,
271        offset: usize,
272        len: usize,
273        from_end: bool,
274    },
275    /// Unique operator.
276    UniqueExec {
277        input: Box<PhysicalPlan>,
278        subset: Option<Vec<String>>,
279    },
280    /// Fill-null operator.
281    FillNullExec {
282        input: Box<PhysicalPlan>,
283        fill: FillNull,
284    },
285    /// Drop-nulls operator.
286    DropNullsExec {
287        input: Box<PhysicalPlan>,
288        subset: Option<Vec<String>>,
289    },
290    /// Null-count operator.
291    NullCountExec { input: Box<PhysicalPlan> },
292    /// Explode one list column.
293    ExplodeExec {
294        input: Box<PhysicalPlan>,
295        column: String,
296    },
297    /// Implode columns into one row of list columns.
298    ImplodeExec { input: Box<PhysicalPlan> },
299}
300
301/// Compile a `LogicalPlan` into a `PhysicalPlan`.
302pub fn compile(logical: &LogicalPlan) -> Result<PhysicalPlan> {
303    let plan = match logical {
304        LogicalPlan::DataFrameScan { df } => PhysicalPlan::ScanExec {
305            source: ScanSource::DataFrame(df.clone()),
306        },
307        LogicalPlan::CsvScan {
308            path,
309            predicate,
310            projection,
311        } => PhysicalPlan::ScanExec {
312            source: ScanSource::Csv {
313                path: path.clone(),
314                predicate: predicate.clone(),
315                projection: projection.clone(),
316            },
317        },
318        LogicalPlan::ParquetScan {
319            path,
320            predicate,
321            projection,
322        } => PhysicalPlan::ScanExec {
323            source: ScanSource::Parquet {
324                path: path.clone(),
325                predicate: predicate.clone(),
326                projection: projection.clone(),
327            },
328        },
329        LogicalPlan::Concat { inputs, schema } => PhysicalPlan::ConcatExec {
330            inputs: inputs.iter().map(compile).collect::<Result<Vec<_>>>()?,
331            schema: schema.clone(),
332        },
333        LogicalPlan::Projection { input, exprs, kind } => PhysicalPlan::ProjectionExec {
334            input: Box::new(compile(input)?),
335            exprs: exprs.clone(),
336            kind: kind.clone(),
337        },
338        LogicalPlan::Filter { input, predicate } => PhysicalPlan::FilterExec {
339            input: Box::new(compile(input)?),
340            predicate: predicate.clone(),
341        },
342        LogicalPlan::Aggregate {
343            input,
344            group_by,
345            aggs,
346        } => PhysicalPlan::AggregateExec {
347            input: Box::new(compile(input)?),
348            group_by: group_by.clone(),
349            aggs: aggs.clone(),
350        },
351        LogicalPlan::Join {
352            left,
353            right,
354            keys,
355            how,
356        } => PhysicalPlan::JoinExec {
357            left: Box::new(compile(left)?),
358            right: Box::new(compile(right)?),
359            keys: keys.clone(),
360            how: *how,
361        },
362        LogicalPlan::Sort { input, options } => PhysicalPlan::SortExec {
363            input: Box::new(compile(input)?),
364            options: options.clone(),
365        },
366        LogicalPlan::Slice {
367            input,
368            offset,
369            len,
370            from_end,
371        } => PhysicalPlan::SliceExec {
372            input: Box::new(compile(input)?),
373            offset: *offset,
374            len: *len,
375            from_end: *from_end,
376        },
377        LogicalPlan::Unique { input, subset } => PhysicalPlan::UniqueExec {
378            input: Box::new(compile(input)?),
379            subset: subset.clone(),
380        },
381        LogicalPlan::FillNull { input, fill } => PhysicalPlan::FillNullExec {
382            input: Box::new(compile(input)?),
383            fill: fill.clone(),
384        },
385        LogicalPlan::DropNulls { input, subset } => PhysicalPlan::DropNullsExec {
386            input: Box::new(compile(input)?),
387            subset: subset.clone(),
388        },
389        LogicalPlan::NullCount { input } => PhysicalPlan::NullCountExec {
390            input: Box::new(compile(input)?),
391        },
392        LogicalPlan::Explode { input, column } => PhysicalPlan::ExplodeExec {
393            input: Box::new(compile(input)?),
394            column: column.clone(),
395        },
396        LogicalPlan::Implode { input } => PhysicalPlan::ImplodeExec {
397            input: Box::new(compile(input)?),
398        },
399    };
400
401    Ok(plan)
402}
403
404/// Analyze a plan before source I/O. Global nodes win over leaf-source results so callers never
405/// open a source merely to discover that a global operator cannot stream.
406pub fn analyze_streaming(plan: &PhysicalPlan, options: StreamOptions) -> StreamingEligibility {
407    if options.memory_limit_bytes == 0 {
408        return StreamingEligibility::Unsupported {
409            subject: PlanSubject::DataFrameScan,
410            reason: StreamingReason::InsufficientResourceBound,
411        };
412    }
413
414    match plan {
415        PhysicalPlan::ScanExec { source } => match source {
416            ScanSource::DataFrame(_) => StreamingEligibility::Unsupported {
417                subject: PlanSubject::DataFrameScan,
418                reason: StreamingReason::LegacyMaterializedSource,
419            },
420            ScanSource::Csv { predicate, .. } => {
421                if predicate
422                    .as_ref()
423                    .is_some_and(|predicate| !expression_is_streamable(predicate, false))
424                {
425                    StreamingEligibility::Unsupported {
426                        subject: PlanSubject::Filter,
427                        reason: StreamingReason::OperatorNotInstalled,
428                    }
429                } else {
430                    StreamingEligibility::Supported {
431                        source_limits: vec![SourceLimit {
432                            source: PlanSubject::CsvScan,
433                            code: "stable_input_order",
434                            description: "CSV streaming preserves physical input record order",
435                        }],
436                    }
437                }
438            }
439            ScanSource::Parquet { predicate, .. } => {
440                if predicate
441                    .as_ref()
442                    .is_some_and(|predicate| !expression_is_streamable(predicate, false))
443                {
444                    StreamingEligibility::Unsupported {
445                        subject: PlanSubject::Filter,
446                        reason: StreamingReason::OperatorNotInstalled,
447                    }
448                } else {
449                    StreamingEligibility::Supported {
450                        source_limits: vec![
451                            SourceLimit {
452                                source: PlanSubject::ParquetScan,
453                                code: "stable_row_group_order",
454                                description: "Parquet streaming preserves selected row-group and row order",
455                            },
456                            SourceLimit {
457                                source: PlanSubject::ParquetScan,
458                                code: "row_group_must_fit_resource_bound",
459                                description: "A selected row group whose declared upper bound exceeds the resource limit is rejected before page decode",
460                            },
461                        ],
462                    }
463                }
464            }
465        },
466        PhysicalPlan::ConcatExec { inputs, .. } => {
467            let mut source_limits = Vec::new();
468            for input in inputs {
469                match analyze_streaming(input, options) {
470                    StreamingEligibility::Supported {
471                        source_limits: input_limits,
472                    } => source_limits.extend(input_limits),
473                    outcome => return outcome,
474                }
475            }
476            StreamingEligibility::Supported { source_limits }
477        }
478        PhysicalPlan::ProjectionExec { input, exprs, .. } => {
479            if !exprs
480                .iter()
481                .all(|expression| expression_is_streamable(expression, true))
482            {
483                return StreamingEligibility::Unsupported {
484                    subject: PlanSubject::Projection,
485                    reason: StreamingReason::OperatorNotInstalled,
486                };
487            }
488            analyze_streaming(input, options)
489        }
490        PhysicalPlan::FilterExec { input, predicate } => {
491            if !expression_is_streamable(predicate, false) {
492                return StreamingEligibility::Unsupported {
493                    subject: PlanSubject::Filter,
494                    reason: StreamingReason::OperatorNotInstalled,
495                };
496            }
497            analyze_streaming(input, options)
498        }
499        PhysicalPlan::SliceExec {
500            input,
501            from_end: false,
502            ..
503        } => analyze_streaming(input, options),
504        PhysicalPlan::SliceExec { from_end: true, .. } => {
505            StreamingEligibility::RequiresMaterialization {
506                subject: PlanSubject::Slice,
507                bounded_possible: false,
508            }
509        }
510        PhysicalPlan::AggregateExec { .. } => StreamingEligibility::RequiresMaterialization {
511            subject: PlanSubject::Aggregate,
512            bounded_possible: false,
513        },
514        PhysicalPlan::JoinExec { .. } => StreamingEligibility::RequiresMaterialization {
515            subject: PlanSubject::Join,
516            bounded_possible: false,
517        },
518        PhysicalPlan::SortExec { .. } => StreamingEligibility::RequiresMaterialization {
519            subject: PlanSubject::Sort,
520            bounded_possible: false,
521        },
522        PhysicalPlan::UniqueExec { .. } => StreamingEligibility::RequiresMaterialization {
523            subject: PlanSubject::Unique,
524            bounded_possible: false,
525        },
526        PhysicalPlan::FillNullExec { .. } => StreamingEligibility::Unsupported {
527            subject: PlanSubject::FillNull,
528            reason: StreamingReason::GlobalState,
529        },
530        PhysicalPlan::DropNullsExec { .. } => StreamingEligibility::Unsupported {
531            subject: PlanSubject::DropNulls,
532            reason: StreamingReason::GlobalState,
533        },
534        PhysicalPlan::NullCountExec { .. } => StreamingEligibility::RequiresMaterialization {
535            subject: PlanSubject::NullCount,
536            bounded_possible: false,
537        },
538        PhysicalPlan::ExplodeExec { .. } => StreamingEligibility::Unsupported {
539            subject: PlanSubject::Explode,
540            reason: StreamingReason::GlobalState,
541        },
542        PhysicalPlan::ImplodeExec { .. } => StreamingEligibility::RequiresMaterialization {
543            subject: PlanSubject::Implode,
544            bounded_possible: false,
545        },
546    }
547}
548
549/// Return whether a v0.8 bounded evaluator has an allocation upper bound for this expression.
550/// Namespace functions remain eager-only until each one has an independently verified bound.
551fn expression_is_streamable(expression: &Expr, allow_wildcard: bool) -> bool {
552    match expression {
553        Expr::Column(_) | Expr::Literal(_) => true,
554        Expr::Alias { expr, .. } | Expr::UnaryOp { expr, .. } => {
555            expression_is_streamable(expr, allow_wildcard)
556        }
557        Expr::BinaryOp { left, right, .. } => {
558            expression_is_streamable(left, allow_wildcard)
559                && expression_is_streamable(right, allow_wildcard)
560        }
561        Expr::ConcatStr { inputs, .. } => {
562            inputs.len() >= 2
563                && inputs
564                    .iter()
565                    .all(|input| expression_is_streamable(input, allow_wildcard))
566        }
567        Expr::Wildcard => allow_wildcard,
568        Expr::Agg { .. } | Expr::Function { .. } => false,
569    }
570}
571
572/// Compile only a preflight-supported plan into a source plus batch-at-a-time operators.
573pub fn compile_streaming(
574    plan: &PhysicalPlan,
575    options: StreamOptions,
576) -> Result<StreamingPhysicalPlan> {
577    let source_limits = analyze_streaming(plan, options).into_result()?;
578    let mut operators = Vec::new();
579    let source = compile_streaming_inner(plan, &mut operators, options)?;
580    Ok(StreamingPhysicalPlan {
581        source,
582        operators,
583        source_limits,
584    })
585}
586
587fn compile_streaming_inner(
588    plan: &PhysicalPlan,
589    operators: &mut Vec<StreamingOperator>,
590    options: StreamOptions,
591) -> Result<StreamingSource> {
592    match plan {
593        PhysicalPlan::ScanExec { source } => {
594            let mut source = source.clone();
595            let predicate = match &mut source {
596                ScanSource::Csv { predicate, .. } | ScanSource::Parquet { predicate, .. } => {
597                    predicate.take()
598                }
599                ScanSource::DataFrame(_) => None,
600            };
601            if let Some(predicate) = predicate {
602                operators.push(StreamingOperator::Filter { predicate });
603            }
604            Ok(StreamingSource::Scan(source))
605        }
606        PhysicalPlan::ConcatExec { inputs, schema } => Ok(StreamingSource::Concat {
607            inputs: inputs
608                .iter()
609                .map(|input| compile_streaming(input, options))
610                .collect::<Result<Vec<_>>>()?,
611            schema: schema.clone(),
612        }),
613        PhysicalPlan::ProjectionExec { input, exprs, kind } => {
614            let source = compile_streaming_inner(input, operators, options)?;
615            operators.push(StreamingOperator::Projection {
616                exprs: exprs.clone(),
617                kind: kind.clone(),
618            });
619            Ok(source)
620        }
621        PhysicalPlan::FilterExec { input, predicate } => {
622            let source = compile_streaming_inner(input, operators, options)?;
623            operators.push(StreamingOperator::Filter {
624                predicate: predicate.clone(),
625            });
626            Ok(source)
627        }
628        PhysicalPlan::SliceExec {
629            input,
630            offset,
631            len,
632            from_end: false,
633        } => {
634            let source = compile_streaming_inner(input, operators, options)?;
635            operators.push(StreamingOperator::ForwardSlice {
636                offset: *offset,
637                len: *len,
638            });
639            Ok(source)
640        }
641        _ => Err(crate::DataFrameError::streaming_requires_materialization(
642            "physical_plan",
643            "no_batch_at_a_time_compiler",
644        )),
645    }
646}
647
648#[cfg(test)]
649mod streaming_tests {
650    use std::num::NonZeroUsize;
651    use std::sync::Arc;
652
653    use arrow::datatypes::{DataType, Field, Schema};
654
655    use super::{
656        analyze_streaming, compile_streaming, PhysicalPlan, ScanSource, StreamingEligibility,
657        StreamingSource,
658    };
659    use crate::physical::budget::StreamOptions;
660
661    fn options() -> StreamOptions {
662        StreamOptions::new(
663            1024,
664            NonZeroUsize::new(1).unwrap(),
665            NonZeroUsize::new(1).unwrap(),
666        )
667    }
668
669    #[test]
670    fn global_operator_is_rejected_before_csv_source_opening() {
671        let plan = PhysicalPlan::AggregateExec {
672            input: Box::new(PhysicalPlan::ScanExec {
673                source: ScanSource::Csv {
674                    path: "must-not-open.csv".into(),
675                    predicate: None,
676                    projection: None,
677                },
678            }),
679            group_by: Vec::new(),
680            aggs: Vec::new(),
681        };
682
683        assert!(matches!(
684            analyze_streaming(&plan, options()),
685            StreamingEligibility::RequiresMaterialization { .. }
686        ));
687        assert!(compile_streaming(&plan, options()).is_err());
688    }
689
690    #[test]
691    fn csv_source_is_supported_after_its_bounded_factory_is_registered() {
692        let plan = PhysicalPlan::ScanExec {
693            source: ScanSource::Csv {
694                path: "does-not-exist.csv".into(),
695                predicate: None,
696                projection: None,
697            },
698        };
699
700        assert!(matches!(
701            analyze_streaming(&plan, options()),
702            StreamingEligibility::Supported { .. }
703        ));
704        assert!(compile_streaming(&plan, options()).is_ok());
705    }
706
707    #[test]
708    fn concat_streaming_compiler_keeps_child_sources_in_declared_order() {
709        let schema = Arc::new(Schema::new(vec![Field::new(
710            "value",
711            DataType::Int64,
712            true,
713        )]));
714        let scan = |path: &str| PhysicalPlan::ScanExec {
715            source: ScanSource::Csv {
716                path: path.into(),
717                predicate: None,
718                projection: None,
719            },
720        };
721        let plan = PhysicalPlan::ConcatExec {
722            inputs: vec![scan("first.csv"), scan("second.csv")],
723            schema: Some(schema),
724        };
725
726        let compiled = compile_streaming(&plan, options()).unwrap();
727        let StreamingSource::Concat { inputs, .. } = compiled.source else {
728            panic!("expected concat streaming source");
729        };
730        assert_eq!(inputs.len(), 2);
731        for (input, expected_path) in inputs.iter().zip(["first.csv", "second.csv"]) {
732            match &input.source {
733                StreamingSource::Scan(ScanSource::Csv { path, .. }) => {
734                    assert_eq!(path, std::path::Path::new(expected_path));
735                }
736                other => panic!("expected CSV child source, got {other:?}"),
737            }
738        }
739    }
740}
alopex_dataframe/physical/plan.rs

alopex_dataframe/physical/
plan.rs
