SQL 支持

本章讨论的源代码可以在 KQuery 项目的 sql 模块中找到。

除了拥有手工编码逻辑计划的能力外，在某些情况下，直接写 SQL 会更方便。本章将建立一个 SQL 解析器和查询规划器，将 SQL 查询翻译成逻辑计划。

分词器

第一步是将 SQL 查询字符串转换成代表关键字、字面量、标识符和操作符的分词列表。

这是所有可能的分词的一个子集，但对当前来说已经足够了。

interface Token
data class IdentifierToken(val s: String) : Token
data class LiteralStringToken(val s: String) : Token
data class LiteralLongToken(val s: String) : Token
data class KeywordToken(val s: String) : Token
data class OperatorToken(val s: String) : Token

然后我们将需要一个分词器类。在这里展示全部代码并不特别有趣，完整的源代码可以在配套的 GitHub 仓库中找到。

class Tokenizer {
  fun tokenize(sql: String): List<Token> {
    // see github repo for code
  }
}

给定输入 "SELECT a + b FROM c"，我们期望输出如下：

listOf(
  KeywordToken("SELECT"),
  IdentifierToken("a"),
  OperatorToken("+"),
  IdentifierToken("b"),
  KeywordToken("FROM"),
  IdentifierToken("c")
)

Pratt 解析器

我们将根据 Vaughan R. Pratt 在 1973 年发表的 Top Down Operator Precedence 论文来手工编码一个 SQL 解析器。虽然有其他的方法来构建 SQL 解析器，比如使用解析器生成器和解析器组合器，但我发现 Pratt 的方法效果很好，它产生的代码高效、易于理解，而且易于调试。

下面是一个 Pratt 解析器的基本实现。在我看来，它有一种简洁的美。表达式解析是由一个简单的循环来完成的，它解析一个“前缀”表达式，后面是一个可选的 “后缀”表达式，并且一直这样做，直到优先级发生变化，使解析器发现它已经完成了对表达式的解析。当然，parsePrefix 和 parseInfix 的实现可以递归地回调到 parse 方法中，这是它变得非常强大的地方。

interface PrattParser {

  /** Parse an expression */
  fun parse(precedence: Int = 0): SqlExpr? {
    var expr = parsePrefix() ?: return null
    while (precedence < nextPrecedence()) {
      expr = parseInfix(expr, nextPrecedence())
    }
    return expr
  }

  /** Get the precedence of the next token */
  fun nextPrecedence(): Int

  /** Parse the next prefix expression */
  fun parsePrefix(): SqlExpr?

  /** Parse the next infix expression */
  fun parseInfix(left: SqlExpr, precedence: Int): SqlExpr

}

这个接口用一个新的类 SqlExpr 表示，它是对解析后的表达式的表示，并且在很大程度上是对逻辑计划中定义的表达式的一对一映射，但是对于二进制表达式，我们可以使用一个更通用的结构，其中运算符是一个字符串，而不是为我们将要支持的所有不同的二进制表达式创建单独的数据结构。

下面是一些 SqlExpr 实现的例子。

/** SQL Expression */
interface SqlExpr

/** Simple SQL identifier such as a table or column name */
data class SqlIdentifier(val id: String) : SqlExpr {
  override fun toString() = id
}

/** Binary expression */
data class SqlBinaryExpr(val l: SqlExpr, val op: String, val r: SqlExpr) : SqlExpr {
  override fun toString(): String = "$l $op $r"
}

/** SQL literal string */
data class SqlString(val value: String) : SqlExpr {
  override fun toString() = "'$value'"
}

有了这些类，就可以用以下代码来表示表达式 foo = 'bar'。

val sqlExpr = SqlBinaryExpr(SqlIdentifier("foo"), "=", SqlString("bar"))

解析 SQL 表达式

让我们通过这种方法来解析一个简单的数学表达式，如 1 + 2 * 3。这个表达式由以下分词组成。

listOf(
  LiteralLongToken("1"),
  OperatorToken("+"),
  LiteralLongToken("2"),
  OperatorToken("*"),
  LiteralLongToken("3")
)

我们需要创建一个 PrattParser trait 的实现，并将分词传到它的构造函数中。这些标记被包装在一个 TokenStream 类中，该类提供了一些便捷的方法，如用于消费下一个分词的 next，以及用于当我们想在不消费分词的情况下进行回溯的 peek。

class SqlParser(val tokens: TokenStream) : PrattParser {
}

实现 nextPrecedence 方法很简单，因为我们在这里只有少量的分词有任意优先级，我们需要让乘法和除法运算符比加法和减法运算符有更高的优先级。请注意，这个方法返回的具体数字并不重要，因为它们只是用来进行比较的。有一个关于运算符的优先级的很好的参考资料可以在 PostgreSQL 文档 中找到。

