-
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 5 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,81 @@ | ||
/* | ||
* 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{ | ||
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 a space after 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 |
||
|
||
/** | ||
* The underline plan for the query, could be logical plan or physical plan. | ||
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. "Either a logical plan or a physical plan. The generated tree string (explain output) uses this field to explain the subquery." |
||
* | ||
* This is used to generate tree string. | ||
*/ | ||
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 is not evaluable, it should be converted into SparkScalaSubquery. | ||
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 will be converted into [[execution.ScalarSubquery]] during physical planning. (make sure it links properly) |
||
*/ | ||
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 can only have 1 column, but got " + | ||
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. needs tests. Maybe, probably 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 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. "Scalar subquery must return only one column, but got " ... (postgres') |
||
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 |
---|---|---|
|
@@ -88,7 +88,19 @@ abstract class Optimizer extends RuleExecutor[LogicalPlan] { | |
Batch("Decimal Optimizations", FixedPoint(100), | ||
DecimalAggregates) :: | ||
Batch("LocalRelation", FixedPoint(100), | ||
ConvertToLocalRelation) :: Nil | ||
ConvertToLocalRelation) :: | ||
Batch("Subquery", Once, | ||
Subquery) :: Nil | ||
} | ||
|
||
/** | ||
* Optimize all the subqueries inside expression. | ||
*/ | ||
object Subquery extends Rule[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. Subquery -> OptimizeSubqueries |
||
def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions { | ||
case subquery: SubqueryExpression => | ||
subquery.withNewPlan(execute(subquery.query)) | ||
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. where is execute coming from? 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 parent class, Optimizer. 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 replace it with "Optimizer.this.execute" so it is more clear. Thanks! |
||
} | ||
} | ||
} | ||
|
||
|
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,49 @@ class CatalystQlSuite extends PlanTest { | |
parser.parsePlan("select sum(product + 1) over (partition by (product + (1)) order by 2) " + | ||
"from windowData") | ||
} | ||
|
||
test("subquery") { | ||
comparePlans( | ||
parser.parsePlan("select (select max(b) from s) ss from t"), | ||
Project( | ||
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. @hvanhovell I'm going to remove these plan checking, it's very easy to break. The details of plan does not mean much, we will have other tests to verify the correctness. |
||
UnresolvedAlias( | ||
Alias( | ||
ScalarSubquery( | ||
Project( | ||
UnresolvedAlias( | ||
UnresolvedFunction("max", UnresolvedAttribute("b") :: Nil, false)) :: Nil, | ||
UnresolvedRelation(TableIdentifier("s")))), | ||
"ss")(ExprId(0))) :: Nil, | ||
UnresolvedRelation(TableIdentifier("t")))) | ||
comparePlans( | ||
parser.parsePlan("select * from t where a = (select b from s)"), | ||
Project( | ||
UnresolvedAlias( | ||
UnresolvedStar(None)) :: Nil, | ||
Filter( | ||
EqualTo( | ||
UnresolvedAttribute("a"), | ||
ScalarSubquery( | ||
Project( | ||
UnresolvedAlias( | ||
UnresolvedAttribute("b")) :: Nil, | ||
UnresolvedRelation(TableIdentifier("s"))))), | ||
UnresolvedRelation(TableIdentifier("t"))))) | ||
comparePlans( | ||
parser.parsePlan("select * from t group by g having a > (select b from s)"), | ||
Filter( | ||
Cast( | ||
GreaterThan( | ||
UnresolvedAttribute("a"), | ||
ScalarSubquery( | ||
Project( | ||
UnresolvedAlias( | ||
UnresolvedAttribute("b")) :: Nil, | ||
UnresolvedRelation(TableIdentifier("s"))))), | ||
BooleanType), | ||
Aggregate( | ||
UnresolvedAttribute("g") :: Nil, | ||
UnresolvedAlias(UnresolvedStar(None)) :: Nil, | ||
UnresolvedRelation(TableIdentifier("t"))))) | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -113,6 +113,12 @@ class AnalysisErrorSuite extends AnalysisTest { | |
|
||
val dateLit = Literal.create(null, DateType) | ||
|
||
errorTest( | ||
"invalid scalar subquery", | ||
testRelation.select( | ||
(ScalarSubquery(testRelation.select('a, dateLit.as('b))) + Literal(1)).as('a)), | ||
"Scalar subquery can only have 1 column, but got 2" :: Nil) | ||
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 a test case for 0 column if possible |
||
|
||
errorTest( | ||
"single invalid type, single arg", | ||
testRelation.select(TestFunction(dateLit :: Nil, IntegerType :: Nil).as('a)), | ||
|
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,32 @@ 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[(SparkScalarSubquery, Future[Array[InternalRow]])]() | ||
val allSubqueries = expressions.flatMap(_.collect {case e: SparkScalarSubquery => e}) | ||
allSubqueries.foreach { e => | ||
val futureResult = Future { | ||
e.plan.executeCollect() | ||
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. e.executedPlan rather than e.plan, if you want to keep executedPlan |
||
}(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"Scalar subquery should return at most one row, but got ${rows.length}: " + | ||
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 can use postgres' error message: "more than one row returned by a subquery used as an 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. I have never thought we should match the exactly error message with PostgreSQL, that's great. 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 current error message has more information than postgres', should we change? 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. 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) 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. Good idea, changed to call |
||
s"${e.query.treeString}") | ||
} | ||
// Analyzer will make sure that it only return on column | ||
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 should make sure this only returns one column" and add an assert after this. |
||
if (rows.length > 0) { | ||
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 rows.length ever be 0 here? if it can only be 1, why we are testing > 0 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. the length could be zero, then the value is null. 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 do we write a query with 0 column? The comment above said the analyzer would make sure there's only one column. If it is possible to have 0 column, then I'd make it explicitly here to set the value to null. 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. Also if it is possible to have 0 column, we also need to add a test case. 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.
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 makes sense. please change the check to rows.length == 1 it's pretty confusing to first check it's greater than 1, and then check it is greater than 0, when you are just expecting 1. 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. and same thing applies - the test coverage for this is pretty bad. add a test case where the subquery returns 0 or more than 1 rows. 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. Repeated: 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. Yea, it'd be better to make it more explicit, e.g.
|
||
e.updateResult(rows(0).get(0, e.dataType)) | ||
} | ||
} | ||
} | ||
} | ||
|
||
|
@@ -231,6 +259,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