从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

1 阅读 0 评论 0 点赞

从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

在大数据与数据库领域，理解 SQL 引擎的内部工作原理是每个系统程序员的必修课。本文将用 Rust 从零实现一个简化但完整的 SQL 引擎，覆盖 SQL 语法解析、逻辑计划构建、查询优化到向量化执行的全链路，并分享工程实践中的关键设计决策与性能陷阱。

一、为什么从零构建？

现有数据库系统如 PostgreSQL、DuckDB、ClickHouse 各自拥有复杂的优化器与执行器，代码量动辄数十万行。从零构建不是为了替代它们，而是为了：

深入理解查询执行的每一个环节
针对特定场景做极致优化（如嵌入式分析、边缘计算）
学习 Rust 在系统编程中的类型状态模式与零成本抽象实践

二、SQL 解析器：从 Token 到 AST

SQL 解析是引擎的第一步。经典的解析器生成器（如 YACC/Bison）生成的 C 代码缺乏内存安全保证，而 Rust 的 nom 或手写递归下降解析器则是更安全的选择。

词法分析 (Lexer)

词法分析器将原始 SQL 字符串拆分为 Token 流。以下是核心实现：

#[derive(Debug, Clone, PartialEq)]
pub enum Token {
    // 关键字
    Select, From, Where, Order, By, Limit, And, Or, Not,
    Insert, Into, Values, Create, Table, Index,

    // 运算符
    Eq, Ne, Lt, Gt, Le, Ge,
    Plus, Minus, Star, Slash,

    // 字面量
    Ident(String),
    Number(String),
    String(String),

    // 符号
    LParen, RParen, Comma, Semi,
}

pub struct Lexer {
    input: Vec<char>,
    pos: usize,
}

impl Lexer {
    pub fn new(input: &str) -> Self {
        Lexer { input: input.chars().collect(), pos: 0 }
    }

    pub fn next_token(&mut self) -> Result<Token, String> {
        self.skip_whitespace();

        if self.pos >= self.input.len() {
            return Err("Unexpected EOF".into());
        }

        let ch = self.input[self.pos];

        match ch {
            '=' => { self.pos += 1; Ok(Token::Eq) }
            '!' => {
                self.pos += 1;
                if self.peek() == '=' { self.pos += 1; Ok(Token::Ne) }
                else { Err("Expected != after !".into()) }
            }
            '<' => {
                self.pos += 1;
                match self.peek() {
                    '=' => { self.pos += 1; Ok(Token::Le) }
                    '>' => { self.pos += 1; Ok(Token::Ne) }
                    _ => Ok(Token::Lt),
                }
            }
            '>' => { self.pos += 1; if self.peek()=='='{self.pos+=1;Ok(Token::Ge)} else {Ok(Token::Gt)} }
            '+' => { self.pos += 1; Ok(Token::Plus) }
            '-' => { self.pos += 1; Ok(Token::Minus) }
            '*' => { self.pos += 1; Ok(Token::Star) }
            '/' => { self.pos += 1; Ok(Token::Slash) }
            ',' => { self.pos += 1; Ok(Token::Comma) }
            '(' => { self.pos += 1; Ok(Token::LParen) }
            ')' => { self.pos += 1; Ok(Token::RParen) }
            ';' => { self.pos += 1; Ok(Token::Semi) }
            '\'' => self.read_string(),
            c if c.is_ascii_digit() => self.read_number(),
            c if c.is_alphabetic() || c == '_' => self.read_identifier(),
            _ => Err(format!("Unexpected character: {}", ch)),
        }
    }

    fn read_identifier(&mut self) -> Result<Token, String> {
        let start = self.pos;
        while self.pos < self.input.len() && (self.input[self.pos].is_alphanumeric() || self.input[self.pos] == '_') {
            self.pos += 1;
        }
        let ident: String = self.input[start..self.pos].iter().collect();

        let token = match ident.to_uppercase().as_str() {
            "SELECT" => Token::Select,
            "FROM" => Token::From,
            "WHERE" => Token::Where,
            "ORDER" => Token::Order,
            "BY" => Token::By,
            "LIMIT" => Token::Limit,
            "AND" => Token::And,
            "OR" => Token::Or,
            "NOT" => Token::Not,
            "INSERT" => Token::Insert,
            "INTO" => Token::Into,
            "VALUES" => Token::Values,
            _ => Token::Ident(ident),
        };
        Ok(token)
    }

