-
Notifications
You must be signed in to change notification settings - Fork 28.1k
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.
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
[SPARK-13306] [SQL] uncorrelated scalar subquery #11190
Changes from 7 commits
0665a69
236ac88
016c36c
a4bae33
d0974cf
3a8f08d
7596173
0034172
e082845
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -80,6 +80,7 @@ class Analyzer( | |
ResolveGenerate :: | ||
ResolveFunctions :: | ||
ResolveAliases :: | ||
ResolveSubquery :: | ||
ResolveWindowOrder :: | ||
ResolveWindowFrame :: | ||
ResolveNaturalJoin :: | ||
|
@@ -120,7 +121,14 @@ class Analyzer( | |
withAlias.getOrElse(relation) | ||
} | ||
substituted.getOrElse(u) | ||
case other => | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. quick comment on why this isn't in There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. done |
||
// This can't be done in ResolveSubquery because that does not know the CTE. | ||
other transformExpressions { | ||
case e: SubqueryExpression => | ||
e.withNewPlan(substituteCTE(e.query, cteRelations)) | ||
} | ||
} | ||
|
||
} | ||
} | ||
|
||
|
@@ -693,6 +701,30 @@ class Analyzer( | |
} | ||
} | ||
|
||
/** | ||
* This rule resolve subqueries inside expressions. | ||
* | ||
* Note: CTE are handled in CTESubstitution. | ||
*/ | ||
object ResolveSubquery extends Rule[LogicalPlan] with PredicateHelper { | ||
|
||
private def hasSubquery(e: Expression): Boolean = { | ||
e.find(_.isInstanceOf[SubqueryExpression]).isDefined | ||
} | ||
|
||
private def hasSubquery(q: LogicalPlan): Boolean = { | ||
q.expressions.exists(hasSubquery) | ||
} | ||
|
||
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators { | ||
case q: LogicalPlan if q.childrenResolved && hasSubquery(q) => | ||
q transformExpressions { | ||
case e: SubqueryExpression if !e.query.resolved => | ||
e.withNewPlan(execute(e.query)) | ||
} | ||
} | ||
} | ||
|
||
/** | ||
* Turns projections that contain aggregate expressions into aggregations. | ||
*/ | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.sql.catalyst.expressions | ||
|
||
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult | ||
import org.apache.spark.sql.catalyst.plans.QueryPlan | ||
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Subquery} | ||
import org.apache.spark.sql.types.DataType | ||
|
||
/** | ||
* An interface for subquery that is used in expressions. | ||
*/ | ||
abstract class SubqueryExpression extends LeafExpression { | ||
|
||
/** | ||
* The logical plan of the query. | ||
*/ | ||
def query: LogicalPlan | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why is this needed? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is an helper function used in Analyzer and Optimizer, or we need to do type conversion. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is the base class for both logical plan and physical plan, kind of weird. This is to make the generateTreeString works in QueryPlan There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Analyzer and Optimizer only applies to logical plan right? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. yes |
||
|
||
/** | ||
* Either a logical plan or a physical plan. The generated tree string (explain output) uses this | ||
* field to explain the subquery. | ||
*/ | ||
def plan: QueryPlan[_] | ||
|
||
/** | ||
* Updates the query with new logical plan. | ||
*/ | ||
def withNewPlan(plan: LogicalPlan): SubqueryExpression | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. scala doc There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can't this be just in the logical plan itself? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This should be There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. i think you can just remove this and move it into the logical subquery expression, since it's only used for logical plan anyway? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Then should we have LogicalSubqueryExpression ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I meant ScalarSubquery. That's already the one isn't it? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We will have ExistsSubquery, InSubquery shortly (or next release). |
||
} | ||
|
||
/** | ||
* A subquery that will return only one row and one column. | ||
* | ||
* This will be converted into [[execution.ScalarSubquery]] during physical planning. | ||
* | ||
* Note: `exprId` is used to have unique name in explain string output. | ||
*/ | ||
case class ScalarSubquery( | ||
query: LogicalPlan, | ||
exprId: ExprId = NamedExpression.newExprId) | ||
extends SubqueryExpression with Unevaluable { | ||
|
||
override def plan: LogicalPlan = Subquery(toString, query) | ||
|
||
override lazy val resolved: Boolean = query.resolved | ||
|
||
override def dataType: DataType = query.schema.fields.head.dataType | ||
|
||
override def checkInputDataTypes(): TypeCheckResult = { | ||
if (query.schema.length != 1) { | ||
TypeCheckResult.TypeCheckFailure("Scalar subquery must return only one column, but got " + | ||
query.schema.length.toString) | ||
} else { | ||
TypeCheckResult.TypeCheckSuccess | ||
} | ||
} | ||
|
||
override def foldable: Boolean = false | ||
override def nullable: Boolean = true | ||
|
||
override def withNewPlan(plan: LogicalPlan): ScalarSubquery = ScalarSubquery(plan, exprId) | ||
|
||
override def toString: String = s"subquery#${exprId.id}" | ||
|
||
// TODO: support sql() | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -18,6 +18,7 @@ | |
package org.apache.spark.sql.catalyst.plans | ||
|
||
import org.apache.spark.sql.catalyst.expressions._ | ||
import org.apache.spark.sql.catalyst.plans.logical.Subquery | ||
import org.apache.spark.sql.catalyst.trees.TreeNode | ||
import org.apache.spark.sql.types.{DataType, StructType} | ||
|
||
|
@@ -226,4 +227,29 @@ abstract class QueryPlan[PlanType <: TreeNode[PlanType]] extends TreeNode[PlanTy | |
protected def statePrefix = if (missingInput.nonEmpty && children.nonEmpty) "!" else "" | ||
|
||
override def simpleString: String = statePrefix + super.simpleString | ||
|
||
override def generateTreeString( | ||
depth: Int, lastChildren: Seq[Boolean], builder: StringBuilder): StringBuilder = { | ||
if (depth > 0) { | ||
lastChildren.init.foreach { isLast => | ||
val prefixFragment = if (isLast) " " else ": " | ||
builder.append(prefixFragment) | ||
} | ||
|
||
val branch = if (lastChildren.last) "+- " else ":- " | ||
builder.append(branch) | ||
} | ||
|
||
builder.append(simpleString) | ||
builder.append("\n") | ||
|
||
val allSubqueries = expressions.flatMap(_.collect {case e: SubqueryExpression => e}) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can't we do this in subquery itself? It is pretty odd to have this general base class depend on some specific expression There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. How to do that? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can you try? I don't have time to think/write this one. It just feels bad to do it here. If it is impossible, then of course we have to do it here ... There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If I could figure out a way, would not copy this function here and ask you. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why not create a |
||
val allChildren = children ++ allSubqueries.map(e => e.plan) | ||
if (allChildren.nonEmpty) { | ||
allChildren.init.foreach(_.generateTreeString(depth + 1, lastChildren :+ false, builder)) | ||
allChildren.last.generateTreeString(depth + 1, lastChildren :+ true, builder) | ||
} | ||
|
||
builder | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.analysis._ | |
import org.apache.spark.sql.catalyst.expressions._ | ||
import org.apache.spark.sql.catalyst.plans.PlanTest | ||
import org.apache.spark.sql.catalyst.plans.logical._ | ||
import org.apache.spark.sql.types.BooleanType | ||
import org.apache.spark.unsafe.types.CalendarInterval | ||
|
||
class CatalystQlSuite extends PlanTest { | ||
|
@@ -201,4 +202,10 @@ class CatalystQlSuite extends PlanTest { | |
parser.parsePlan("select sum(product + 1) over (partition by (product + (1)) order by 2) " + | ||
"from windowData") | ||
} | ||
|
||
test("subquery") { | ||
parser.parsePlan("select (select max(b) from s) ss from t") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The only thing we are testing here is that things don't go really really wrong. I'd prefer it if we test the plan as well. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Since plan checking is too easy to break, I added test for plan, finally remove them. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ok that makes sense. |
||
parser.parsePlan("select * from t where a = (select b from s)") | ||
parser.parsePlan("select * from t group by g having a > (select b from s)") | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -20,6 +20,8 @@ package org.apache.spark.sql.execution | |
import java.util.concurrent.atomic.AtomicBoolean | ||
|
||
import scala.collection.mutable.ArrayBuffer | ||
import scala.concurrent.{Await, ExecutionContext, Future} | ||
import scala.concurrent.duration._ | ||
|
||
import org.apache.spark.Logging | ||
import org.apache.spark.rdd.{RDD, RDDOperationScope} | ||
|
@@ -31,6 +33,7 @@ import org.apache.spark.sql.catalyst.plans.QueryPlan | |
import org.apache.spark.sql.catalyst.plans.physical._ | ||
import org.apache.spark.sql.execution.metric.{LongSQLMetric, SQLMetric} | ||
import org.apache.spark.sql.types.DataType | ||
import org.apache.spark.util.ThreadUtils | ||
|
||
/** | ||
* The base class for physical operators. | ||
|
@@ -122,7 +125,34 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ | |
final def prepare(): Unit = { | ||
if (prepareCalled.compareAndSet(false, true)) { | ||
doPrepare() | ||
|
||
// collect all the subqueries and submit jobs to execute them in background | ||
val queryResults = ArrayBuffer[(ScalarSubquery, Future[Array[InternalRow]])]() | ||
val allSubqueries = expressions.flatMap(_.collect {case e: ScalarSubquery => e}) | ||
allSubqueries.foreach { e => | ||
val futureResult = Future { | ||
e.plan.executeTake(2) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. add some comment explaining the 2 here |
||
}(SparkPlan.subqueryExecutionContext) | ||
queryResults += e -> futureResult | ||
} | ||
|
||
children.foreach(_.prepare()) | ||
|
||
// fill in the result of subqueries | ||
queryResults.foreach { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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. There was a problem hiding this comment. Choose a reason for hiding this commentThe 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. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. or a subquery is now blocking broadcasting ... There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is called after There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. what if there is a broadcast join after this? There was a problem hiding this comment. Choose a reason for hiding this commentThe 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) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we really want to wait that long? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. will use sqlContext.conf.broadcastTimeout There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. |
||
if (rows.length > 1) { | ||
sys.error(s"more than one row returned by a subquery used as an expression:\n${e.plan}") | ||
} | ||
if (rows.length == 1) { | ||
assert(rows(0).numFields == 1, "Analyzer should make sure this only returns one column") | ||
e.updateResult(rows(0).get(0, e.dataType)) | ||
} else { | ||
// There is no rows returned, the result should be null. | ||
e.updateResult(null) | ||
} | ||
} | ||
} | ||
} | ||
|
||
|
@@ -231,6 +261,11 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ | |
} | ||
} | ||
|
||
object SparkPlan { | ||
private[execution] val subqueryExecutionContext = ExecutionContext.fromExecutorService( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What threadpool are broadcasts done on? Should it be the same? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This could be refactored later, use the same thread pool for all of them. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
ThreadUtils.newDaemonCachedThreadPool("subquery", 16)) | ||
} | ||
|
||
private[sql] trait LeafNode extends SparkPlan { | ||
override def children: Seq[SparkPlan] = Nil | ||
override def producedAttributes: AttributeSet = outputSet | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -343,3 +343,18 @@ case class OutputFaker(output: Seq[Attribute], child: SparkPlan) extends SparkPl | |
|
||
protected override def doExecute(): RDD[InternalRow] = child.execute() | ||
} | ||
|
||
/** | ||
* A plan as subquery. | ||
* | ||
* This is used to generate tree string for SparkScalarSubquery. | ||
*/ | ||
case class Subquery(name: String, child: SparkPlan) extends UnaryNode { | ||
override def output: Seq[Attribute] = child.output | ||
override def outputPartitioning: Partitioning = child.outputPartitioning | ||
override def outputOrdering: Seq[SortOrder] = child.outputOrdering | ||
|
||
protected override def doExecute(): RDD[InternalRow] = { | ||
child.execute() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. throw unsupported exception? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Maybe it could be used some day, I'd like to leave it as it is, since it's correct. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. But by then we can just implement it? I think once execute is called, this violates your design that this is only used in a subquery, and we might have subtle other problems .. |
||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This might sound excedingly dumb but I cannot find
ScalarSubquery
orSubqueryExpression
. Are they already in the code base? Or did you create branch on top of another branch?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nevermind I just found the other PR...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I missed a file, sorry