alopex_dataframe/physical/
plan.rs

1use std::path::PathBuf;
2
3use crate::lazy::{LogicalPlan, ProjectionKind};
4use crate::ops::{FillNull, JoinKeys, JoinType, SortOptions};
5use crate::{DataFrame, Expr, Result};
6
7/// Source for a physical scan operator.
8#[derive(Debug, Clone)]
9pub enum ScanSource {
10    /// In-memory scan of a `DataFrame`.
11    DataFrame(DataFrame),
12    /// CSV file scan with optional predicate/projection pushdown.
13    Csv {
14        path: PathBuf,
15        predicate: Option<Expr>,
16        projection: Option<Vec<String>>,
17    },
18    /// Parquet file scan with optional predicate/projection pushdown.
19    Parquet {
20        path: PathBuf,
21        predicate: Option<Expr>,
22        projection: Option<Vec<String>>,
23    },
24}
25
26/// Physical execution plan produced from a `LogicalPlan`.
27#[derive(Debug, Clone)]
28pub enum PhysicalPlan {
29    /// Scan operator.
30    ScanExec { source: ScanSource },
31    /// Projection operator.
32    ProjectionExec {
33        input: Box<PhysicalPlan>,
34        exprs: Vec<Expr>,
35        kind: ProjectionKind,
36    },
37    /// Filter operator.
38    FilterExec {
39        input: Box<PhysicalPlan>,
40        predicate: Expr,
41    },
42    /// Aggregate operator.
43    AggregateExec {
44        input: Box<PhysicalPlan>,
45        group_by: Vec<Expr>,
46        aggs: Vec<Expr>,
47    },
48    /// Join operator.
49    JoinExec {
50        left: Box<PhysicalPlan>,
51        right: Box<PhysicalPlan>,
52        keys: JoinKeys,
53        how: JoinType,
54    },
55    /// Sort operator.
56    SortExec {
57        input: Box<PhysicalPlan>,
58        options: SortOptions,
59    },
60    /// Slice operator (head/tail).
61    SliceExec {
62        input: Box<PhysicalPlan>,
63        offset: usize,
64        len: usize,
65        from_end: bool,
66    },
67    /// Unique operator.
68    UniqueExec {
69        input: Box<PhysicalPlan>,
70        subset: Option<Vec<String>>,
71    },
72    /// Fill-null operator.
73    FillNullExec {
74        input: Box<PhysicalPlan>,
75        fill: FillNull,
76    },
77    /// Drop-nulls operator.
78    DropNullsExec {
79        input: Box<PhysicalPlan>,
80        subset: Option<Vec<String>>,
81    },
82    /// Null-count operator.
83    NullCountExec { input: Box<PhysicalPlan> },
84    /// Explode one list column.
85    ExplodeExec {
86        input: Box<PhysicalPlan>,
87        column: String,
88    },
89    /// Implode columns into one row of list columns.
90    ImplodeExec { input: Box<PhysicalPlan> },
91}
92
93/// Compile a `LogicalPlan` into a `PhysicalPlan`.
94pub fn compile(logical: &LogicalPlan) -> Result<PhysicalPlan> {
95    let plan = match logical {
96        LogicalPlan::DataFrameScan { df } => PhysicalPlan::ScanExec {
97            source: ScanSource::DataFrame(df.clone()),
98        },
99        LogicalPlan::CsvScan {
100            path,
101            predicate,
102            projection,
103        } => PhysicalPlan::ScanExec {
104            source: ScanSource::Csv {
105                path: path.clone(),
106                predicate: predicate.clone(),
107                projection: projection.clone(),
108            },
109        },
110        LogicalPlan::ParquetScan {
111            path,
112            predicate,
113            projection,
114        } => PhysicalPlan::ScanExec {
115            source: ScanSource::Parquet {
116                path: path.clone(),
117                predicate: predicate.clone(),
118                projection: projection.clone(),
119            },
120        },
121        LogicalPlan::Projection { input, exprs, kind } => PhysicalPlan::ProjectionExec {
122            input: Box::new(compile(input)?),
123            exprs: exprs.clone(),
124            kind: kind.clone(),
125        },
126        LogicalPlan::Filter { input, predicate } => PhysicalPlan::FilterExec {
127            input: Box::new(compile(input)?),
128            predicate: predicate.clone(),
129        },
130        LogicalPlan::Aggregate {
131            input,
132            group_by,
133            aggs,
134        } => PhysicalPlan::AggregateExec {
135            input: Box::new(compile(input)?),
136            group_by: group_by.clone(),
137            aggs: aggs.clone(),
138        },
139        LogicalPlan::Join {
140            left,
141            right,
142            keys,
143            how,
144        } => PhysicalPlan::JoinExec {
145            left: Box::new(compile(left)?),
146            right: Box::new(compile(right)?),
147            keys: keys.clone(),
148            how: *how,
149        },
150        LogicalPlan::Sort { input, options } => PhysicalPlan::SortExec {
151            input: Box::new(compile(input)?),
152            options: options.clone(),
153        },
154        LogicalPlan::Slice {
155            input,
156            offset,
157            len,
158            from_end,
159        } => PhysicalPlan::SliceExec {
160            input: Box::new(compile(input)?),
161            offset: *offset,
162            len: *len,
163            from_end: *from_end,
164        },
165        LogicalPlan::Unique { input, subset } => PhysicalPlan::UniqueExec {
166            input: Box::new(compile(input)?),
167            subset: subset.clone(),
168        },
169        LogicalPlan::FillNull { input, fill } => PhysicalPlan::FillNullExec {
170            input: Box::new(compile(input)?),
171            fill: fill.clone(),
172        },
173        LogicalPlan::DropNulls { input, subset } => PhysicalPlan::DropNullsExec {
174            input: Box::new(compile(input)?),
175            subset: subset.clone(),
176        },
177        LogicalPlan::NullCount { input } => PhysicalPlan::NullCountExec {
178            input: Box::new(compile(input)?),
179        },
180        LogicalPlan::Explode { input, column } => PhysicalPlan::ExplodeExec {
181            input: Box::new(compile(input)?),
182            column: column.clone(),
183        },
184        LogicalPlan::Implode { input } => PhysicalPlan::ImplodeExec {
185            input: Box::new(compile(input)?),
186        },
187    };
188
189    Ok(plan)
190}
alopex_dataframe/physical/plan.rs

alopex_dataframe/physical/
plan.rs