    fn read_number(&mut self) -> Result<Token, String> {
        let start = self.pos;
        while self.pos < self.input.len() && (self.input[self.pos].is_ascii_digit() || self.input[self.pos] == '.') {
            self.pos += 1;
        }
        Ok(Token::Number(self.input[start..self.pos].iter().collect()))
    }

    fn read_string(&mut self) -> Result<Token, String> {
        self.pos += 1; // skip opening quote
        let start = self.pos;
        while self.pos < self.input.len() && self.input[self.pos] != '\'' {
            self.pos += 1;
        }
        let s: String = self.input[start..self.pos].iter().collect();
        self.pos += 1; // skip closing quote
        Ok(Token::String(s))
    }

    fn peek(&self) -> char {
        if self.pos < self.input.len() { self.input[self.pos] } else { '\0' }
    }

    fn skip_whitespace(&mut self) {
        while self.pos < self.input.len() && self.input[self.pos].is_whitespace() {
            self.pos += 1;
        }
    }
}

语法分析 (Parser)

手写递归下降解析器在处理 SQL 这种上下文无关文法时非常高效。以下是 SELECT 语句的 AST 定义：

#[derive(Debug)]
pub enum Expr {
    Column(String),
    Literal(Value),
    BinaryOp {
        left: Box<Expr>,
        op: BinaryOp,
        right: Box<Expr>,
    },
    Aggregate(AggregateFunction, Box<Expr>),
}

#[derive(Debug)]
pub enum BinaryOp {
    Eq, Ne, Lt, Gt, Le, Ge,
    And, Or,
    Add, Sub, Mul, Div,
}

#[derive(Debug)]
pub enum AggregateFunction {
    Count, Sum, Avg, Min, Max,
}

#[derive(Debug)]
pub struct SelectStmt {
    pub projections: Vec<Projection>,
    pub from: String,
    pub selection: Option<WhereClause>,
    pub group_by: Vec<Expr>,
    pub order_by: Vec<OrderBy>,
    pub limit: Option<usize>,
}

#[derive(Debug)]
pub struct Projection {
    pub expr: Expr,
    pub alias: Option<String>,
}

#[derive(Debug)]
pub struct WhereClause {
    pub condition: Expr,
}

#[derive(Debug)]
pub struct OrderBy {
    pub expr: Expr,
    pub ascending: bool,
}

pub struct Parser {
    lexer: Lexer,
    current: Token,
}

impl Parser {
    pub fn new(input: &str) -> Result<Self, String> {
        let mut lexer = Lexer::new(input);
        let current = lexer.next_token()?;
        Ok(Parser { lexer, current })
    }

    fn advance(&mut self) -> Result<(), String> {
        self.current = self.lexer.next_token()?;
        Ok(())
    }

    fn expect(&mut self, token: Token) -> Result<(), String> {
        if std::mem::discriminant(&self.current) == std::mem::discriminant(&token) {
            self.advance()
        } else {
            Err(format!("Expected {:?}, got {:?}", token, self.current))
        }
    }

    pub fn parse_select(&mut self) -> Result<SelectStmt, String> {
        self.expect(Token::Select)?;

        let projections = self.parse_projections()?;
        self.expect(Token::From)?;
        let from = self.parse_identifier()?;

        let selection = if self.current == Token::Where {
            self.advance()?;
            Some(WhereClause { condition: self.parse_expr()? })
        } else {
            None
        };

        let order_by = if self.current == Token::Order {
            self.advance()?;
            self.expect(Token::By)?;
            self.parse_order_by()?
        } else {
            vec![]
        };

        let limit = if self.current == Token::Limit {
            self.advance()?;
            match &self.current {
                Token::Number(n) => {
                    let l = n.parse::<usize>().map_err(|e| e.to_string())?;
                    self.advance()?;
                    Some(l)
                }
                _ => return Err("Expected number after LIMIT".into()),
            }
        } else {
            None
        };

        Ok(SelectStmt {
            projections,
            from,
            selection,
            group_by: vec![],
            order_by,
            limit,
        })
    }

    fn parse_projections(&mut self) -> Result<Vec<Projection>, String> {
        let mut projections = vec![];

        loop {
            let expr = self.parse_expr()?;
            let alias = if self.current.to_string().to_lowercase() == "as" {
                self.advance()?;
                Some(self.parse_identifier()?)
            } else if matches!(&self.current, Token::Ident(_)) {
                Some(self.parse_identifier()?)
            } else {
                None
            };

            projections.push(Projection { expr, alias });

            if self.current == Token::Comma {
                self.advance()?;
            } else {
                break;
            }
        }

        Ok(projections)
    }

