[SPARK-28213][SQL][followup] code cleanup and bug fix for columnar execution framework #25264

cloud-fan · 2019-07-26T16:18:16Z

What changes were proposed in this pull request?

I did a post-hoc review of #25008 , and would like to propose some cleanups/fixes/improvements:

Do not track the scanTime metrics in ColumnarToRowExec. This metrics is specific to file scan, and doesn't make sense for a general batch-to-row operator.
Because of 2, we need to track scanTime when building RDDs in the file scan node.
use RDD#mapPartitionsInternal instead of flatMap in several places, as mapPartitionsInternal is created for Spark SQL and we use it in almost all the SQL operators.
Add limitNotReachedCond in ColumnarToRowExec. This was in the ColumnarBatchScan before and is critical for performance.
Clear the relationship between codegen stage and columnar stage. The whole-stage-codegen framework is completely row-based, so these 2 kinds of stages can NEVER overlap. When they are adjacent, it's either a RowToColumnarExec above WholeStageExec, or a ColumnarToRowExec above the InputAdapter.
Reuse the ColumnarBatch in RowToColumnarExec. We don't need to create a new one every time, just need to reset it.
Do not skip testing full scan node in LogicalPlanTagInSparkPlanSuite
Add back the removed tests in WholeStageCodegenSuite.

How was this patch tested?

existing tests

cloud-fan · 2019-07-26T16:20:17Z

cc @revans2 @tgravescs @hvanhovell @HyukjinKwon @dongjoon-hyun @viirya @gatorsmile

revans2

Thanks for cleaning up after my patch. I really appreciate it.

I just had one comment about a metric you were removing that I think is still useful.

revans2 · 2019-07-26T19:54:09Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

+  protected override def canCheckLimitNotReached: Boolean = true
+
  override lazy val metrics: Map[String, SQLMetric] = Map(
-    "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows"),


I find num output rows to be useful because ColumnarToRowExec can happen at other times too, I am working on getting it to happen after pandas UDF operations. Plus the performance impact is only on the order of the number of batches. Not on the order of the number of rows, so it should have minimal impact.

Maybe I was too conservative. I'll add it back, and revisit this when I benchmark Spark 3.0.

SparkQA · 2019-07-26T21:40:15Z

Test build #108228 has finished for PR 25264 at commit 3afa85b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ColumnarToRowExec(child: SparkPlan) extends UnaryExecNode with CodegenSupport
case class InputAdapter(child: SparkPlan) extends UnaryExecNode with InputRDDCodegen

gatorsmile · 2019-07-27T04:05:47Z

cc @rednaxelafx Please post your comment about the changes. Thanks!

SparkQA · 2019-07-27T07:05:01Z

Test build #108242 has finished for PR 25264 at commit 816f079.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

rednaxelafx

Question on making ColumarToRowExec effectively a leaf in a codegen stage:

rednaxelafx · 2019-07-27T07:18:05Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

+          child => InputAdapter(insertWholeStageCodegen(child))))
+      // `ColumnarToRowExec` is kind of a leaf node to whole-stage-codegen. Its generated code can
+      // process data from the input RDD directly.
+      case c: ColumnarToRowExec => c


Does this work well with the WholeStageCodegenExec.treeString? i.e. does this work well with printing the * (id) prefix?

viirya

This change isn't small, and maybe we need a new JIRA?

viirya · 2019-07-27T14:59:34Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

-      case p =>
-        p.withNewChildren(p.children.map(insertInputAdapter(_, isColumnar)))
+          child => InputAdapter(insertWholeStageCodegen(child))))
+      // `ColumnarToRowExec` is kind of a leaf node to whole-stage-codegen. Its generated code can


If you want ColumnarToRowExec to be leaf node without InputAdapter, should we move it before case p if !supportCodegen(p) =>? Otherwise, isn't InputAdapter still be added between ColumnarToRowExec and p if p doesn't support codegen?

case p if !supportCodegen(p) won't match ColumnarToRowExec, so it doesn't matter where we put the case c: ColumnarToRowExec. I just want to be consistent with the existing code style and put it after the case j: SortMergeJoinExec

viirya · 2019-07-27T15:01:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

  }

  override def doExecuteColumnar(): RDD[ColumnarBatch] = {
    child.executeColumnar()


InputAdapter doesn't support columnar execution now? Seems we can change supportsColumnar and remove doExecuteColumnar?

viirya · 2019-07-28T07:12:00Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

+          child => InputAdapter(insertWholeStageCodegen(child))))
+      // `ColumnarToRowExec` is kind of a leaf node to whole-stage-codegen. Its generated code can
+      // process data from the input RDD directly.
+      case c: ColumnarToRowExec => c


Shall we recursively call insertInputAdapter on ColumnarToRowExec's children?

no, because ColumnarToRowExec is leaf node of the codegen stage. But you remind me that we should call insertWholeStageCodegen.

oh, yes, it should be insertWholeStageCodegen.

kiszk · 2019-07-28T09:41:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

