aedb 0.2.3

Embedded Rust storage engine with transactional commits, WAL durability, and snapshot-consistent reads
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

use crate::catalog::schema::TableSchema;
use crate::query::error::QueryError;
use crate::query::plan::{Query, QueryPlan};

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PhysicalPlanNode {
    Scan {
        table: String,
        index: Option<String>,
    },
    Filter {
        expr: bool,
        child: Box<PhysicalPlanNode>,
    },
    Sort {
        columns: Vec<String>,
        child: Box<PhysicalPlanNode>,
    },
    Aggregate {
        group_by: Vec<String>,
        aggregate_count: usize,
        child: Box<PhysicalPlanNode>,
    },
    Project {
        columns: Vec<String>,
        child: Box<PhysicalPlanNode>,
    },
    Limit {
        limit: usize,
        child: Box<PhysicalPlanNode>,
    },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PlannedQuery {
    pub plan: QueryPlan,
    pub root: PhysicalPlanNode,
    pub stages: Vec<ExecutionStage>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutionStage {
    Scan,
    Filter,
    Sort,
    Aggregate,
    Having,
    Project,
    Limit,
}

pub fn build_physical_plan(
    schema: &TableSchema,
    query: &Query,
    index_used: Option<String>,
    estimated_scan_rows: u64,
    has_residual_filter: bool,
) -> Result<PlannedQuery, QueryError> {
    let mut node = PhysicalPlanNode::Scan {
        table: query.table.clone(),
        index: index_used.clone(),
    };
    let mut stages = vec![ExecutionStage::Scan];

    if query.predicate.is_some() {
        node = PhysicalPlanNode::Filter {
            expr: true,
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Filter);
    }
    if !query.order_by.is_empty() {
        node = PhysicalPlanNode::Sort {
            columns: query.order_by.iter().map(|(c, _)| c.clone()).collect(),
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Sort);
    }
    if !query.aggregates.is_empty() {
        node = PhysicalPlanNode::Aggregate {
            group_by: query.group_by.clone(),
            aggregate_count: query.aggregates.len(),
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Aggregate);
    }
    if query.having.is_some() {
        node = PhysicalPlanNode::Filter {
            expr: true,
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Having);
    }
    if !query.select.is_empty() && query.select[0] != "*" {
        node = PhysicalPlanNode::Project {
            columns: query.select.clone(),
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Project);
    }
    if let Some(limit) = query.limit {
        node = PhysicalPlanNode::Limit {
            limit,
            child: Box::new(node),
        };
        stages.push(ExecutionStage::Limit);
    }

    let output_columns = if !query.select.is_empty() && query.select[0] != "*" {
        query.select.clone()
    } else {
        schema.columns.iter().map(|c| c.name.clone()).collect()
    };

    Ok(PlannedQuery {
        plan: QueryPlan {
            output_columns,
            index_used,
            estimated_scan_rows,
            has_residual_filter,
        },
        root: node,
        stages,
    })
}