    fn parse_expr(&mut self) -> Result<Expr, String> {
        self.parse_or()
    }

    fn parse_or(&mut self) -> Result<Expr, String> {
        let mut left = self.parse_and()?;

        while self.current == Token::Or {
            self.advance()?;
            let right = self.parse_and()?;
            left = Expr::BinaryOp { 
                left: Box::new(left), 
                op: BinaryOp::Or, 
                right: Box::new(right) 
            };
        }

        Ok(left)
    }

    fn parse_and(&mut self) -> Result<Expr, String> {
        // AND has higher precedence than OR
        let mut left = self.parse_not()?;

        while self.current == Token::And {
            self.advance()?;
            let right = self.parse_not()?;
            left = Expr::BinaryOp { 
                left: Box::new(left), 
                op: BinaryOp::And, 
                right: Box::new(right) 
            };
        }

        Ok(left)
    }

    fn parse_order_by(&mut self) -> Result<Vec<OrderBy>, String> {
        let mut ordering = vec![];
        loop {
            let expr = self.parse_expr()?;
            let ascending = if self.current.to_string().to_uppercase() == "DESC" {
                self.advance()?;
                false
            } else if self.current.to_string().to_uppercase() == "ASC" {
                self.advance()?;
                true
            } else {
                true
            };
            ordering.push(OrderBy { expr, ascending });
            if self.current == Token::Comma { self.advance()?; } else { break; }
        }
        Ok(ordering)
    }

    fn parse_identifier(&mut self) -> Result<String, String> {
        match &self.current {
            Token::Ident(name) => {
                let n = name.clone();
                self.advance()?;
                Ok(n)
            }
            _ => Err(format!("Expected identifier, got {:?}", self.current)),
        }
    }
}

三、逻辑计划与查询优化

解析后的 AST 需要转换为逻辑计划，然后经过优化器处理。这是 SQL 引擎中最具工程价值的部分。

逻辑计划节点

#[derive(Debug)]
pub enum LogicalPlan {
    Scan {
        table: String,
        projections: Vec<String>,
        predicate: Option<Expr>,
    },
    Filter {
        input: Box<LogicalPlan>,
        predicate: Expr,
    },
    Project {
        input: Box<LogicalPlan>,
        expressions: Vec<(Expr, Option<String>)>,
    },
    Aggregate {
        input: Box<LogicalPlan>,
        group_exprs: Vec<Expr>,
        aggr_exprs: Vec<Expr>,
    },
    Sort {
        input: Box<LogicalPlan>,
        order_by: Vec<OrderBy>,
    },
    Limit {
        input: Box<LogicalPlan>,
        limit: usize,
    },
}

impl LogicalPlan {
    /// 递归计算该计划的输出列数
    pub fn output_columns(&self) -> usize {
        match self {
            LogicalPlan::Scan { projections, .. } => projections.len(),
            LogicalPlan::Project { expressions, .. } => expressions.len(),
            LogicalPlan::Filter { input, .. } | 
            LogicalPlan::Sort { input, .. } |
            LogicalPlan::Limit { input, .. } => input.output_columns(),
            LogicalPlan::Aggregate { group_exprs, aggr_exprs, .. } => {
                group_exprs.len() + aggr_exprs.len()
            }
        }
    }
}

查询优化器

查询优化的核心是将慢操作变快、将冗余操作减少。以下实现经典的谓词下推与列裁剪：

pub struct Optimizer;

impl Optimizer {
    pub fn optimize(plan: LogicalPlan) -> LogicalPlan {
        let plan = Self::predicate_pushdown(plan);
        let plan = Self::column_pruning(plan);
        let plan = Self::limit_pushdown(plan);
        plan
    }