-          if (cb != null) {
-            cb.close()
-            cb = null
+    // This avoids calling `output` in the RDD closure, so that we don't need to include the entire


nit: output -> schema?

cloud-fan · 2019-07-29T04:15:18Z

After more thoughts, I think we should still insert InputAdapter under ColumnarToRowExec. But we can simplify the whole-stage-codegen planning logic a bit, because codegen stage can never overlap with a columnar stage, as whole-stage-codegen is completely row-based. When these 2 stages are adjacent, it's either a RowToColumnarExec above WholeStageExec, or a ColumnarToRowExec above the InputAdapter.

SparkQA · 2019-07-29T06:32:34Z

Test build #108290 has finished for PR 25264 at commit ec2a2b8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-07-30T07:05:01Z

Test build #108371 has finished for PR 25264 at commit 6bcbcc6.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-07-30T07:33:39Z

retest this please

SparkQA · 2019-07-30T09:01:08Z

Test build #108375 has finished for PR 25264 at commit 6bcbcc6.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-07-30T09:07:40Z

retest this please

SparkQA · 2019-07-30T12:55:23Z

Test build #108380 has finished for PR 25264 at commit 6bcbcc6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tgravescs · 2019-07-31T19:46:40Z

@cloud-fan are you going to update InputAdapter supportsColumnar and doExecuteColumnar?

cloud-fan · 2019-08-01T06:29:51Z

@tgravescs no I'm not going to. It's correct that InputAdapter can support columnar execution if its child supports, by calling child.doExecuteColumnar.

tgravescs · 2019-08-01T14:03:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala

+      // The `FileScanRDD` returns an iterator which scans the file during the `hasNext` call.
+      val startNs = System.nanoTime()
+      val re = fileScanIterator.hasNext
+      scanTimeMetrics += ((System.nanoTime() - startNs) / (1000 * 1000))


NANOSECONDS.toMillis

tgravescs · 2019-08-01T17:57:06Z

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

-  }
+  // `InputAdapter` can only generate code to process the rows from its child. If the child produces
+  // columnar batches, there must be a `ColumnarToRowExec` above `InputAdapter` to handle it by
+  // overriding `inputRDD`.


I had misread/misunderstood this originally.

we are overriding "inputRDDs" (note the s) in ColumnarToRowExec which is bypassing calling inputRDD here and calling InputAdapter.executeColumnar from ColumnarToRowExec

So the ColumnarToRowExec doubles as in InputAdapter? Wouldn't it be less confusing to replace the InputAdapter with ColumnarToRowExec?

I tried the idea of replace the InputAdapter with ColumnarToRowExec before. But then it's a little weird to see ColumnarToRowExec act as both the boundary of columnar execution stage and codegen stage. We need to

handle ColumnarToRowExec specially when planning whole-stage-codegen

implement ColumnarToRowExec.treeString, which should print this columnar node and remove the whole-stage-codegen mark from its child's treeString

I feel it's simpler to always let InputAdapter be the boundary of codegen stage.

SparkQA · 2019-08-01T18:42:18Z

Test build #108524 has finished for PR 25264 at commit af177aa.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell · 2019-08-01T19:09:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

-    })
+    // This avoids calling `output` in the RDD closure, so that we don't need to include the entire
+    // plan (this) in the closure.
+    val localOutput = this.output


hvanhovell · 2019-08-01T19:10:38Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