override fun nextPrecedence(): Int {
  val token = tokens.peek()
  return when (token) {
    is OperatorToken -> {
      when (token.s) {
        "+", "-" -> 50
        "*", "/" -> 60
        else -> 0
      }
    }
    else -> 0
  }
}

前缀解析器只需要知道如何解析数值字面量。

override fun parsePrefix(): SqlExpr? {
  val token = tokens.next() ?: return null
  return when (token) {
    is LiteralLongToken -> SqlLong(token.s.toLong())
    else -> throw IllegalStateException("Unexpected token $token")
  }
}

中缀解析器只需要知道如何解析运算符就可以了。请注意，在解析完一个运算符后，这个方法会递归调用顶层解析方法来解析运算符后面的表达式（二进制表达式的右侧）。

override fun parseInfix(left: SqlExpr, precedence: Int): SqlExpr {
  val token = tokens.peek()
  return when (token) {
    is OperatorToken -> {
      tokens.next()
      SqlBinaryExpr(left, token.s, parse(precedence) ?: 
                    throw SQLException("Error parsing infix"))
    }
    else -> throw IllegalStateException("Unexpected infix token $token")
  }
}

优先级逻辑可以通过解析数学表达式 1 + 2 * 3 和 1 * 2 + 3 来证明，它们应该分别被解析为 1 + (2 * 3) 和 (1 * 2) + 3。

示例：解析 1 + 2 * 3

以下这些是分词和它们的优先级。

Tokens:      [1]  [+]  [2]  [*]  [3]
Precedence:  [0] [50]  [0] [60]  [0]

最后的结果正确地表示为 1 + (2 * 3) 的表达式。

SqlBinaryExpr(
    SqlLong(1),
    "+",
    SqlBinaryExpr(SqlLong(2), "*", SqlLong(3))
)

解析 SELECT 语句

现在已经有能力解析一些简单的表达式了，下一步是扩展解析器以支持将 SELECT 语句解析为具体语法树（CST）。请注意，在其他解析方法中，例如使用像 ANTLR 这样的解析器生成器，有一个中间阶段，称为抽象语法树（AST），然后需要翻译成具体语法树，但是 Pratt 解析器方法会直接将分词转换为 CST 。

这里有一个 CST 的示例，它可以表示一个简单的带有投影和选择的单表查询。后面的章节会进行扩展，支持更复杂的查询。

data class SqlSelect(
    val projection: List<SqlExpr>,
    val selection: SqlExpr,
    val tableName: String) : SqlRelation

SQL 查询规划器

SQL 查询规划器将 SQL 查询树翻译成逻辑计划。由于 SQL 语言的灵活性，这比把逻辑计划翻译成物理计划要难得多。例如下面这个简单的查询。

SELECT id, first_name, last_name, salary/12 AS monthly_salary
FROM employee
WHERE state = 'CO' AND monthly_salary > 1000

虽然这对阅读查询的人来说是很直观的，但是查询的选择部分（WHERE子句）用到了一个表达式（state），这个表达式不包括在投影的输出中，所以显然需要在投影之前应用，但是也用到了另一个表达式（salary/12 AS monthly_salary），这个表达式只有在应用了投影之后才能使用。在 GROUP BY、HAVING 和 ORDER BY 子句中面临类似的问题。

对这个问题有多种解决办法。一种方法是把这个查询翻译成下面的逻辑计划，把选择表达式分成两步，一个在投影之前，一个在投影之后。然而，这只是因为选择表达式是一个共轭谓词（只有当所有部分都是真的时候，表达式才是真的），对于更复杂的表达式，这种方法可能不可行。如果表达式是 state = 'CO' OR monthly_salary > 1000，那么我们就不能这样做。

Filter: #monthly_salary > 1000
  Projection: #id, #first_name, #last_name, #salary/12 AS monthly_salary
    Filter: #state = 'CO'
      Scan: table=employee

一个更简单、更通用的方法是将所有需要的表达式添加到投影中，这样就可以在投影后应用选择，然后通过将输出包装在另一个投影中来删除任何添加的列。

Projection: #id, #first_name, #last_name, #monthly_salary
  Filter: #state = 'CO' AND #monthly_salary > 1000
    Projection: #id, #first_name, #last_name, #salary/12 AS monthly_salary, #state
      Scan: table=employee

值得注意的是，我们将在后面的章节中建立一个“谓词下推”的查询优化规则，它能够优化这个计划，并将谓词的 state = 'CO' 部分在计划中进一步下推，使其位于投影之前。

翻译 SQL 表达式

将 SQL 表达式转化为逻辑表达式是相当简单的，正如本例代码所演示的那样。

