RisingLight 学习笔记-Binder

SQL 解析后，会调用 Binder::bind 方法对 AST 中的符号进行绑定，这个过程会查找 Catalog，如果 Catalog 中没有相关的信息(比如找不到相关表，或者在表中找不到相关列)，这条 SQL 就不能执行。如果有相关信息，就会对这些信息进行绑定，转化为内部可以识别的形式。binder 相关代码在 src/binder/下

`select a, sum(b) from t` 经过 binder 阶段后的信息

Select(
  BoundSelect {
    select_list: [
      Column #BaseTableColumnRefId { database_id: 0, schema_id: 0, table_id: 18, column_id: 0 },
      Sum([Column #BaseTableColumnRefId { database_id: 0, schema_id: 0, table_id: 18, column_id: 1 }]) -> Int32 (nullable) (agg),
    ],
    from_table: Some(
      JoinTableRef {
        relation: BaseTableRef {
          ref_id: 0.0.18,
          table_name: "t",
          column_ids: [
            0,
            1,
          ],
          column_descs: [
            ColumnDesc {
              datatype: Int32 (nullable),
              name: "a",
              is_primary: false,
            },
            ColumnDesc {
              datatype: Int32 (nullable),
              name: "b",
              is_primary: false,
            },
          ],
          is_internal: false,
        },
        join_tables: [],
      },
    ),
    where_clause: None,
    select_distinct: false,
    group_by: [
      Column #BaseTableColumnRefId { database_id: 0, schema_id: 0, table_id: 18, column_id: 0 },
    ],
  },
)

binder 的结构如下所示：

/// The binder resolves all expressions referring to schema objects such as
/// tables or views with their column names and types.
pub struct Binder {
    egraph: egg::EGraph<Node, TypeSchemaAnalysis>,
    catalog: Arc<RootCatalog>,
    contexts: Vec<Context>,
    /// The number of occurrences of each table in the query.
    table_occurrences: HashMap<TableRefId, u32>,
}

/// The context of binder execution.
#[derive(Debug, Default)]
struct Context {
    /// Table names that can be accessed from the current query.
    table_aliases: HashSet<String>,
    /// Column names that can be accessed from the current query.
    /// column_name -> (table_name -> id)
    aliases: HashMap<String, HashMap<String, Id>>,
    /// Column names that can be accessed from the outside query.
    /// column_name -> id
    output_aliases: HashMap<String, Id>, 
}

在 bind 的时候，会根据 SQL 的类型执行不同的操作，比如 CREATE 会执行 bind_create_table，SELECT 会执行 bind_query …

fn bind_stmt(&mut self, stmt: Statement) -> Result {
    match stmt {
        Statement::CreateTable {
            name, columns, constraints, ..
        } => self.bind_create_table(name, &columns, &constraints),
        Statement::Drop {
            object_type, if_exists, names, cascade, ..
        } => self.bind_drop(object_type, if_exists, names, cascade),
        Statement::Insert {
            table_name, columns, source, ..
        } => self.bind_insert(table_name, columns, source),
        Statement::Delete {
            from, selection, ..
        } => self.bind_delete(from, selection),
        Statement::Copy {
            source, to, target, options, ..
        } => self.bind_copy(source, to, target, &options),
        Statement::Query(query) => self.bind_query(*query).map(|(id, _)| id),
        Statement::Explain { statement, .. } => self.bind_explain(*statement),
        Statement::ShowVariable { .. }
        | Statement::ShowCreate { .. }
        | Statement::ShowColumns { .. } => Err(BindError::NotSupportedTSQL),
        _ => Err(BindError::InvalidSQL),
    }

Create Table

CREATE TABLE 在 bind 过程中主要做了下面这些检查：

• 检查 table 是否已经存在了
• 检查要创建的表中的 Column 是否有重复名称
• 检查是否有多个主键，如果一个主键有多个列，应该使用 primary key(col1, col2...) 这样的格式

pub(super) fn bind_create_table(&mut self, name: ObjectName,
    columns: &[ColumnDef],
    constraints: &[TableConstraint],
) -> Result {
    let name = lower_case_name(&name);
    let (schema_name, table_name) = split_name(&name)?;
    let schema = self.catalog.get_schema_by_name(schema_name)
        .ok_or_else(|| BindError::InvalidSchema(schema_name.into()))?;
    if schema.get_table_by_name(table_name).is_some() { // 检查 table 是否已经存在
        return Err(BindError::TableExists(table_name.into()));
    }

    // check duplicated column names 检查是否有重复列
    let mut set = HashSet::new();
    for col in columns.iter() {
        if !set.insert(col.name.value.to_lowercase()) {
            return Err(BindError::ColumnExists(col.name.value.to_lowercase()));
        }
    }

    let mut ordered_pk_ids = Binder::ordered_pks_from_columns(columns);
    let has_pk_from_column = !ordered_pk_ids.is_empty();

    if ordered_pk_ids.len() > 1 { // 不能单独有多个主键
        // multi primary key should be declared by "primary key(c1, c2...)" syntax
        return Err(BindError::NotSupportedTSQL);
    }

    let pks_name_from_constraints = Binder::pks_name_from_constraints(constraints);
    if has_pk_from_column && !pks_name_from_constraints.is_empty() {
        // can't get primary key both from "primary key(c1, c2...)" syntax and column's option
        return Err(BindError::NotSupportedTSQL);
    } else if !has_pk_from_column { // 主键是 primary key(c1, c2...) 这样的格式
        for name in &pks_name_from_constraints {
            if !set.contains(name) { // 检查表中是否包含这些列
                return Err(BindError::InvalidColumn(name.clone()));
            }
        }
        // We have used `pks_name_from_constraints` to get the primary keys' name sorted by
        // declaration order in "primary key(c1, c2..)" syntax. Now we transfer the name to id
        // to get the sorted ID 将主键拆开，得到所有的列
        ordered_pk_ids = pks_name_from_constraints
            .iter().map(|name| {
                columns.iter()
                    .position(|c| c.name.value.eq_ignore_ascii_case(name))
                    .unwrap() as ColumnId
            }).collect();
    }

    let mut columns: Vec<ColumnCatalog> = columns // 生成 `ColumnCatalog`
        .iter().enumerate().map(|(idx, col)| {
            let mut col = ColumnCatalog::from(col);
            col.set_id(idx as ColumnId);
            col
        }).collect();

    for &index in &ordered_pk_ids { // 设置 Column 主键标识
        columns[index as usize].set_primary(true);
        columns[index as usize].set_nullable(false);
    }

    // 添加一颗 CreateTable 节点，这样后续流程就可以根据这个节点执行相应的操作
    let create = self.egraph.add(Node::CreateTable(CreateTable {
        schema_id: schema.id(),
        table_name: table_name.into(),
        columns,
        ordered_pk_ids,
    }));
    Ok(create)
}

需要注意的是代码的前两行，第一行将名称转为小写，说明 RisingLight 是不区分大小写的。第二行将 name 拆分为了 scheme_name 和 table_name，说明要定位一个 table 需要 scheme name + table name 两个信息，这也印证了在 Catalog 那篇文章中的内容。

Drop Table

DROP TABLE 的内容相比 CREATE TABLE 要简单许多，他只需要校验有没有这个 table 即可

pub(super) fn bind_drop(
    &mut self,
    object_type: ObjectType,
    if_exists: bool,
    names: Vec<ObjectName>,
    cascade: bool,
) -> Result {
    match object_type {
        ObjectType::Table => {
            let name = lower_case_name(&names[0]);
            let (schema_name, table_name) = split_name(&name)?;
            let table_ref_id = self.catalog
                .get_table_id_by_name(schema_name, table_name)
                .ok_or_else(|| BindError::InvalidTable(table_name.into()))?;

            Ok(self.egraph.add(Node::Drop(BoundDrop {
                object: Object::Table(table_ref_id),
                if_exists,
                cascade,
            })))
        }
        _ => todo!(),
    }
}

Query

查询语句的实现应该是最复杂的了。一个查询语句的可能形式如下

SELECT t1.col1, t2.col2 FROM t1, t2 
    WHERE xxx AND xxx OR xxx
        GROUP BY xxx
            HAVING xxx
    ORDER BY xxx

因此 binder 需要能够检查 FROM 后面的 table、 WHERE 后面的条件、 GROUP BY、HAVING、ORDER BY、DISTINCT、需要映射的列是否都符合要求

• 涉及的 table 是否存在
• 涉及 column 的是否存在
• 条件表达式是否合法
• 操作是否合法，比如聚合操作的位置是否合法(比如不能再 where 后，也不能在 group 后)，ORDER BY 的 columns 是否都在 projection 的 columns 中等
• …

fn bind_select(&mut self, select: Select, order_by: Vec<OrderByExpr>) -> Result {
    let from = self.bind_from(select.from)?;
    let projection = self.bind_projection(select.projection, from)?;
    let where_ = self.bind_where(select.selection)?;
    let groupby = match select.group_by { // 设置 GROUP BY
        group_by if group_by.is_empty() => None,
        group_by => Some(self.bind_groupby(group_by)?),
    };
    let having = self.bind_having(select.having)?;
    let orderby = self.bind_orderby(order_by)?;
    let distinct = match select.distinct {
        None => self.egraph.add(Node::List([].into())),
        Some(Distinct::Distinct) => projection,
        Some(Distinct::On(exprs)) => self.bind_exprs(exprs)?,
    };

    let mut plan = self.egraph.add(Node::Filter([where_, from]));
    let mut to_rewrite = [projection, distinct, having, orderby];
    // 抽取聚合操作，生成聚合节点
    plan = self.plan_agg(&mut to_rewrite, groupby, plan)?;
    let [mut projection, distinct, having, orderby] = to_rewrite;
    plan = self.egraph.add(Node::Filter([having, plan]));
    plan = self.plan_window(projection, distinct, orderby, plan)?;
    plan = self.plan_distinct(distinct, orderby, &mut projection, plan)?;
    plan = self.egraph.add(Node::Order([orderby, plan]));
    plan = self.egraph.add(Node::Proj([projection, plan]));
    Ok(plan)
}

• bind_from 会校验表是否存在(bind_table_def)，并将 FROM 后面的表作为 Join 节点加入到 AST 中。如果表有别名，会将别名记录下来(也会校验是否有重复的表别名)，同时也会记录表中 Column name 到别名的映射
• bind_projection 会遍历需要映射的每一项，并根据其类型做出相应的行为：
  • SelectItem::UnnamedExpr ，可能是字面量，也可能是 table 中的 column，也可能是函数。加入 output_aliases 列表。output_aliases 记录了查询访问到的 column name。如果是 column，还需要检查 column 是否有效(bind_ident)
  • SelectItem::ExprWithAlias，带有别名，比如 apple AS a，会将别名记录下来，并加入 output_aliases 列表
  • SelectItem::Wildcard 通配符
• bind_where 根据条件执行相应的操作，这部分类型有点多，就不展开了。过如果条件操作中涉及 column，也会校验 column 是否合法。这一部分的主要操作在 bind_expr 中与条件相关的部分
• bind_having 与 bind_where 类似，区别在于如果在 bind_where 后面出现聚合操作会报错
• bind_orderby 会添加 order by 后面的列表到一个节点中，默认是升序排列。之后在 plan_distinct 也会检查所有的 order by columns 都应该在 distinct 中，即 order by columns 必须出现在 projection 的列中
• bind_groupby 会将 GROUP BY 后面的列表加入到一个节点中

这些校验方法直接或者间接地都与 bind_expr 方法有关。这个方法根据表达式的类型，执行不同的操作

Insert

INSERT 在 binder 过程中主要做了下面这些检查：

• 检查 table 是否存在
• 检查要插入的 Column 是否存在
• 检查 VALUES 列表中的元素的个数是否都相同

pub fn bind_insert(
    &mut self,
    table_name: ObjectName,
    columns: Vec<Ident>,
    source: Box<Query>,
) -> Result {
    let (table, is_internal) = self.bind_table_id(&table_name)?;
    if is_internal {
        return Err(BindError::NotSupportedOnInternalTable);
    }
    let cols = self.bind_table_columns(&table_name, &columns)?;
    let source = self.bind_query(*source)?.0;
    let id = self.egraph.add(Node::Insert([table, cols, source]));
    Ok(id)
}

bind_table_columns 会校验所有的 column 是否合法，并将每一列封装为一个节点(Node::Column) 并添加到另一个节点中去(Node::List)。

bind_query 方法与 SELECT 调用的是同一个方法，但是进入的是不同的分支，因此在 INSERT 中会调用 bind_values 方法。这个方法会比较 values 列表中元素的个数是否都是一样的，如果不一样就会返回一个 BindError::InvalidExpression

let child = match *query.body {
    SetExpr::Select(select) => self.bind_select(*select, query.order_by)?,
    SetExpr::Values(values) => self.bind_values(values)?,
    _ => todo!("handle query ???"),
};

/// Binds the VALUES clause. Returns a [`Values`](Node::Values) plan.
fn bind_values(&mut self, values: Values) -> Result {
    let values = values.rows;
    let mut bound_values = Vec::with_capacity(values.len());
    if values.is_empty() {
        return Ok(self.egraph.add(Node::Values([].into())));
    }

    let column_len = values[0].len();
    for row in values {
        if row.len() != column_len {
            return Err(BindError::InvalidExpression(
                "VALUES lists must all be the same length".into(),
            ));
        }
        bound_values.push(self.bind_exprs(row)?);
    }
    let id = self.egraph.add(Node::Values(bound_values.into()));
    self.type_(id)?;
    Ok(id)
}