+    // plan (this) in the closure.
+    val localOutput = this.output
+    child.executeColumnar().mapPartitionsInternal { batches =>
+      val outputProject = UnsafeProjection.create(localOutput, localOutput)


NIT: maybe name this toUnsafe to better convey the intent of the projection

hvanhovell · 2019-08-01T19:31:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala

+      scanTimeMetrics: SQLMetric) extends Iterator[T] {
+
+    override def hasNext: Boolean = {
+      // The `FileScanRDD` returns an iterator which scans the file during the `hasNext` call.


Calling System.nanoTime() per tuple isn't exactly cheap. This probably regresses cases where the scan produces rows instead of batches. I am actually not sure if there is an (easy) way around this.

The only workaround is that we don't expose scan time for row based formats.

SparkQA · 2019-08-02T07:05:02Z

Test build #108543 has finished for PR 25264 at commit 308fc11.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2019-08-02T07:20:50Z

retest this please

SparkQA · 2019-08-02T11:01:08Z

Test build #108559 has finished for PR 25264 at commit 308fc11.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tgravescs

+1

cloud-fan · 2019-08-06T02:11:35Z

thanks for the review, merging to master!

…side in ColumnarToRowExec https://issues.apache.org/jira/browse/SPARK-52484 ### What changes were proposed in this pull request? The PR removes the unnecessary assertion in `ColumnarToRowExec` introduced by #25264 to guarantee some flexibilities for 3rd Spark plugins. Especially in Apache Gluten, the assertion blocks some of our effort in query optimization because we needed an intermediate state of the query plan which Spark may see as illegal. Moreover, some typical reasons this intermediate state is needed in Gluten are: 1. Gluten has a cost evaluator API to evaluate the cost of a `transition rule` (which adds a unary node on top of an input plan). In the case Gluten will need a fake leaf to let the rule apply on it for cost evaluation. This leaf node has to be made a columnar one to bypass this assertion, which is a bit hacky. 2. Gluten has a cascades-style query optimizer (RAS) which could set a leaf, dummy, row-based plan node to hide up a child-tree of a brach query plan node, during which this leaf is to represent a so-called cascades 'group'. Although this pattern (C2R on a row-based plan) is illegal, it could still be used as the input of an optimizer rule to potentially be matched on and then to be converted into a valid query plan. This PR is to remove the assertion to ensure some flexibilities to the 3rd plugins. This should be no harm for the upstream Apache Spark, because the query execution will still be failed by [this error](https://github.com/apache/spark/blob/5d0b2f41794bf4dd25b3ce19bc4f634082b40876/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala#L343-L351) without this assertion on an illegal query plan. Some workarounds used by Gluten for bypassing this assertion: 1. https://github.com/apache/incubator-gluten/blob/0a1b5c28678653242ab0fd7b28ebba1dca43ccb1/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/transition/package.scala#L83 2. https://github.com/apache/incubator-gluten/blob/0a1b5c28678653242ab0fd7b28ebba1dca43ccb1/gluten-core/src/main/scala/org/apache/gluten/extension/columnar/enumerated/planner/plan/GlutenPlanModel.scala#L51-L55 Once the assertion is removed, Gluten will be able to remove these workarounds to simply code. ### Does this PR introduce _any_ user-facing change? Basically no. An assertion error in plan-building time will be replaced by an exception in execution time (still from the driver side) when an illegal query plan is generated. ### How was this patch tested? Existing UTs. Closes #51183 from zhztheplayer/wip-rm-c2r-check. Authored-by: Hongze Zhang <[email protected]> Signed-off-by: Kent Yao <[email protected]>

gatorsmile changed the title ~~[SPARK-28213][SQL][followup] code cleanup for columnar execution framework~~ [SPARK-28213][SQL][followup] code cleanup and bug fix for columnar execution framework Jul 26, 2019

dongjoon-hyun added the SQL label Jul 26, 2019

revans2 reviewed Jul 26, 2019

View reviewed changes

rednaxelafx reviewed Jul 27, 2019

View reviewed changes

viirya reviewed Jul 27, 2019

View reviewed changes

viirya reviewed Jul 28, 2019

View reviewed changes

kiszk reviewed Jul 28, 2019

View reviewed changes

cloud-fan added 3 commits July 29, 2019 10:21

code cleanup for columnar execution framework

867d94a

add back numOutputRows metrics

009d760

address comments

ec2a2b8

cloud-fan force-pushed the minor branch from 816f079 to ec2a2b8 Compare July 29, 2019 04:08

HyukjinKwon mentioned this pull request Jul 29, 2019

[SPARK-28213][SQL] Replace ColumnarBatchScan with equivilant from Columnar #25008

Closed

fix test

6bcbcc6

tgravescs reviewed Aug 1, 2019

View reviewed changes

address comments

af177aa

tgravescs reviewed Aug 1, 2019

View reviewed changes

hvanhovell reviewed Aug 1, 2019

View reviewed changes

address comments

308fc11

tgravescs approved these changes Aug 2, 2019

View reviewed changes

cloud-fan closed this in 03e3006 Aug 6, 2019

HyukjinKwon mentioned this pull request Aug 6, 2019

[SPARK-28537][SQL][HOTFIX][FOLLOW-UP] Add supportColumnar in DebugExec #25365

Closed

zhztheplayer mentioned this pull request Jun 16, 2025

[SPARK-52484][SQL] Skip child.supportsColumnar assertion from driver side in ColumnarToRowExec #51183

Closed

[SPARK-28213][SQL][followup] code cleanup and bug fix for columnar execution framework #25264

[SPARK-28213][SQL][followup] code cleanup and bug fix for columnar execution framework #25264

Uh oh!

Conversation

cloud-fan commented Jul 26, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

cloud-fan commented Jul 26, 2019

Uh oh!

revans2 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 26, 2019

Uh oh!

gatorsmile commented Jul 27, 2019

Uh oh!

SparkQA commented Jul 27, 2019

Uh oh!

rednaxelafx left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jul 29, 2019

Uh oh!

SparkQA commented Jul 29, 2019

Uh oh!

SparkQA commented Jul 30, 2019

Uh oh!

cloud-fan commented Jul 30, 2019

Uh oh!

SparkQA commented Jul 30, 2019

Uh oh!

cloud-fan commented Jul 30, 2019

Uh oh!

SparkQA commented Jul 30, 2019

Uh oh!

tgravescs commented Jul 31, 2019

Uh oh!

cloud-fan commented Aug 1, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 1, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jul 26, 2019 •

edited

Loading