Catalyst 优化逻辑执行计划规则

Optimizer

object Optimizer extends RuleExecutor[LogicalPlan] {
val batches =
Batch("ConstantFolding", Once,
ConstantFolding, // 可静态分析的常量表达式
BooleanSimplification, // 布尔表达式提前短路
SimplifyFilters, // 简化过滤操作(false, true, null)
SimplifyCasts) :: // 简化转换(对象所属类已经是Cast目标类)
Batch("Filter Pushdown", Once,
CombineFilters, // 相邻(上下级)Filter操作合并
PushPredicateThroughProject, // 映射操作中的Filter谓词下推
PushPredicateThroughInnerJoin) :: Nil // inner join操作谓词下推
}

ConstantFolding 

object ConstantFolding extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case q: LogicalPlan => q transformExpressionsDown {
// Skip redundant folding of literals.
case l: Literal => l
case e if e.foldable => Literal(e.apply(null), e.dataType)
}
}
}

BooleanSimplification 

object BooleanSimplification extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case q: LogicalPlan => q transformExpressionsUp {
case and @ And(left, right) =>
(left, right) match {
case (Literal(true, BooleanType), r) => r
case (l, Literal(true, BooleanType)) => l
case (Literal(false, BooleanType), _) => Literal(false)
case (_, Literal(false, BooleanType)) => Literal(false)
case (_, _) => and
}

case or @ Or(left, right) =>
(left, right) match {
case (Literal(true, BooleanType), _) => Literal(true)
case (_, Literal(true, BooleanType)) => Literal(true)
case (Literal(false, BooleanType), r) => r
case (l, Literal(false, BooleanType)) => l
case (_, _) => or
}
}
}
}

SimplifyFilters 

object SimplifyFilters extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case Filter(Literal(true, BooleanType), child) =>
child
case Filter(Literal(null, _), child) =>
LocalRelation(child.output)
case Filter(Literal(false, BooleanType), child) =>
LocalRelation(child.output)
}
}

SimplifyCasts 

object SimplifyCasts extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
case Cast(e, dataType) if e.dataType == dataType => e
}
}

CombineFilters 

object CombineFilters extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case ff @ Filter(fc, nf @ Filter(nc, grandChild)) => Filter(And(nc, fc), grandChild)
}
}

PushPredicateThroughProject 

object PushPredicateThroughProject extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case filter @ Filter(condition, project @ Project(fields, grandChild)) =>
val sourceAliases = fields.collect { case a @ Alias(c, _) =>
(a.toAttribute: Attribute) -> c
}.toMap // 把fields中的别名属性都取出来
project.copy(child = filter.copy( // 生成新的Filter操作
replaceAlias(condition, sourceAliases), // condition中有别名的替换掉
grandChild))
}

def replaceAlias(condition: Expression, sourceAliases: Map[Attribute, Expression]): Expression = {
condition transform {
case a: AttributeReference => sourceAliases.getOrElse(a, a)
}
}
}

PushPredicateThroughInnerJoin 

object PushPredicateThroughInnerJoin extends Rule[LogicalPlan] with PredicateHelper {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case f @ Filter(filterCondition, Join(left, right, Inner, joinCondition)) =>
// 这一步是把过滤条件和join条件里的condition都提取出来
val allConditions = splitConjunctivePredicates(filterCondition) ++
joinCondition.map(splitConjunctivePredicates).getOrElse(Nil)

// 把参考属性都属于右侧输出属性的condition挑选到rightCondition里
val (rightConditions, leftOrJoinConditions) =
allConditions.partition(_.references subsetOf right.outputSet)
// 同理，把剩余condition里面，参考属性都属于左侧输出属性的condition挑选到
// leftCondition里，剩余的就属于joinCondition
val (leftConditions, joinConditions) =
leftOrJoinConditions.partition(_.references subsetOf left.outputSet)

// 生成新的left和right：先把condition里的操作用AND折叠起来，然后将该折叠后的表达式和原始的left/right logical plan合起来生成新的Filter操作，即新的Fil      // ter logical plan
// 这样就做到了把过滤条件中的谓词下推到了left/right里，即本次inner join的“外部”
val newLeft = leftConditions.reduceLeftOption(And).map(Filter(_, left)).getOrElse(left)
val newRight = rightConditions.reduceLeftOption(And).map(Filter(_, right)).getOrElse(right)
Join(newLeft, newRight, Inner, joinConditions.reduceLeftOption(And))
}
}

case class Join(
left: LogicalPlan,
right: LogicalPlan,
joinType: JoinType,
condition: Option[Expression]) extends BinaryNode {

def references = condition.map(_.references).getOrElse(Set.empty)
def output = left.output ++ right.output
}

case class Filter(condition: Expression, child: LogicalPlan) extends UnaryNode {
def output = child.output
def references = condition.references
}

def reduceLeftOption[B >: A](op: (B, A) => B): Option =
if (isEmpty) None else Some(reduceLeft(op))

trait PredicateHelper {
def splitConjunctivePredicates(condition: Expression): Seq[Expression] = condition match {
case And(cond1, cond2) => splitConjunctivePredicates(cond1) ++ splitConjunctivePredicates(cond2)
case other => other :: Nil
}
}

case two