add missing canonicalization to ScalarSubquery, fix planing of Dynami…

…cPruningSubquery as subquery
apache · Nov 19, 2019 · e0b4ca6 · e0b4ca6
1 parent 3f9e240
commit e0b4ca6
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/dynamicpruning/PlanDynamicPruningFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/dynamicpruning/PlanDynamicPruningFilters.scala
@@ -55,7 +55,7 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession)
     plan transformAllExpressions {
       case DynamicPruningSubquery(
           value, buildPlan, buildKeys, broadcastKeyIndex, onlyInBroadcast, exprId) =>
-        val qe = new QueryExecution(sparkSession, buildPlan)
+        val qe = new QueryExecution(sparkSession, buildPlan, subQuery = true)
         // Using `sparkPlan` is a little hacky as it is based on the assumption that this rule is
         // the first to be applied (apart from `InsertAdaptiveSparkPlan`).
         val canReuseExchange = reuseBroadcast && buildKeys.nonEmpty &&

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala
@@ -104,6 +104,10 @@ case class ScalarSubquery(
     require(updated, s"$this has not finished")
     Literal.create(result, dataType).doGenCode(ctx, ev)
   }
+
+  override lazy val canonicalized: ScalarSubquery = {
+    copy(plan = plan.canonicalized.asInstanceOf[BaseSubqueryExec], exprId = ExprId(0))
+  }
 }
 
 /**