    /// 谓词下推：将 Filter 尽可能靠近数据源，减少中间数据量
    fn predicate_pushdown(plan: LogicalPlan) -> LogicalPlan {
        match plan {
            LogicalPlan::Filter { input, predicate } => {
                let input = Self::predicate_pushdown(*input);

                match input {
                    // 将 Filter 穿过 Project 下推
                    LogicalPlan::Project { input: sub_input, expressions } => {
                        let pushed_predicate = Self::rewrite_predicate_through_project(
                            &predicate, &expressions
                        );
                        LogicalPlan::Project {
                            input: Box::new(LogicalPlan::Filter {
                                input: sub_input,
                                predicate: pushed_predicate,
                            }),
                            expressions,
                        }
                    }

                    // 将 Filter 与 Scan 合并
                    LogicalPlan::Scan { table, projections, predicate: scan_pred } => {
                        let combined_pred = match scan_pred {
                            Some(existing) => Expr::BinaryOp {
                                left: Box::new(existing),
                                op: BinaryOp::And,
                                right: Box::new(predicate),
                            },
                            None => predicate,
                        };
                        LogicalPlan::Scan {
                            table,
                            projections,
                            predicate: Some(combined_pred),
                        }
                    }

                    _ => LogicalPlan::Filter { input: Box::new(input), predicate },
                }
            }

            LogicalPlan::Limit { input, limit } => {
                let input = Self::predicate_pushdown(*input);
                LogicalPlan::Limit { input: Box::new(input), limit }
            }

            LogicalPlan::Project { input, expressions } => {
                let input = Self::predicate_pushdown(*input);
                LogicalPlan::Project { input: Box::new(input), expressions }
            }

            _ => plan,
        }
    }

    /// 列裁剪：只读取查询中实际需要的列
    fn column_pruning(plan: LogicalPlan) -> LogicalPlan {
        match plan {
            LogicalPlan::Project { input, expressions } => {
                let input = Self::column_pruning(*input);
                let required_cols: Vec<String> = expressions.iter()
                    .flat_map(|(e, _)| collect_column_refs(e))
                    .collect();

                match input {
                    LogicalPlan::Scan { table, projections, predicate } => {
                        let pruned = projections.into_iter()
                            .filter(|p| required_cols.contains(p))
                            .collect();
                        LogicalPlan::Project {
                            input: Box::new(LogicalPlan::Scan {
                                table,
                                projections: pruned,
                                predicate,
                            }),
                            expressions,
                        }
                    }
                    _ => LogicalPlan::Project { input: Box::new(input), expressions },
                }
            }

            LogicalPlan::Filter { input, predicate } => {
                let input = Self::column_pruning(*input);
                let filter_cols = collect_column_refs(&predicate);
                match input {
                    LogicalPlan::Scan { table, projections, predicate: scan_pred } => {
                        let mut all_cols = projections;
                        all_cols.extend(filter_cols);
                        all_cols.sort_unstable();
                        all_cols.dedup();
                        LogicalPlan::Filter {
                            input: Box::new(LogicalPlan::Scan {
                                table,
                                projections: all_cols,
                                predicate: scan_pred,
                            }),
                            predicate,
                        }
                    }
                    _ => LogicalPlan::Filter { input: Box::new(input), predicate },
                }
            }

            _ => plan,
        }
    }

    /// 重写谓词中的列引用，穿透 Projection 层
    fn rewrite_predicate_through_project(
        predicate: &Expr,
        projections: &[(Expr, Option<String>)],
    ) -> Expr {
        match predicate {
            Expr::Column(name) => {
                for (expr, alias) in projections {
                    if let Some(a) = alias {
                        if a == name { return expr.clone(); }
                    }
                    if let Expr::Column(col_name) = expr {
                        if col_name == name {
                            return expr.clone();
                        }
                    }
                }
                predicate.clone()
            }
            Expr::BinaryOp { left, op, right } => {
                Expr::BinaryOp {
                    left: Box::new(Self::rewrite_predicate_through_project(left, projections)),
                    op: op.clone(),
                    right: Box::new(Self::rewrite_predicate_through_project(right, projections)),
                }
            }
            _ => predicate.clone(),
        }
    }

    /// LIMIT 下推：减少上游数据处理量
    fn limit_pushdown(plan: LogicalPlan) -> LogicalPlan {
        // LIMIT 下推到 Scan 层或 Sort 层
        // 实际实现需要考虑是否有 ORDER BY 等情况
        plan
    }
}

fn collect_column_refs(expr: &Expr) -> Vec<String> {
    let mut cols = Vec::new();
    collect_columns(expr, &mut cols);
    cols
}

fn collect_columns(expr: &Expr, cols: &mut Vec<String>) {
    match expr {
        Expr::Column(name) => cols.push(name.clone()),
        Expr::BinaryOp { left, right, .. } => {
            collect_columns(left, cols);
            collect_columns(right, cols);
        }
        Expr::Aggregate(_, inner) => collect_columns(inner, cols),
        Expr::Literal(_) => {}
    }
}

四、向量化执行引擎

现代分析型数据库普遍采用向量化执行（也称批处理执行），相比传统的 Volcano 模型（一次处理一行），向量化执行一次处理一批数据（通常 1024 行），能充分利用 CPU 流水线和 SIMD 指令。