private fun createLogicalExpr(expr: SqlExpr, input: DataFrame) : LogicalExpr {
  return when (expr) {
    is SqlIdentifier -> Column(expr.id)
    is SqlAlias -> Alias(createLogicalExpr(expr.expr, input), expr.alias.id)
    is SqlString -> LiteralString(expr.value)
    is SqlLong -> LiteralLong(expr.value)
    is SqlDouble -> LiteralDouble(expr.value)
    is SqlBinaryExpr -> {
      val l = createLogicalExpr(expr.l, input)
      val r = createLogicalExpr(expr.r, input)
      when(expr.op) {
        // comparison operators
        "=" -> Eq(l, r)
        "!=" -> Neq(l, r)
        ">" -> Gt(l, r)
        ">=" -> GtEq(l, r)
        "<" -> Lt(l, r)
        "<=" -> LtEq(l, r)
        // boolean operators
        "AND" -> And(l, r)
        "OR" -> Or(l, r)
        // math operators
        "+" -> Add(l, r)
        "-" -> Subtract(l, r)
        "*" -> Multiply(l, r)
        "/" -> Divide(l, r)
        "%" -> Modulus(l, r)
        else -> throw SQLException("Invalid operator ${expr.op}")
      }
    }

    else -> throw new UnsupportedOperationException()
  }
}

规划 SELECT

如果我们只想支持在选择中引用的所有列也存在于投影中的情况，我们可以用一些非常简单的逻辑来建立查询计划。

fun createDataFrame(select: SqlSelect, tables: Map<String, DataFrame>) : DataFrame {

  // get a reference to the data source
  var df = tables[select.tableName] ?:
      throw SQLException("No table named '${select.tableName}'")

  val projectionExpr = select.projection.map { createLogicalExpr(it, df) }

  if (select.selection == null) {
    // apply projection
    return df.select(projectionExpr)
  }

  // apply projection then wrap in a selection (filter)
  return df.select(projectionExpr)
           .filter(createLogicalExpr(select.selection, df))
}

然而，由于选择可以同时引用投影的输入和投影的输出，需要用一个中间投影创建一个更复杂的计划。第一步是确定哪些列是由选择过滤器表达式引用的。为了做到这一点，需要使用访问者模式来遍历表达式树，并建立一个列名的可变集合。

下面遍历表达式树的工具方法。

private fun visit(expr: LogicalExpr, accumulator: MutableSet<String>) {
  when (expr) {
    is Column -> accumulator.add(expr.name)
    is Alias -> visit(expr.expr, accumulator)
    is BinaryExpr -> {
      visit(expr.l, accumulator)
      visit(expr.r, accumulator)
     }
  }
}

有了这些以后，就可以编写以下代码，将 SELECT 语句转换成有效的逻辑计划。这个代码样本并不完美，可能包含了一些边缘情况的错误，即数据源中的列和别名表达式之间存在名称冲突，但为保持代码简单，暂时忽略这一点。

fun createDataFrame(select: SqlSelect, tables: Map<String, DataFrame>) : DataFrame {

  // get a reference to the data source
  var df = tables[select.tableName] ?:
    throw SQLException("No table named '${select.tableName}'")

  // create the logical expressions for the projection
  val projectionExpr = select.projection.map { createLogicalExpr(it, df) }

  if (select.selection == null) {
    // if there is no selection then we can just return the projection
    return df.select(projectionExpr)
  }

  // create the logical expression to represent the selection
  val filterExpr = createLogicalExpr(select.selection, df)

  // get a list of columns references in the projection expression
  val columnsInProjection = projectionExpr
    .map { it.toField(df.logicalPlan()).name}
    .toSet()

  // get a list of columns referenced in the selection expression
  val columnNames = mutableSetOf<String>()
  visit(filterExpr, columnNames)

  // determine if the selection references any columns not in the projection
  val missing = columnNames - columnsInProjection

  // if the selection only references outputs from the projection we can
  // simply apply the filter expression to the DataFrame representing
  // the projection
  if (missing.size == 0) {
    return df.select(projectionExpr)
             .filter(filterExpr)
  }

  // because the selection references some columns that are not in the
  // projection output we need to create an interim projection that has
  // the additional columns and then we need to remove them after the
  // selection has been applied
  return df.select(projectionExpr + missing.map { Column(it) })
           .filter(filterExpr)
           .select(projectionExpr.map {
              Column(it.toField(df.logicalPlan()).name)
            })
}

规划聚合查询

正如你所看到的，SQL 查询规划器是相对复杂的，解析聚合查询的代码也相当复杂。如果你有兴趣了解更多，请参考源代码。

本书的电子版、MOBI和PDF格式也可从 https://leanpub.com/how-query-engines-work 购买。

Copyright © 2020-2022 Grove Enterprises, LLC。保留所有权利。