在之前的文章中Spark DPP(动态分区裁剪)导致的DataSourceScanExec NullPointerException问题分析以及解决,我们直接跳过了动态代码生成失败这版本一步部分，这次我们来分析一下，SQL还是在以上提到的文章中。
分析运行完该sql，我们可以看到如下的物理计划：
我们看到FilterExec和ColumnarRoRowExec并没有在一个WholeStageCodegen 中, 这是为什么呢？
这是因为exists方法是继承自CodegenFallback Trait。
我们可以跟踪一下物理规则CollapseCodegenStages,对应的代码如下:
private def insertWholeStageCodegen(plan: SparkPlan): SparkPlan = {
plan match {
// For operators that will output domain object, do not insert WholeStageCodegen for it as
// domain object can not be written into unsafe row.
case plan if plan.output.length == 1 && plan.output.head.dataType.isInstanceOf[ObjectType] =>
plan.withNewChildren(plan.children.map(insertWholeStageCodegen))
case plan: LocalTableScanExec =>
// Do not make LogicalTableScanExec the root of WholeStageCodegen
// to support the fast driver-local collect/take paths.
plan
case plan: CodegenSupport if supportCodegen(plan) =>
// The whole-stage-codegen framework is row-based. If a plan supports columnar execution,
// it can't support whole-stage-codegen at the same time.
assert(!plan.supportsColumnar)
WholeStageCodegenExec(insertInputAdapter(plan))(codegenStageCounter.incrementAndGet())
case other =>
other.withNewChildren(other.children.map(insertWholeStageCodegen))
}
在FilterExec做supportCodegen判断的时候，还会扫描他的表达式是否存在CodegenFallback的子类，如果存在则不会把FilterExec做全代码代码生成。显然这里是不满足条件的， 而他的child，也就是ColumnarRoRowExec 他是符合的，所以ColumnarRoRowExec是可以用全代码代码生成的。
所以在driver端生成RDD的时候，FilterExec还是会走自身的doExecute方法，也就是先会运行createCodeGeneratedObject代码部分，最终还是会调用到subexpressionElimination这个方法，从而报错。
其实我们可以修改一下对应的sql，让它走全代码生成,去掉exists的过滤条件.
FROM test_b where scenes='gogo' and exists(array(date1),x-> x =='2021-03-04') -> FROM test_b where scenes='gogo'
这样我们可以得到如下的物理计划：
可以看到FilterExec和ColumnerExec是在一个WholeStageCodegen中，从代码级别来说，就是会生成类似如下的数据结构：
WholeStageCodegenExec(FilterExec(ColumnarToRowExec(InputAdapter(FileSourceScanExec))))
所在在driver端生成对应RDD的时候，就会走到WholeStageCodegenExec的doExecute方法，如下：
override def doExecute(): RDD[InternalRow] = {
val (ctx, cleanedSource) = doCodeGen()
// try to compile and fallback if it failed
val (_, compiledCodeStats) = try {
CodeGenerator.compile(cleanedSource)
} catch {
case NonFatal(_) if !Utils.isTesting && sqlContext.conf.codegenFallback =>
// We should already saw the error message
logWarning(s"Whole-stage codegen disabled for plan (id=$codegenStageId):n $treeString")
return child.execute()
}
...
val rdds = child.asInstanceOf[CodegenSupport].inputRDDs()
assert(rdds.size <= 2, "Up to two input RDDs can be supported")
if (rdds.length == 1) {
rdds.head.mapPartitionsWithIndex { (index, iter) =>
val (clazz, _) = CodeGenerator.compile(cleanedSource)
val buffer = clazz.generate(references).asInstanceOf[BufferedRowIterator]
buffer.init(index, Array(iter)
...
而 doCodeGen方法还是按照produce->doProduce->consume-doConsume>的调用方式来组织各个物理计划间的代码。
而最终还是会调用到FilterExec的doConsume代码，而这里面涉及到的DPP的表达式代码就是通过代码生成的：
override def doConsume(ctx: CodegenContext, input: Seq[ExprCode], row: ExprCode): String = {
val numOutput = metricTerm(ctx, "numOutputRows")
def genPredicate(c: expression, in: Seq[ExprCode], attrs: Seq[Attribute]): String = {
val bound = BindReferences.bindReference(c, attrs)
val evaluated = evaluateRequiredVariables(child.output, in, c.references)
// Generate the code for the predicate.
val ev = expressionCanonicalizer.execute(bound).genCode(ctx)
val nullCheck = if (bound.nullable) {
s"${ev.isNull} || "
} else {
...
通过genCode方法调用expression的 doGenCode方法。
对应到我们sql对应的表达式为DynamicPruningexpression(InSubqueryExec(value, broadcastValues, exprId)) .
而DynamicPruningexpression的还是调用child的genCode的方法，也就是最终会调用到InSubqueryExec的doGenCode方法：
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
inSet.doGenCode(ctx, ev)
}
...
private def prepareResult(): Unit = {
require(resultBroadcast != null, s"$this has not finished")
if (result == null) {
result = resultBroadcast.value
}
}
prepareResult方法是准备broadcast的值。
inSet 方法只是正常对一行row进行计算。
全代码生成完后，进行代码的编译，如下：
val rdds = child.asInstanceOf[CodegenSupport].inputRDDs()
assert(rdds.size <= 2, "Up to two input RDDs can be supported")
if (rdds.length == 1) {
rdds.head.mapPartitionsWithIndex { (index, iter) =>
val (clazz, _) = CodeGenerator.compile(cleanedSource)
val buffer = clazz.generate(references).asInstanceOf[BufferedRowIterator]
buffer.init(index, Array(iter))
这里的rdds是会调用inputRDDs方法，而该方法是递归调用child的inputRDDs方法，最终会调用到ColumnarToRowExec的inputRDDs方法，
从而调用InputAdapter的executeColumnar方法，最终会调用到FileSourceScanExec的doExecuteColumnar方法，从而生成对应的RDD。
生成的代码会在executor端再次被编译，从而进行运算。
注意：resultBroadcast的值是在WholeStageCodegenExec的方法execute中完成的：
final def execute(): RDD[InternalRow] = executeQuery {
if (isCanonicalizedPlan) {
throw new IllegalStateException("A canonicalized plan is not supposed to be executed.")
}
doExecute()
}
其中executeQuery方法如下：
protected final def executeQuery[T](query: => T): T = {
RDDOperationScope.withScope(sparkContext, nodeName, false, true) {
prepare()
waitForSubqueries()
query
}
}
...
final def prepare(): Unit = {
// doPrepare() may depend on it's children, we should call prepare() on all the children first.
children.foreach(_.prepare())
synchronized {
if (!prepared) {
prepareSubqueries()
doPrepare()
prepared = true
}
}
}
prepare和prepareSubqueries方法递归调用，从而使子节点都能把准备工作做好，如这里的driver端的广播，从而在executor端能够获取对应的广播变量。
至此，分析就完成了。