列式内存布局

use std::sync::Arc;

/// 类型安全的值枚举
#[derive(Debug, Clone)]
pub enum Value {
    Null,
    Boolean(bool),
    Int64(i64),
    Float64(f64),
    String(String),
}

/// 列式批处理：一个批次包含多列数据
#[derive(Debug)]
pub struct RecordBatch {
    pub columns: Vec<Column>,
    pub row_count: usize,
}

#[derive(Debug)]
pub struct Column {
    pub name: String,
    pub data: ColumnData,
}

#[derive(Debug)]
pub enum ColumnData {
    Int64(Vec<i64>),
    Float64(Vec<f64>),
    String<Vec<String>),
    Boolean(Vec<bool>),
}

impl RecordBatch {
    pub fn new(columns: Vec<Column>) -> Self {
        let row_count = columns.first().map(|c| c.len()).unwrap_or(0);
        RecordBatch { columns, row_count }
    }

    pub fn column(&self, idx: usize) -> Option<&Column> {
        self.columns.get(idx)
    }
}

impl Column {
    pub fn len(&self) -> usize {
        match &self.data {
            ColumnData::Int64(v) => v.len(),
            ColumnData::Float64(v) => v.len(),
            ColumnData::String(v) => v.len(),
            ColumnData::Boolean(v) => v.len(),
        }
    }

    pub fn is_empty(&self) -> bool { self.len() == 0 }
}

向量化表达式求值

/// 向量化表达式求值器
pub struct ExpressionEvaluator<'a> {
    batch: &'a RecordBatch,
}

impl<'a> ExpressionEvaluator<'a> {
    pub fn new(batch: &'a RecordBatch) -> Self {
        ExpressionEvaluator { batch }
    }

    /// 向量化求值：对整个批次执行表达式，返回布尔列
    pub fn evaluate(&self, expr: &Expr) -> Result<ColumnData, String> {
        match expr {
            Expr::BinaryOp { left, op, right } => {
                self.evaluate_binary_op(left, op, right)
            }
            Expr::Column(name) => {
                self.get_column(name)
            }
            Expr::Literal(val) => {
                self.scalar_to_column(val)
            }
            _ => Err("Unsupported expression in evaluator".into()),
        }
    }

    fn evaluate_binary_op(
        &self,
        left: &Expr,
        op: &BinaryOp,
        right: &Expr,
    ) -> Result<ColumnData, String> {
        let left_eval = self.evaluate(left)?;
        let right_eval = self.evaluate(right)?;

        match (left_eval, right_eval) {
            (ColumnData::Int64(a), ColumnData::Int64(b)) => {
                let result: Vec<bool> = a.iter().zip(b.iter())
                    .map(|(x, y)| Self::compare_int(x, op, y))
                    .collect();
                Ok(ColumnData::Boolean(result))
            }
            (ColumnData::Float64(a), ColumnData::Float64(b)) => {
                let result: Vec<bool> = a.iter().zip(b.iter())
                    .map(|(x, y)| Self::compare_float(x, op, y))
                    .collect();
                Ok(ColumnData::Boolean(result))
            }
            (ColumnData::Int64(a), ColumnData::Float64(b)) => {
                let a: Vec<f64> = a.iter().map(|x| *x as f64).collect();
                let result: Vec<bool> = a.iter().zip(b.iter())
                    .map(|(x, y)| Self::compare_float(x, op, y))
                    .collect();
                Ok(ColumnData::Boolean(result))
            }
            _ => Err("Type mismatch in binary operation".into()),
        }
    }

    fn compare_int(a: &i64, op: &BinaryOp, b: &i64) -> bool {
        match op {
            BinaryOp::Eq => a == b,
            BinaryOp::Ne => a != b,
            BinaryOp::Lt => a < b,
            BinaryOp::Gt => a > b,
            BinaryOp::Le => a <= b,
            BinaryOp::Ge => a >= b,
            _ => false,
        }
    }

    fn compare_float(a: &f64, op: &BinaryOp, b: &f64) -> bool {
        match op {
            BinaryOp::Eq => (a - b).abs() < f64::EPSILON,
            BinaryOp::Ne => (a - b).abs() >= f64::EPSILON,
            BinaryOp::Lt => a < b,
            BinaryOp::Gt => a > b,
            BinaryOp::Le => a <= b,
            BinaryOp::Ge => a >= b,
            _ => false,
        }
    }

