Skip to content

[SPARK-13306] [SQL] uncorrelated scalar subquery #11190

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
davies wants to merge 9 commits into from
Closed

[SPARK-13306] [SQL] uncorrelated scalar subquery #11190

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
0665a69
uncorrelated scalar subquery
Feb 12, 2016
236ac88
add missing file
Feb 12, 2016
016c36c
use broadcastTimeout
Feb 12, 2016
a4bae33
address comments
Feb 13, 2016
d0974cf
improve explain on subquery
Feb 20, 2016
3a8f08d
address comments
Feb 20, 2016
7596173
add tests and codegen
Feb 20, 2016
0034172
move wait subqueries into execute()/produce()
Feb 20, 2016
e082845
address comments
Feb 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
use broadcastTimeout
  • Loading branch information
Davies Liu committed Feb 12, 2016
commit 016c36c316e0fe2a544646291ba3e4573aa4bb4c
13 changes: 11 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import java.util.concurrent.atomic.AtomicBoolean

import scala.collection.mutable.ArrayBuffer
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration.Duration
import scala.concurrent.duration._

import org.apache.spark.Logging
import org.apache.spark.rdd.{RDD, RDDOperationScope}
Expand Down Expand Up @@ -139,10 +139,19 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ

children.foreach(_.prepare())

val timeout: Duration = {
val timeoutValue = sqlContext.conf.broadcastTimeout
if (timeoutValue < 0) {
Duration.Inf
} else {
timeoutValue.seconds
}
}

// fill in the result of subqueries
queryResults.foreach {
Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should move the blocking phase into execute, otherwise if multiple nodes have subqueries, it becomes blocking.

Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah ok you can't have a general execute.

I guess this is why some query engines have init and then prepare.

Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

or a subquery is now blocking broadcasting ...

Copy link
Contributor Author

@davies davies Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is called after doPrepare(), and after prepare() of it's children, so it will NOT block broadcasting (will happen in the same time).

Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what if there is a broadcast join after this?

Copy link
Contributor Author

@davies davies Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Broadcast will be issued before this.

case (e, futureResult) =>
val rows = Await.result(futureResult, Duration.Inf)
val rows = Await.result(futureResult, timeout)
Copy link
Contributor

@marmbrus marmbrus Feb 19, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This timeout is kind of weird right? Like the max timeout here is numberOfSubqueries * timeout?

Copy link
Contributor Author

@davies davies Feb 19, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All the subqueries are submitted in the same (during beginning of prepare()), so the total time should be timeout.

Should we create another config for subquery or rename the broadcast one?

if (rows.length > 1) {
sys.error(s"Scalar subquery should return at most one row, but got ${rows.length}: " +
Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can use postgres' error message: "more than one row returned by a subquery used as an expression"

Copy link
Contributor Author

@davies davies Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have never thought we should match the exactly error message with PostgreSQL, that's great.

Copy link
Contributor Author

@davies davies Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The current error message has more information than postgres', should we change?

Copy link
Contributor

@rxin rxin Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not 100% sure. maybe it's better to just say more than one, so we don't need to run the whole plan (e..g i'm thinking maybe we should inject a limit of 2 to subquery)

Copy link
Contributor Author

@davies davies Feb 20, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea, changed to call executeTake(2)

s"${e.query.treeString}")
Expand Down