TiKV 源码解析系列文章（十五）表达式计算框架

比如说我们用这个 SQL 作为例子：

SELECT (count * price) AS sum FROM orders WHERE order_id < 100

其中 order_id < 10
 就是一个表达式，它有一个列输入参数： order_id
，输出：Bool
。

/// An expression in Reverse Polish notation, which is simply a list of RPN expression nodes.
///
/// You may want to build it using `RpnExpressionBuilder`.
#[derive(Debug, Clone)]
pub struct RpnExpression(Vec<RpnExpressionNode>);

/// A type for each node in the RPN expression list.
#[derive(Debug, Clone)]
pub enum RpnExpressionNode {
     Represents a function call.
    FnCall {
        func_meta: RpnFnMeta,
        args_len: usize,
        field_type: FieldType,
        implicit_args: Vec<ScalarValue>,
    },

     Represents a scalar constant value.
    Constant {
        value: ScalarValue,
        field_type: FieldType,
    },

     Represents a reference to a column in the columns specified in evaluation.
    ColumnRef { offset: usize },
}

// A type for each node in the RPN evaluation stack. It can be one of a scalar value node or a
/// vector value node. The vector value node can be either an owned vector value or a reference.
#[derive(Debug)]
pub enum RpnStackNode<'a> {
     Represents a scalar value. Comes from a constant node in expression list.
    Scalar {
        value: &'a ScalarValue,
        field_type: &'a FieldType,
    },

     Represents a vector value. Comes from a column reference or evaluated result.
    Vector {
        value: RpnStackNodeVectorValue<'a>,
        field_type: &'a FieldType,
    },
}

• 标量：由 Constant
   生成。
• 向量：执行 ColumnRe f
   生成，或是 FnCall
   调用返回的结果。

/// Represents a vector value node in the RPN stack.
///
/// It can be either an owned node or a reference node.
///
/// When node comes from a column reference, it is a reference node (both value and field_type
/// are references).
///
/// When nodes comes from an evaluated result, it is an owned node.
#[derive(Debug)]
pub enum RpnStackNodeVectorValue<'a> {
    Generated {
        physical_value: VectorValue,
        logical_rows: Arc<[usize]>,
    },
    Ref {
        physical_value: &'a VectorValue,
        logical_rows: &'a [usize],
    },
}

#[rpn_fn]
#[inline]
pub fn int_plus_int(
    lhs: &Option<Int>,
    rhs: &Option<Int>,
) -> Result<Option<Int>> {
    if let (Some(lhs), Some(rhs)) = (arg0, arg1) {
        lhs.checked_add(*rhs)
            .ok_or_else(|| Error::overflow("BIGINT", &format!("({} + {})", lhs, rhs)).into())
            .map(Some)
    } else {
        Ok(None)
    }
}

pub fn int_plus_int_fn_meta(
        _ctx: &mut EvalContext,
        output_rows: usize,
        args: &[RpnStackNode<'_>],
        _extra: &mut RpnFnCallExtra<'_>,
    ) -> Result<VectorValue>
{
    assert!(args.len() >= 2);

    let lhs = args[0];
    let rhs = args[1];

    let mut result: Vec<Int> = Vec::with_capacity(output_rows);

    match lhs {
        RpnStackNode::Scalar { value: ScalarValue::Int(lhs) , .. } => {
            match rhs {
                RpnStackNode::Scalar { value: ScalarValue::Int(rhs) , .. } => {
                    let value = int_plus_int(lhs, rhs);
                    result.push(result);
                }
                RpnStackNode::Vector { value: VectorValue::Int(rhs_vector) , .. } => {
                    for rhs_row in rhs_vector.logical_rows() {
                        let rhs = rhs_vector.physical_value[rhs_row];
                        let value = int_plus_int(lhs, rhs);
                        result.push(result);
                    }
                }
                _ => panic!("invalid expression")
            }
        }
        RpnStackNode::Vector { value: VectorValue::Int(lhs_vector) , .. } => {
            match rhs {
                RpnStackNode::Scalar { value: ScalarValue::Int(rhs) , .. } => {
                    for lhs in lhs_vector {
                        let value = int_plus_int(lhs, rhs);
                        result.push(result);
                    }
                }
                RpnStackNode::Vector { value: VectorValue::Int(rhs_vector) , .. } => {
                    for (lhs, rhs) in lhs_vector.logical_rows().iter().zip(rhs_vector.logical_rows()) {
                        let lhs = lhs_vector.physical_value[lhs_row];
                        let rhs = rhs_vector.physical_value[rhs_row];
                        let value = int_plus_int(lhs, rhs);
                        result.push(result);
                    }
                }
                _ => panic!("invalid expression")
            }
        }
        _ => panic!("invalid expression")
    }

    result
}

💡文中划线部分均有跳转，点击【阅读原文】查看原版文章

 TiKV 源码解析系列文章 

（一）序

（二）raft-rs proposal 示例情景分析

（三）Prometheus（上）

（四）Prometheus（下）

（五）fail-rs 介绍

（六）raft-rs 日志复制过程分析

（七）gRPC Server 的初始化和启动流程

（八）grpc-rs 的封装与实现

（九）Service 层处理流程解析

（十）Snapshot 的发送和接收

（十一）Storage - 事务控制层

（十二）分布式事务

（十三）MVCC 数据读取

（十四）Coprocessor 概览