    fn get_column(&self, name: &str) -> Result<ColumnData, String> {
        self.batch.columns.iter()
            .find(|c| c.name == name)
            .map(|c| c.data.clone())
            .ok_or_else(|| format!("Column '{}' not found", name))
    }

    fn scalar_to_column(&self, val: &Value) -> Result<ColumnData, String> {
        let n = self.batch.row_count;
        match val {
            Value::Int64(v) => Ok(ColumnData::Int64(vec![*v; n])),
            Value::Float64(v) => Ok(ColumnData::Float64(vec![*v; n])),
            _ => Err("Unsupported literal type".into()),
        }
    }
}

执行器与流水线

/// 执行器 trait - 定义节点执行接口
pub trait Executor: Send {
    /// 执行并返回一批结果
    fn execute(&mut self) -> Result<Option<RecordBatch>, String>;

    /// 子节点
    fn children(&self) -> &[Box<dyn Executor>];
}

/// 全表扫描执行器
pub struct ScanExecutor {
    table: String,
    projections: Vec<String>,
    predicate: Option<Expr>,
    position: usize,
}

impl ScanExecutor {
    pub fn new(table: String, projections: Vec<String>, predicate: Option<Expr>) -> Self {
        ScanExecutor { table, projections, predicate, position: 0 }
    }
}

impl Executor for ScanExecutor {
    fn execute(&mut self) -> Result<Option<RecordBatch>, String> {
        // 模拟从存储读取数据
        let batch = self.read_next_batch()?;
        Ok(Some(batch))
    }

    fn children(&self) -> &[Box<dyn Executor>] { &[] }
}

impl ScanExecutor {
    fn read_next_batch(&mut self) -> Result<RecordBatch, String> {
        // 实际实现中会异步读取消下一页数据
        let row_count = 1024;
        let columns: Vec<Column> = self.projections.iter()
            .map(|name| Column {
                name: name.clone(),
                data: ColumnData::Int64((0..row_count).map(|i| i as i64).collect()),
            })
            .collect();
        self.position += row_count;
        Ok(RecordBatch::new(columns))
    }
}

/// 过滤执行器
pub struct FilterExecutor {
    input: Box<dyn Executor>,
    predicate: Expr,
}

impl Executor for FilterExecutor {
    fn execute(&mut self) -> Result<Option<RecordBatch>, String> {
        let batch = match self.input.execute()? {
            Some(b) => b,
            None => return Ok(None),
        };

        let evaluator = ExpressionEvaluator::new(&batch);
        let mask = evaluator.evaluate(&self.predicate)?;

        let filtered = apply_mask(&batch, &mask)?;
        Ok(Some(filtered))
    }

    fn children(&self) -> &[Box<dyn Executor>] { std::slice::from_ref(&self.input) }
}

/// 投影执行器
pub struct ProjectExecutor {
    input: Box<dyn Executor>,
    expressions: Vec<(Expr, Option<String>)>,
}

impl Executor for ProjectExecutor {
    fn execute(&mut self) -> Result<Option<RecordBatch>, String> {
        let batch = match self.input.execute()? {
            Some(b) => b,
            None => return Ok(None),
        };

        let evaluator = ExpressionEvaluator::new(&batch);
        let mut new_columns = Vec::with_capacity(self.expressions.len());

        for (expr, alias) in &self.expressions {
            let result = evaluator.evaluate(expr)?;
            let name = alias.clone().unwrap_or_else(|| format!("{:?}", expr));
            new_columns.push(Column { name, data: result });
        }

        Ok(Some(RecordBatch::new(new_columns)))
    }

    fn children(&self) -> &[Box<dyn Executor>] { std::slice::from_ref(&self.input) }
}

