pandrs 0.3.0

A high-performance DataFrame library for Rust, providing pandas-like API with advanced features including SIMD optimization, parallel processing, and distributed computing capabilities
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151

//! # Execution Plan Conversion
//!
//! This module provides functionality for converting execution plans to plan nodes.

use super::core::PlanNode;
use crate::distributed::execution::{ExecutionPlan, Operation};
use crate::distributed::window::WindowFunction;
use crate::error::Result;

/// Explains an execution plan
pub fn explain_plan(plan: &ExecutionPlan, options: &super::core::ExplainOptions) -> Result<String> {
    let plan_node = execution_plan_to_plan_node(plan)?;
    Ok(plan_node.explain(options))
}

/// Converts an execution plan to a plan node
pub fn execution_plan_to_plan_node(plan: &ExecutionPlan) -> Result<PlanNode> {
    let node = match plan.operation() {
        Operation::Select { columns } => {
            let input = dummy_scan_node(plan);

            PlanNode::Project {
                columns: columns.clone(),
                input: Box::new(input),
            }
        }
        Operation::Filter { predicate } => {
            let input = dummy_scan_node(plan);

            PlanNode::Filter {
                predicate: predicate.clone(),
                input: Box::new(input),
                selectivity: None,
            }
        }
        Operation::Join {
            right,
            join_type,
            left_keys,
            right_keys,
        } => {
            let left_input = dummy_scan_node(plan);
            let right_input = PlanNode::Scan {
                table: right.clone(),
                columns: vec![],
                statistics: None,
            };

            let join_type_str = match join_type {
                crate::distributed::execution::JoinType::Inner => "INNER",
                crate::distributed::execution::JoinType::Left => "LEFT",
                crate::distributed::execution::JoinType::Right => "RIGHT",
                crate::distributed::execution::JoinType::Full => "FULL OUTER",
                crate::distributed::execution::JoinType::Cross => "CROSS",
            };

            let keys = left_keys
                .iter()
                .zip(right_keys.iter())
                .map(|(l, r)| (l.clone(), r.clone()))
                .collect();

            PlanNode::Join {
                join_type: join_type_str.to_string(),
                left: Box::new(left_input),
                right: Box::new(right_input),
                keys,
            }
        }
        Operation::GroupBy { keys, aggregates } => {
            let input = dummy_scan_node(plan);
            let agg_exprs = aggregates
                .iter()
                .map(|agg| format!("{}({}) as {}", agg.function, agg.input, agg.output))
                .collect();

            PlanNode::Aggregate {
                keys: keys.clone(),
                aggregates: agg_exprs,
                input: Box::new(input),
            }
        }
        Operation::OrderBy { sort_exprs } => {
            let input = dummy_scan_node(plan);
            let sort_exprs_str = sort_exprs
                .iter()
                .map(|expr| {
                    let direction = if expr.ascending { "ASC" } else { "DESC" };
                    let nulls = if expr.nulls_first {
                        "NULLS FIRST"
                    } else {
                        "NULLS LAST"
                    };
                    format!("{} {} {}", expr.column, direction, nulls)
                })
                .collect();

            PlanNode::Sort {
                sort_exprs: sort_exprs_str,
                input: Box::new(input),
            }
        }
        Operation::Limit { limit } => {
            let input = dummy_scan_node(plan);

            PlanNode::Limit {
                limit: *limit,
                input: Box::new(input),
            }
        }
        Operation::Window { window_functions } => {
            let input = dummy_scan_node(plan);
            let window_exprs = window_functions.iter().map(|wf| wf.to_sql()).collect();

            PlanNode::Window {
                window_functions: window_exprs,
                input: Box::new(input),
            }
        }
        Operation::Custom { name, params } => {
            let input = dummy_scan_node(plan);

            PlanNode::Custom {
                name: name.clone(),
                params: params.clone(),
                input: Box::new(input),
            }
        }
    };

    Ok(node)
}

/// Creates a dummy scan node for an execution plan's input
pub fn dummy_scan_node(plan: &ExecutionPlan) -> PlanNode {
    if plan.inputs().is_empty() {
        // No inputs, use a dummy table
        PlanNode::Scan {
            table: "DUMMY".to_string(),
            columns: vec![],
            statistics: None,
        }
    } else {
        // Use the first input as the scan source
        PlanNode::Scan {
            table: plan.inputs()[0].clone(),
            columns: vec![],
            statistics: None,
        }
    }
}