Repository: spark Updated Branches: refs/heads/master da7bfac48 -> 25bba58d1
[SPARK-13404] [SQL] Create variables for input row when it's actually used ## What changes were proposed in this pull request? This PR change the way how we generate the code for the output variables passing from a plan to it's parent. Right now, they are generated before call consume() of it's parent. It's not efficient, if the parent is a Filter or Join, which could filter out most the rows, the time to access some of the columns that are not used by the Filter or Join are wasted. This PR try to improve this by defering the access of columns until they are actually used by a plan. After this PR, a plan does not need to generate code to evaluate the variables for output, just passing the ExprCode to its parent by `consume()`. In `parent.consumeChild()`, it will check the output from child and `usedInputs`, generate the code for those columns that is part of `usedInputs` before calling `doConsume()`. This PR also change the `if` from ``` if (cond) { xxx } ``` to ``` if (!cond) continue; xxx ``` The new one could help to reduce the nested indents for multiple levels of Filter and BroadcastHashJoin. It also added some comments for operators. ## How was the this patch tested? Unit tests. Manually ran TPCDS Q55, this PR improve the performance about 30% (scale=10, from 2.56s to 1.96s) Author: Davies Liu <dav...@databricks.com> Closes #11274 from davies/gen_defer. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/25bba58d Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/25bba58d Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/25bba58d Branch: refs/heads/master Commit: 25bba58d160d0d24e40db1ca595200a52db922ed Parents: da7bfac Author: Davies Liu <dav...@databricks.com> Authored: Mon Mar 7 20:09:08 2016 -0800 Committer: Davies Liu <davies....@gmail.com> Committed: Mon Mar 7 20:09:08 2016 -0800 ---------------------------------------------------------------------- .../catalyst/expressions/BoundAttribute.scala | 7 +- .../expressions/codegen/CodeGenerator.scala | 2 + .../spark/sql/execution/ExistingRDD.scala | 41 +++++---- .../org/apache/spark/sql/execution/Expand.scala | 4 +- .../spark/sql/execution/WholeStageCodegen.scala | 96 ++++++++++++++------ .../execution/aggregate/TungstenAggregate.scala | 50 +++++----- .../spark/sql/execution/basicOperators.scala | 32 ++++--- .../sql/execution/joins/BroadcastHashJoin.scala | 84 ++++++++--------- .../sql/execution/joins/SortMergeJoin.scala | 63 +++++++------ 9 files changed, 224 insertions(+), 155 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala ---------------------------------------------------------------------- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala index 4727ff1..72fe065 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala @@ -62,9 +62,10 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean) val javaType = ctx.javaType(dataType) val value = ctx.getValue(ctx.INPUT_ROW, dataType, ordinal.toString) if (ctx.currentVars != null && ctx.currentVars(ordinal) != null) { - ev.isNull = ctx.currentVars(ordinal).isNull - ev.value = ctx.currentVars(ordinal).value - "" + val oev = ctx.currentVars(ordinal) + ev.isNull = oev.isNull + ev.value = oev.value + oev.code } else if (nullable) { s""" boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal); http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala ---------------------------------------------------------------------- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala index 63e1956..c4265a7 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala @@ -37,6 +37,8 @@ import org.apache.spark.util.Utils * Java source for evaluating an [[Expression]] given a [[InternalRow]] of input. * * @param code The sequence of statements required to evaluate the expression. + * It should be empty string, if `isNull` and `value` are already existed, or no code + * needed to evaluate them (literals). * @param isNull A term that holds a boolean value representing whether the expression evaluated * to null. * @param value A term for a (possibly primitive) value of the result of the evaluation. Not http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala index 4ad0750..3662ed7 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala @@ -151,9 +151,6 @@ private[sql] case class PhysicalRDD( val exprs = output.zipWithIndex.map(x => new BoundReference(x._2, x._1.dataType, true)) val row = ctx.freshName("row") val numOutputRows = metricTerm(ctx, "numOutputRows") - ctx.INPUT_ROW = row - ctx.currentVars = null - val columns = exprs.map(_.gen(ctx)) // The input RDD can either return (all) ColumnarBatches or InternalRows. We determine this // by looking at the first value of the RDD and then calling the function which will process @@ -161,7 +158,9 @@ private[sql] case class PhysicalRDD( // TODO: The abstractions between this class and SqlNewHadoopRDD makes it difficult to know // here which path to use. Fix this. - + ctx.INPUT_ROW = row + ctx.currentVars = null + val columns1 = exprs.map(_.gen(ctx)) val scanBatches = ctx.freshName("processBatches") ctx.addNewFunction(scanBatches, s""" @@ -170,12 +169,11 @@ private[sql] case class PhysicalRDD( | int numRows = $batch.numRows(); | if ($idx == 0) $numOutputRows.add(numRows); | - | while ($idx < numRows) { + | while (!shouldStop() && $idx < numRows) { | InternalRow $row = $batch.getRow($idx++); - | ${columns.map(_.code).mkString("\n").trim} - | ${consume(ctx, columns).trim} - | if (shouldStop()) return; + | ${consume(ctx, columns1).trim} | } + | if (shouldStop()) return; | | if (!$input.hasNext()) { | $batch = null; @@ -186,30 +184,37 @@ private[sql] case class PhysicalRDD( | } | }""".stripMargin) + ctx.INPUT_ROW = row + ctx.currentVars = null + val columns2 = exprs.map(_.gen(ctx)) + val inputRow = if (isUnsafeRow) row else null val scanRows = ctx.freshName("processRows") ctx.addNewFunction(scanRows, s""" | private void $scanRows(InternalRow $row) throws java.io.IOException { - | while (true) { + | boolean firstRow = true; + | while (!shouldStop() && (firstRow || $input.hasNext())) { + | if (firstRow) { + | firstRow = false; + | } else { + | $row = (InternalRow) $input.next(); + | } | $numOutputRows.add(1); - | ${columns.map(_.code).mkString("\n").trim} - | ${consume(ctx, columns).trim} - | if (shouldStop()) return; - | if (!$input.hasNext()) break; - | $row = (InternalRow)$input.next(); + | ${consume(ctx, columns2, inputRow).trim} | } | }""".stripMargin) + val value = ctx.freshName("value") s""" | if ($batch != null) { | $scanBatches(); | } else if ($input.hasNext()) { - | Object value = $input.next(); - | if (value instanceof $columnarBatchClz) { - | $batch = ($columnarBatchClz)value; + | Object $value = $input.next(); + | if ($value instanceof $columnarBatchClz) { + | $batch = ($columnarBatchClz)$value; | $scanBatches(); | } else { - | $scanRows((InternalRow)value); + | $scanRows((InternalRow) $value); | } | } """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/Expand.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Expand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Expand.scala index 12998a3..524285b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Expand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/Expand.scala @@ -185,8 +185,10 @@ case class Expand( val numOutput = metricTerm(ctx, "numOutputRows") val i = ctx.freshName("i") + // these column have to declared before the loop. + val evaluate = evaluateVariables(outputColumns) s""" - |${outputColumns.map(_.code).mkString("\n").trim} + |$evaluate |for (int $i = 0; $i < ${projections.length}; $i ++) { | switch ($i) { | ${cases.mkString("\n").trim} http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala index 6d231bf..45578d5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegen.scala @@ -81,11 +81,14 @@ trait CodegenSupport extends SparkPlan { this.parent = parent ctx.freshNamePrefix = variablePrefix waitForSubqueries() - doProduce(ctx) + s""" + |/*** PRODUCE: ${toCommentSafeString(this.simpleString)} */ + |${doProduce(ctx)} + """.stripMargin } /** - * Generate the Java source code to process, should be overrided by subclass to support codegen. + * Generate the Java source code to process, should be overridden by subclass to support codegen. * * doProduce() usually generate the framework, for example, aggregation could generate this: * @@ -94,11 +97,11 @@ trait CodegenSupport extends SparkPlan { * # call child.produce() * initialized = true; * } - * while (hashmap.hasNext()) { + * while (!shouldStop() && hashmap.hasNext()) { * row = hashmap.next(); * # build the aggregation results - * # create varialbles for results - * # call consume(), wich will call parent.doConsume() + * # create variables for results + * # call consume(), which will call parent.doConsume() * } */ protected def doProduce(ctx: CodegenContext): String @@ -114,27 +117,71 @@ trait CodegenSupport extends SparkPlan { } /** - * Consume the columns generated from it's child, call doConsume() or emit the rows. + * Returns source code to evaluate all the variables, and clear the code of them, to prevent + * them to be evaluated twice. + */ + protected def evaluateVariables(variables: Seq[ExprCode]): String = { + val evaluate = variables.filter(_.code != "").map(_.code.trim).mkString("\n") + variables.foreach(_.code = "") + evaluate + } + + /** + * Returns source code to evaluate the variables for required attributes, and clear the code + * of evaluated variables, to prevent them to be evaluated twice.. */ + protected def evaluateRequiredVariables( + attributes: Seq[Attribute], + variables: Seq[ExprCode], + required: AttributeSet): String = { + var evaluateVars = "" + variables.zipWithIndex.foreach { case (ev, i) => + if (ev.code != "" && required.contains(attributes(i))) { + evaluateVars += ev.code.trim + "\n" + ev.code = "" + } + } + evaluateVars + } + + /** + * The subset of inputSet those should be evaluated before this plan. + * + * We will use this to insert some code to access those columns that are actually used by current + * plan before calling doConsume(). + */ + def usedInputs: AttributeSet = references + + /** + * Consume the columns generated from its child, call doConsume() or emit the rows. + * + * An operator could generate variables for the output, or a row, either one could be null. + * + * If the row is not null, we create variables to access the columns that are actually used by + * current plan before calling doConsume(). + */ def consumeChild( ctx: CodegenContext, child: SparkPlan, input: Seq[ExprCode], row: String = null): String = { ctx.freshNamePrefix = variablePrefix - if (row != null) { - ctx.currentVars = null - ctx.INPUT_ROW = row - val evals = child.output.zipWithIndex.map { case (attr, i) => - BoundReference(i, attr.dataType, attr.nullable).gen(ctx) + val inputVars = + if (row != null) { + ctx.currentVars = null + ctx.INPUT_ROW = row + child.output.zipWithIndex.map { case (attr, i) => + BoundReference(i, attr.dataType, attr.nullable).gen(ctx) + } + } else { + input } - s""" - | ${evals.map(_.code).mkString("\n")} - | ${doConsume(ctx, evals)} - """.stripMargin - } else { - doConsume(ctx, input) - } + s""" + | + |/*** CONSUME: ${toCommentSafeString(this.simpleString)} */ + |${evaluateRequiredVariables(child.output, inputVars, usedInputs)} + |${doConsume(ctx, inputVars)} + """.stripMargin } /** @@ -145,9 +192,8 @@ trait CodegenSupport extends SparkPlan { * For example, Filter will generate the code like this: * * # code to evaluate the predicate expression, result is isNull1 and value2 - * if (isNull1 || value2) { - * # call consume(), which will call parent.doConsume() - * } + * if (isNull1 || !value2) continue; + * # call consume(), which will call parent.doConsume() */ protected def doConsume(ctx: CodegenContext, input: Seq[ExprCode]): String = { throw new UnsupportedOperationException @@ -190,13 +236,9 @@ case class InputAdapter(child: SparkPlan) extends UnaryNode with CodegenSupport ctx.currentVars = null val columns = exprs.map(_.gen(ctx)) s""" - | while ($input.hasNext()) { + | while (!shouldStop() && $input.hasNext()) { | InternalRow $row = (InternalRow) $input.next(); - | ${columns.map(_.code).mkString("\n").trim} | ${consume(ctx, columns).trim} - | if (shouldStop()) { - | return; - | } | } """.stripMargin } @@ -332,10 +374,12 @@ case class WholeStageCodegen(child: SparkPlan) extends UnaryNode with CodegenSup val colExprs = output.zipWithIndex.map { case (attr, i) => BoundReference(i, attr.dataType, attr.nullable) } + val evaluateInputs = evaluateVariables(input) // generate the code to create a UnsafeRow ctx.currentVars = input val code = GenerateUnsafeProjection.createCode(ctx, colExprs, false) s""" + |$evaluateInputs |${code.code.trim} |append(${code.value}.copy()); """.stripMargin.trim http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala index a467229..f07add8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala @@ -116,6 +116,8 @@ case class TungstenAggregate( // all the mode of aggregate expressions private val modes = aggregateExpressions.map(_.mode).distinct + override def usedInputs: AttributeSet = inputSet + override def supportCodegen: Boolean = { // ImperativeAggregate is not supported right now !aggregateExpressions.exists(_.aggregateFunction.isInstanceOf[ImperativeAggregate]) @@ -164,23 +166,24 @@ case class TungstenAggregate( """.stripMargin ExprCode(ev.code + initVars, isNull, value) } + val initBufVar = evaluateVariables(bufVars) // generate variables for output - val bufferAttrs = functions.flatMap(_.aggBufferAttributes) val (resultVars, genResult) = if (modes.contains(Final) || modes.contains(Complete)) { // evaluate aggregate results ctx.currentVars = bufVars val aggResults = functions.map(_.evaluateExpression).map { e => - BindReferences.bindReference(e, bufferAttrs).gen(ctx) + BindReferences.bindReference(e, aggregateBufferAttributes).gen(ctx) } + val evaluateAggResults = evaluateVariables(aggResults) // evaluate result expressions ctx.currentVars = aggResults val resultVars = resultExpressions.map { e => BindReferences.bindReference(e, aggregateAttributes).gen(ctx) } (resultVars, s""" - | ${aggResults.map(_.code).mkString("\n")} - | ${resultVars.map(_.code).mkString("\n")} + |$evaluateAggResults + |${evaluateVariables(resultVars)} """.stripMargin) } else if (modes.contains(Partial) || modes.contains(PartialMerge)) { // output the aggregate buffer directly @@ -188,7 +191,7 @@ case class TungstenAggregate( } else { // no aggregate function, the result should be literals val resultVars = resultExpressions.map(_.gen(ctx)) - (resultVars, resultVars.map(_.code).mkString("\n")) + (resultVars, evaluateVariables(resultVars)) } val doAgg = ctx.freshName("doAggregateWithoutKey") @@ -196,7 +199,7 @@ case class TungstenAggregate( s""" | private void $doAgg() throws java.io.IOException { | // initialize aggregation buffer - | ${bufVars.map(_.code).mkString("\n")} + | $initBufVar | | ${child.asInstanceOf[CodegenSupport].produce(ctx, this)} | } @@ -204,7 +207,7 @@ case class TungstenAggregate( val numOutput = metricTerm(ctx, "numOutputRows") s""" - | if (!$initAgg) { + | while (!$initAgg) { | $initAgg = true; | $doAgg(); | @@ -241,7 +244,7 @@ case class TungstenAggregate( } s""" | // do aggregate - | ${aggVals.map(_.code).mkString("\n").trim} + | ${evaluateVariables(aggVals)} | // update aggregation buffer | ${updates.mkString("\n").trim} """.stripMargin @@ -252,8 +255,7 @@ case class TungstenAggregate( private val declFunctions = aggregateExpressions.map(_.aggregateFunction) .filter(_.isInstanceOf[DeclarativeAggregate]) .map(_.asInstanceOf[DeclarativeAggregate]) - private val bufferAttributes = declFunctions.flatMap(_.aggBufferAttributes) - private val bufferSchema = StructType.fromAttributes(bufferAttributes) + private val bufferSchema = StructType.fromAttributes(aggregateBufferAttributes) // The name for HashMap private var hashMapTerm: String = _ @@ -318,7 +320,7 @@ case class TungstenAggregate( val mergeExpr = declFunctions.flatMap(_.mergeExpressions) val mergeProjection = newMutableProjection( mergeExpr, - bufferAttributes ++ declFunctions.flatMap(_.inputAggBufferAttributes), + aggregateBufferAttributes ++ declFunctions.flatMap(_.inputAggBufferAttributes), subexpressionEliminationEnabled)() val joinedRow = new JoinedRow() @@ -380,15 +382,18 @@ case class TungstenAggregate( val keyVars = groupingExpressions.zipWithIndex.map { case (e, i) => BoundReference(i, e.dataType, e.nullable).gen(ctx) } + val evaluateKeyVars = evaluateVariables(keyVars) ctx.INPUT_ROW = bufferTerm - val bufferVars = bufferAttributes.zipWithIndex.map { case (e, i) => + val bufferVars = aggregateBufferAttributes.zipWithIndex.map { case (e, i) => BoundReference(i, e.dataType, e.nullable).gen(ctx) } + val evaluateBufferVars = evaluateVariables(bufferVars) // evaluate the aggregation result ctx.currentVars = bufferVars val aggResults = declFunctions.map(_.evaluateExpression).map { e => - BindReferences.bindReference(e, bufferAttributes).gen(ctx) + BindReferences.bindReference(e, aggregateBufferAttributes).gen(ctx) } + val evaluateAggResults = evaluateVariables(aggResults) // generate the final result ctx.currentVars = keyVars ++ aggResults val inputAttrs = groupingAttributes ++ aggregateAttributes @@ -396,11 +401,9 @@ case class TungstenAggregate( BindReferences.bindReference(e, inputAttrs).gen(ctx) } s""" - ${keyVars.map(_.code).mkString("\n")} - ${bufferVars.map(_.code).mkString("\n")} - ${aggResults.map(_.code).mkString("\n")} - ${resultVars.map(_.code).mkString("\n")} - + $evaluateKeyVars + $evaluateBufferVars + $evaluateAggResults ${consume(ctx, resultVars)} """ @@ -422,10 +425,7 @@ case class TungstenAggregate( val eval = resultExpressions.map{ e => BindReferences.bindReference(e, groupingAttributes).gen(ctx) } - s""" - ${eval.map(_.code).mkString("\n")} - ${consume(ctx, eval)} - """ + consume(ctx, eval) } } @@ -508,8 +508,8 @@ case class TungstenAggregate( ctx.currentVars = input val hashEval = BindReferences.bindReference(hashExpr, child.output).gen(ctx) - val inputAttr = bufferAttributes ++ child.output - ctx.currentVars = new Array[ExprCode](bufferAttributes.length) ++ input + val inputAttr = aggregateBufferAttributes ++ child.output + ctx.currentVars = new Array[ExprCode](aggregateBufferAttributes.length) ++ input ctx.INPUT_ROW = buffer // TODO: support subexpression elimination val evals = updateExpr.map(BindReferences.bindReference(_, inputAttr).gen(ctx)) @@ -557,7 +557,7 @@ case class TungstenAggregate( $incCounter // evaluate aggregate function - ${evals.map(_.code).mkString("\n").trim} + ${evaluateVariables(evals)} // update aggregate buffer ${updates.mkString("\n").trim} """ http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala index b2f443c..4a9e736 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicOperators.scala @@ -39,15 +39,26 @@ case class Project(projectList: Seq[NamedExpression], child: SparkPlan) child.asInstanceOf[CodegenSupport].produce(ctx, this) } + override def usedInputs: AttributeSet = { + // only the attributes those are used at least twice should be evaluated before this plan, + // otherwise we could defer the evaluation until output attribute is actually used. + val usedExprIds = projectList.flatMap(_.collect { + case a: Attribute => a.exprId + }) + val usedMoreThanOnce = usedExprIds.groupBy(id => id).filter(_._2.size > 1).keySet + references.filter(a => usedMoreThanOnce.contains(a.exprId)) + } + override def doConsume(ctx: CodegenContext, input: Seq[ExprCode]): String = { val exprs = projectList.map(x => ExpressionCanonicalizer.execute(BindReferences.bindReference(x, child.output))) ctx.currentVars = input - val output = exprs.map(_.gen(ctx)) + val resultVars = exprs.map(_.gen(ctx)) + // Evaluation of non-deterministic expressions can't be deferred. + val nonDeterministicAttrs = projectList.filterNot(_.deterministic).map(_.toAttribute) s""" - | ${output.map(_.code).mkString("\n")} - | - | ${consume(ctx, output)} + |${evaluateRequiredVariables(output, resultVars, AttributeSet(nonDeterministicAttrs))} + |${consume(ctx, resultVars)} """.stripMargin } @@ -89,11 +100,10 @@ case class Filter(condition: Expression, child: SparkPlan) extends UnaryNode wit s"" } s""" - | ${eval.code} - | if ($nullCheck ${eval.value}) { - | $numOutput.add(1); - | ${consume(ctx, ctx.currentVars)} - | } + |${eval.code} + |if (!($nullCheck ${eval.value})) continue; + |$numOutput.add(1); + |${consume(ctx, ctx.currentVars)} """.stripMargin } @@ -228,15 +238,13 @@ case class Range( | } | } | - | while (!$overflow && $checkEnd) { + | while (!$overflow && $checkEnd && !shouldStop()) { | long $value = $number; | $number += ${step}L; | if ($number < $value ^ ${step}L < 0) { | $overflow = true; | } | ${consume(ctx, Seq(ev))} - | - | if (shouldStop()) return; | } """.stripMargin } http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala index 6699dba..c52662a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoin.scala @@ -190,40 +190,38 @@ case class BroadcastHashJoin( val (keyEv, anyNull) = genStreamSideJoinKey(ctx, input) val matched = ctx.freshName("matched") val buildVars = genBuildSideVars(ctx, matched) - val resultVars = buildSide match { - case BuildLeft => buildVars ++ input - case BuildRight => input ++ buildVars - } val numOutput = metricTerm(ctx, "numOutputRows") - val outputCode = if (condition.isDefined) { + val checkCondition = if (condition.isDefined) { + val expr = condition.get + // evaluate the variables from build side that used by condition + val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references) // filter the output via condition - ctx.currentVars = resultVars - val ev = BindReferences.bindReference(condition.get, this.output).gen(ctx) + ctx.currentVars = input ++ buildVars + val ev = BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).gen(ctx) s""" + |$eval |${ev.code} - |if (!${ev.isNull} && ${ev.value}) { - | $numOutput.add(1); - | ${consume(ctx, resultVars)} - |} + |if (${ev.isNull} || !${ev.value}) continue; """.stripMargin } else { - s""" - |$numOutput.add(1); - |${consume(ctx, resultVars)} - """.stripMargin + "" } + val resultVars = buildSide match { + case BuildLeft => buildVars ++ input + case BuildRight => input ++ buildVars + } if (broadcastRelation.value.isInstanceOf[UniqueHashedRelation]) { s""" |// generate join key for stream side |${keyEv.code} |// find matches from HashedRelation |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value}); - |if ($matched != null) { - | ${buildVars.map(_.code).mkString("\n")} - | $outputCode - |} + |if ($matched == null) continue; + |$checkCondition + |$numOutput.add(1); + |${consume(ctx, resultVars)} """.stripMargin } else { @@ -236,13 +234,13 @@ case class BroadcastHashJoin( |${keyEv.code} |// find matches from HashRelation |$bufferType $matches = $anyNull ? null : ($bufferType)$relationTerm.get(${keyEv.value}); - |if ($matches != null) { - | int $size = $matches.size(); - | for (int $i = 0; $i < $size; $i++) { - | UnsafeRow $matched = (UnsafeRow) $matches.apply($i); - | ${buildVars.map(_.code).mkString("\n")} - | $outputCode - | } + |if ($matches == null) continue; + |int $size = $matches.size(); + |for (int $i = 0; $i < $size; $i++) { + | UnsafeRow $matched = (UnsafeRow) $matches.apply($i); + | $checkCondition + | $numOutput.add(1); + | ${consume(ctx, resultVars)} |} """.stripMargin } @@ -257,21 +255,21 @@ case class BroadcastHashJoin( val (keyEv, anyNull) = genStreamSideJoinKey(ctx, input) val matched = ctx.freshName("matched") val buildVars = genBuildSideVars(ctx, matched) - val resultVars = buildSide match { - case BuildLeft => buildVars ++ input - case BuildRight => input ++ buildVars - } val numOutput = metricTerm(ctx, "numOutputRows") // filter the output via condition val conditionPassed = ctx.freshName("conditionPassed") val checkCondition = if (condition.isDefined) { - ctx.currentVars = resultVars - val ev = BindReferences.bindReference(condition.get, this.output).gen(ctx) + val expr = condition.get + // evaluate the variables from build side that used by condition + val eval = evaluateRequiredVariables(buildPlan.output, buildVars, expr.references) + ctx.currentVars = input ++ buildVars + val ev = BindReferences.bindReference(expr, streamedPlan.output ++ buildPlan.output).gen(ctx) s""" |boolean $conditionPassed = true; + |${eval.trim} + |${ev.code} |if ($matched != null) { - | ${ev.code} | $conditionPassed = !${ev.isNull} && ${ev.value}; |} """.stripMargin @@ -279,17 +277,21 @@ case class BroadcastHashJoin( s"final boolean $conditionPassed = true;" } + val resultVars = buildSide match { + case BuildLeft => buildVars ++ input + case BuildRight => input ++ buildVars + } if (broadcastRelation.value.isInstanceOf[UniqueHashedRelation]) { s""" |// generate join key for stream side |${keyEv.code} |// find matches from HashedRelation |UnsafeRow $matched = $anyNull ? null: (UnsafeRow)$relationTerm.getValue(${keyEv.value}); - |${buildVars.map(_.code).mkString("\n")} |${checkCondition.trim} |if (!$conditionPassed) { - | // reset to null - | ${buildVars.map(v => s"${v.isNull} = true;").mkString("\n")} + | $matched = null; + | // reset the variables those are already evaluated. + | ${buildVars.filter(_.code == "").map(v => s"${v.isNull} = true;").mkString("\n")} |} |$numOutput.add(1); |${consume(ctx, resultVars)} @@ -311,13 +313,11 @@ case class BroadcastHashJoin( |// the last iteration of this loop is to emit an empty row if there is no matched rows. |for (int $i = 0; $i <= $size; $i++) { | UnsafeRow $matched = $i < $size ? (UnsafeRow) $matches.apply($i) : null; - | ${buildVars.map(_.code).mkString("\n")} | ${checkCondition.trim} - | if ($conditionPassed && ($i < $size || !$found)) { - | $found = true; - | $numOutput.add(1); - | ${consume(ctx, resultVars)} - | } + | if (!$conditionPassed || ($i == $size && $found)) continue; + | $found = true; + | $numOutput.add(1); + | ${consume(ctx, resultVars)} |} """.stripMargin } http://git-wip-us.apache.org/repos/asf/spark/blob/25bba58d/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala ---------------------------------------------------------------------- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala index 7ec4027..cffd6f6 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala @@ -306,11 +306,11 @@ case class SortMergeJoin( val (used, notUsed) = attributes.zip(variables).partition{ case (a, ev) => condRefs.contains(a) } - val beforeCond = used.map(_._2.code).mkString("\n") - val afterCond = notUsed.map(_._2.code).mkString("\n") + val beforeCond = evaluateVariables(used.map(_._2)) + val afterCond = evaluateVariables(notUsed.map(_._2)) (beforeCond, afterCond) } else { - (variables.map(_.code).mkString("\n"), "") + (evaluateVariables(variables), "") } } @@ -326,41 +326,48 @@ case class SortMergeJoin( val leftVars = createLeftVars(ctx, leftRow) val rightRow = ctx.freshName("rightRow") val rightVars = createRightVar(ctx, rightRow) - val resultVars = leftVars ++ rightVars - - // Check condition - ctx.currentVars = resultVars - val cond = if (condition.isDefined) { - BindReferences.bindReference(condition.get, output).gen(ctx) - } else { - ExprCode("", "false", "true") - } - // Split the code of creating variables based on whether it's used by condition or not. - val loaded = ctx.freshName("loaded") - val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars) - val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars) - val size = ctx.freshName("size") val i = ctx.freshName("i") val numOutput = metricTerm(ctx, "numOutputRows") + val (beforeLoop, condCheck) = if (condition.isDefined) { + // Split the code of creating variables based on whether it's used by condition or not. + val loaded = ctx.freshName("loaded") + val (leftBefore, leftAfter) = splitVarsByCondition(left.output, leftVars) + val (rightBefore, rightAfter) = splitVarsByCondition(right.output, rightVars) + // Generate code for condition + ctx.currentVars = leftVars ++ rightVars + val cond = BindReferences.bindReference(condition.get, output).gen(ctx) + // evaluate the columns those used by condition before loop + val before = s""" + |boolean $loaded = false; + |$leftBefore + """.stripMargin + + val checking = s""" + |$rightBefore + |${cond.code} + |if (${cond.isNull} || !${cond.value}) continue; + |if (!$loaded) { + | $loaded = true; + | $leftAfter + |} + |$rightAfter + """.stripMargin + (before, checking) + } else { + (evaluateVariables(leftVars), "") + } + s""" |while (findNextInnerJoinRows($leftInput, $rightInput)) { | int $size = $matches.size(); - | boolean $loaded = false; - | $leftBefore + | ${beforeLoop.trim} | for (int $i = 0; $i < $size; $i ++) { | InternalRow $rightRow = (InternalRow) $matches.get($i); - | $rightBefore - | ${cond.code} - | if (${cond.isNull} || !${cond.value}) continue; - | if (!$loaded) { - | $loaded = true; - | $leftAfter - | } - | $rightAfter + | ${condCheck.trim} | $numOutput.add(1); - | ${consume(ctx, resultVars)} + | ${consume(ctx, leftVars ++ rightVars)} | } | if (shouldStop()) return; |} --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org