/// 应用布尔掩码过滤行
fn apply_mask(batch: &RecordBatch, mask: &ColumnData) -> Result<RecordBatch, String> {
    let bools = match mask {
        ColumnData::Boolean(v) => v,
        _ => return Err("Mask must be boolean".into()),
    };

    let mut new_columns = Vec::with_capacity(batch.columns.len());

    for col in &batch.columns {
        let filtered = match &col.data {
            ColumnData::Int64(data) => {
                let v: Vec<i64> = data.iter().zip(bools.iter())
                    .filter(|(_, m)| **m)
                    .map(|(v, _)| *v)
                    .collect();
                ColumnData::Int64(v)
            }
            ColumnData::Float64(data) => {
                let v: Vec<f64> = data.iter().zip(bools.iter())
                    .filter(|(_, m)| **m)
                    .map(|(v, _)| *v)
                    .collect();
                ColumnData::Float64(v)
            }
            ColumnData::String(data) => {
                let v: Vec<String> = data.iter().zip(bools.iter())
                    .filter(|(_, m)| **m)
                    .map(|(v, _)| v.clone())
                    .collect();
                ColumnData::String(v)
            }
            ColumnData::Boolean(data) => {
                let v: Vec<bool> = data.iter().zip(bools.iter())
                    .filter(|(_, m)| **m)
                    .map(|(v, _)| *v)
                    .collect();
                ColumnData::Boolean(v)
            }
        };
        new_columns.push(Column { name: col.name.clone(), data: filtered });
    }

    Ok(RecordBatch::new(new_columns))
}

五、完整查询执行流程

将以上组件串联起来，构成完整的查询执行流程：

/// SQL 引擎主结构
pub struct SqlEngine {
    catalog: Catalog,
}

#[derive(Debug, Default)]
pub struct Catalog {
    tables: HashMap<String, TableMeta>,
}

#[derive(Debug)]
pub struct TableMeta {
    pub name: String,
    pub columns: Vec<ColumnMeta>,
}

#[derive(Debug)]
pub struct ColumnMeta {
    pub name: String,
    pub data_type: DataType,
}

#[derive(Debug)]
pub enum DataType {
    Int64, Float64, String, Boolean,
}

impl SqlEngine {
    pub fn new() -> Self {
        SqlEngine { catalog: Catalog::default() }
    }

    pub fn register_table(&mut self, name: &str, columns: Vec<(&str, DataType)>) {
        let meta = TableMeta {
            name: name.to_string(),
            columns: columns.iter().map(|(n, t)| ColumnMeta {
                name: n.to_string(),
                data_type: t.clone(),
            }).collect(),
        };
        self.catalog.tables.insert(name.to_string(), meta);
    }

    /// 执行 SQL 语句
    pub fn execute(&self, sql: &str) -> Result<Vec<RecordBatch>, String> {
        // 1. 解析 SQL
        let mut parser = Parser::new(sql)?;
        let ast = parser.parse_select()?;

        // 2. 构建逻辑计划
        let plan = self.build_logical_plan(&ast)?;

        // 3. 优化逻辑计划
        let optimized = Optimizer::optimize(plan);

        // 4. 构建物理计划
        let physical = self.build_physical_plan(&optimized);

        // 5. 执行物理计划
        let mut results = Vec::new();
        loop {
            match physical.execute()? {
                Some(batch) => results.push(batch),
                None => break,
            }
        }

        Ok(results)
    }

    fn build_logical_plan(&self, ast: &SelectStmt) -> Result<LogicalPlan, String> {
        let scan = LogicalPlan::Scan {
            table: ast.from.clone(),
            projections: self.infer_projections(&ast.projections),
            predicate: ast.selection.as_ref().map(|w| w.condition.clone()),
        };

        let filter = match &ast.selection {
            Some(where_clause) => {
                LogicalPlan::Filter {
                    input: Box::new(scan),
                    predicate: where_clause.condition.clone(),
                }
            }
            None => scan,
        };

        let project = LogicalPlan::Project {
            input: Box::new(filter),
            expressions: ast.projections.iter()
                .map(|p| (p.expr.clone(), p.alias.clone()))
                .collect(),
        };

        let plan = match &ast.order_by.is_empty() {
            false => LogicalPlan::Sort { input: Box::new(project), order_by: ast.order_by.clone() },
            true => project,
        };

        Ok(plan)
    }

    fn build_physical_plan(&self, plan: &LogicalPlan) -> Box<dyn Executor> {
        match plan {
            LogicalPlan::Scan { table, projections, predicate } => {
                Box::new(ScanExecutor::new(table.clone(), projections.clone(), predicate.clone()))
            }
            LogicalPlan::Filter { input, predicate } => {
                let child = self.build_physical_plan(input);
                Box::new(FilterExecutor { input: child, predicate: predicate.clone() })
            }
            LogicalPlan::Project { input, expressions } => {
                let child = self.build_physical_plan(input);
                Box::new(ProjectExecutor { input: child, expressions: expressions.clone() })
            }
            _ => todo!("Implement more physical operators"),
        }
    }

    fn infer_projections(&self, projections: &[Projection]) -> Vec<String> {
        projections.iter()
            .map(|p| match &p.expr {
                Expr::Column(name) => name.clone(),
                _ => format!("{:?}", p.expr),
            })
            .collect()
    }
}

