[SPARK-25602][SQL] SparkPlan.getByteArrayRdd should not consume the input when not necessary #22621

cloud-fan · 2018-10-03T13:52:31Z

What changes were proposed in this pull request?

In SparkPlan.getByteArrayRdd, we should only call it.hasNext when the limit is not hit, as iter.hasNext may produce one row and buffer it, and cause wrong metrics.

How was this patch tested?

new tests

cloud-fan · 2018-10-03T13:53:19Z

cc @kiszk @viirya @mgaido91

cloud-fan · 2018-10-03T13:57:27Z

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

-      s"if (shouldStop()) { $number = $value + ${step}L; return; }"
+
+    val processingLoop = if (parent.needStopCheck) {
+      // TODO (cloud-fan): do we really need to do the stop check within batch?


This is the motivation of bringing the discussion at #10989 (comment)

If it's OK to not interrupt the loop and buffer result rows for join, I think it's also OK here.

if we don't, then we would consume more rows than needed, don't we?

we just buffer more rows in BufferRowIterator.currentRows, it's only about performance IIUC.

mmmh, but localIdx would become localEnd then, right? So the UTs you added would fail, or am I missing something?

I think that there is BroadcastHashJoin case doesn't mean it is generally ok to buffer more rows. If it is possible, we still should avoid it.

cloud-fan · 2018-10-03T14:04:11Z

ok to test

mgaido91

thanks for pinging me @cloud-fan.

mgaido91 · 2018-10-03T13:58:39Z

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

-      while (iter.hasNext && (n < 0 || count < n)) {
+      // `iter.hasNext` may produce one row and buffer it, we should only call it when the limit is
+      // not hit.
+      while ((n < 0 || count < n) && iter.hasNext) {


nice catch this one!

mgaido91 · 2018-10-03T14:24:12Z

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

-      |       $shouldStop
+      |     $processingLoop
+      |   } else {
+      |     long $nextBatchTodo;


why did you move these lines in the else?

Now we don't do return when we need to interrupt the loop. Move these lines to else, so that we won't hit this code path when loop is interrupted.

I see, but in this way we are looping 2 more times in the outer loop, because we either go in the if or in the else while previously we were doing both on the same iteration IIUC. I don't think it is a big issue but it may introduce a (very small probably) overhead compared to the previous case.

Since if IIUC in the first iteration we just go to the else branch now, since batchEnd is inited to nextIndex, do you think it is feasible to move this block before the inner loop? So we would solve both issues, right?

mgaido91 · 2018-10-03T14:33:18Z

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

-      s"if (shouldStop()) { $number = $value + ${step}L; return; }"
+
+    val processingLoop = if (parent.needStopCheck) {
+      // TODO (cloud-fan): do we really need to do the stop check within batch?


if we don't, then we would consume more rows than needed, don't we?

mgaido91 · 2018-10-03T14:40:15Z

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

+      df.queryExecution.executedPlan.foreach {
+        case w: WholeStageCodegenExec =>
+          w.child.foreach {
+            case f: FilterExec => assert(f.metrics("numOutputRows").value == 10L)


not a big issue, but if later we change things and these are not anymore here, we would not run the assert here. I would suggest to collect the FilterExec and the RangeExec and enforce that we collected 1 of both and then assert on them. What do you think?

Moreover, nit: would it be possible to dedup the code here? The tests are very similar with codegen on and off, only collecting the two exec nodes differs...

viirya · 2018-10-03T15:23:05Z

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

    // and a new batch is started.
    // In the implementation below, the code in the inner loop is producing all the values
    // within a batch, while the code in the outer loop is setting batch parameters and updating
    // the metrics.


This comment should be updated too.

viirya · 2018-10-03T15:58:58Z

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

+    }
+
+    withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
+      // Top-most limit will only run the first task, so totally the Filter produces 2 rows, and


Does first task mean first partition?

viirya · 2018-10-03T16:02:08Z

retest this please.

SparkQA · 2018-10-03T23:11:59Z

Test build #96895 has finished for PR 22621 at commit 454efab.

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

cloud-fan · 2018-10-04T04:39:46Z

I simplified this PR to focus on SparkPlan.getByteArrayRdd only. Will submit PR to fix range later.

SparkQA · 2018-10-04T06:03:40Z

Test build #96920 has finished for PR 22621 at commit 1c94d13.

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

SparkQA · 2018-10-04T07:05:01Z

Test build #96924 has finished for PR 22621 at commit 3b9b41f.

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

gatorsmile · 2018-10-04T07:09:22Z

retest this please

gatorsmile

LGTM pending Jenkins

mgaido91

LGTM apart from one minor comment. Thanks

mgaido91 · 2018-10-04T09:39:10Z

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

+        df: DataFrame,
+        filterNumOutputs: Int,
+        rangeNumOutputs: Int): Unit = {
+      var filter: FilterExec = null


what about something like this:

def collectExecNode[T](pf: PartialFunction[SparkPlan, T]): PartialFunction[SparkPlan, T] = { pf.orElse { case w: WholeStageCodegenExec => w.child.collect(pf).head } } val range = df.queryExecution.executedPlan.collectFirst( collectExecNode { case r: RangeExec => r }) val filter = df.queryExecution.executedPlan.collectFirst( collectExecNode { case f: FilterExec => f })

In the future if we need to catch more nodes, we should abstract it. But for now it's only range and filter, I think it's ok.

viirya · 2018-10-04T10:15:04Z

LGTM

SparkQA · 2018-10-04T11:03:11Z

Test build #96927 has finished for PR 22621 at commit 3b9b41f.

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

cloud-fan · 2018-10-04T12:16:55Z

thanks, merging to master/2.4!

…nput when not necessary ## What changes were proposed in this pull request? In `SparkPlan.getByteArrayRdd`, we should only call `it.hasNext` when the limit is not hit, as `iter.hasNext` may produce one row and buffer it, and cause wrong metrics. ## How was this patch tested? new tests Closes #22621 from cloud-fan/range. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 71c24aa) Signed-off-by: Wenchen Fan <[email protected]>

HyukjinKwon · 2018-10-25T06:02:38Z

Let's say, this can be behaivour changes too since metrics are now changed. Should we update migration guide for safety?

cloud-fan · 2018-10-25T10:55:16Z

why do we need migration guide for bug fix?

HyukjinKwon · 2018-10-25T12:27:31Z

That's my point. Why do we have to document for fixing unexpected results fixed

…nput when not necessary ## What changes were proposed in this pull request? In `SparkPlan.getByteArrayRdd`, we should only call `it.hasNext` when the limit is not hit, as `iter.hasNext` may produce one row and buffer it, and cause wrong metrics. ## How was this patch tested? new tests Closes apache#22621 from cloud-fan/range. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

range metrics can be wrong if the result rows are not fully consumed

454efab

cloud-fan force-pushed the range branch from 01c1738 to 454efab Compare October 3, 2018 13:56

cloud-fan commented Oct 3, 2018

View reviewed changes

mgaido91 reviewed Oct 3, 2018

View reviewed changes

viirya reviewed Oct 3, 2018

View reviewed changes

cloud-fan added 4 commits October 4, 2018 10:15

address comments

1c94d13

different idea

bf0e891

revert range change

8751000

rename test

3b9b41f

cloud-fan changed the title ~~[SPARK-25602][SQL] range metrics can be wrong if the result rows are not fully consumed~~ [SPARK-25602][SQL] SparkPlan.getByteArrayRdd should not consume the input when not necessary Oct 4, 2018

gatorsmile reviewed Oct 4, 2018

View reviewed changes

mgaido91 reviewed Oct 4, 2018

View reviewed changes

asfgit closed this in 71c24aa Oct 4, 2018

[SPARK-25602][SQL] SparkPlan.getByteArrayRdd should not consume the input when not necessary #22621

[SPARK-25602][SQL] SparkPlan.getByteArrayRdd should not consume the input when not necessary #22621

Uh oh!

Conversation

cloud-fan commented Oct 3, 2018 • 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 Oct 3, 2018

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 Oct 3, 2018

Uh oh!

mgaido91 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!

cloud-fan Oct 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Choose a reason for hiding this comment

Uh oh!

viirya commented Oct 3, 2018

Uh oh!

SparkQA commented Oct 3, 2018

Uh oh!

cloud-fan commented Oct 4, 2018

Uh oh!

SparkQA commented Oct 4, 2018

Uh oh!

SparkQA commented Oct 4, 2018

Uh oh!

gatorsmile commented Oct 4, 2018

Uh oh!

gatorsmile left a comment

Choose a reason for hiding this comment

Uh oh!

mgaido91 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!

viirya commented Oct 4, 2018

Uh oh!

SparkQA commented Oct 4, 2018

Uh oh!

cloud-fan commented Oct 4, 2018

Uh oh!

HyukjinKwon commented Oct 25, 2018

Uh oh!

cloud-fan commented Oct 25, 2018

Uh oh!

cloud-fan commented Oct 3, 2018 •

edited

Loading

cloud-fan Oct 3, 2018 •

edited

Loading