六、工程实践中的性能陷阱与优化

1. 内存分配优化

向量化执行中，每次 batch 都会创建大量临时 Vec。mimalloc 或 jemalloc 比系统默认分配器更优：

use mimalloc::MiMalloc;

#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;

2. 零拷贝列引用

使用 Arc<[T]> 替代 Vec 实现列的共享引用，避免不必要的克隆：

pub struct ColumnView {
    pub name: String,
    pub data: Arc<[Value]>,
}

3. 分支预测友好的表达式求值

对谓词求值，按选择性排序条件。高选择性（能过滤更多数据）的谓词先求值：

fn sort_predicates_by_selectivity(predicates: &mut [(Expr, f64)]) {
    // 按选择性从高到低排序（选择性 = 通过率，值越小过滤越多）
    predicates.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap());
}

4. 编译时常量折叠

在解析阶段就将常量表达式求值，避免运行时重复计算：

fn fold_constants(expr: Expr) -> Expr {
    match expr {
        Expr::BinaryOp { left, op, right } => {
            let left = fold_constants(*left);
            let right = fold_constants(*right);

            if let (Expr::Literal(a), Expr::Literal(b)) = (&left, &right) {
                if let Some(result) = eval_literal_op(a, &op, b) {
                    return Expr::Literal(result);
                }
            }

            Expr::BinaryOp { left: Box::new(left), op, right: Box::new(right) }
        }
        _ => expr,
    }
}

七、总结

从零构建 SQL 引擎是理解数据库系统运作的最佳实践。通过本文的实现，我们覆盖了：

词法与语法解析：用 nom 或手写递归下降实现 Token 流处理
逻辑计划与优化器：谓词下推、列裁剪等经典优化规则
向量化执行引擎：列式内存布局、批处理求值、流水线执行
性能工程：内存分配器选择、零拷贝、分支预测友好设计

工程实践中，一个生产级 SQL 引擎还需要事务支持、WAL 日志、并发控制、分布式执行等复杂机制。但核心的解析-优化-执行三阶段架构是相通的。用 Rust 来实现不仅能保证内存安全，其 trait 系统和枚举类型还能让阶段间的数据流更加类型安全，许多逻辑错误在编译期就能被拦截。

最终代码可参考 DuckDB 的极简版本 Limbo（原 libSQL），它用 Rust 实现了 SQLite 的完整语义，是学习 SQL 引擎实现的绝佳素材。

参考资料

点赞(0) 打赏

本文分类：AI应用开发
本文标签：Rust Database SQL Engine Vectorized Execution Query Optimization
浏览次数：1 次浏览
发布日期：2026-10-02 19:36:28
本文链接：https://www.ybb.press/ai-applications/14130.html

从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

一、为什么从零构建？

二、SQL 解析器：从 Token 到 AST

词法分析 (Lexer)

语法分析 (Parser)

三、逻辑计划与查询优化

逻辑计划节点

查询优化器

四、向量化执行引擎

列式内存布局

向量化表达式求值

执行器与流水线

五、完整查询执行流程

六、工程实践中的性能陷阱与优化

1. 内存分配优化

2. 零拷贝列引用

3. 分支预测友好的表达式求值

4. 编译时常量折叠

七、总结

参考资料

评论列表共有 0 条评论

发表评论取消回复

从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

从零构建 Rust SQL 引擎：从解析器到向量化执行器的工程实践

一、为什么从零构建？

二、SQL 解析器：从 Token 到 AST

词法分析 (Lexer)

语法分析 (Parser)

三、逻辑计划与查询优化

逻辑计划节点

查询优化器

四、向量化执行引擎

列式内存布局

向量化表达式求值

执行器与流水线

五、完整查询执行流程

六、工程实践中的性能陷阱与优化

1. 内存分配优化

2. 零拷贝列引用

3. 分支预测友好的表达式求值

4. 编译时常量折叠

七、总结

参考资料

从零构建 SSH 服务器：协议握手、密钥交换与通道复用的 Rust 工程实战

从零构建 DuckDB 嵌入式分析引擎：列存向量化执行与 SIMD 优化实战

Rust Web 框架实战：axum 从原理到生产级部署深度指南

Polars 深度实战：Rust 重写的 DataFrame 引擎如何颠覆 Python 数据分析范式

评论列表 共有 0 条评论

发表评论 取消回复

评论列表共有 0 条评论

发表评论取消回复