Adaptive query execution works correctly when first stage partitions size is 0 #62
Comments
7mming7
added a commit
that referenced
this issue
Oct 12, 2019
#62 [SPARK-29284][SQL] Adaptive query execution works correctly when …
7mming7
pushed a commit
that referenced
this issue
Nov 4, 2020
…zing floats
### What changes were proposed in this pull request?
This PR intends to fix a minor bug when normalizing floats for struct types;
```
scala> import org.apache.spark.sql.execution.aggregate.HashAggregateExec
scala> val df = Seq(Tuple1(Tuple1(-0.0d)), Tuple1(Tuple1(0.0d))).toDF("k")
scala> val agg = df.distinct()
scala> agg.explain()
== Physical Plan ==
*(2) HashAggregate(keys=[k#40], functions=[])
+- Exchange hashpartitioning(k#40, 200), true, [id=#62]
+- *(1) HashAggregate(keys=[knownfloatingpointnormalized(if (isnull(k#40)) null else named_struct(col1, knownfloatingpointnormalized(normalizenanandzero(k#40._1)))) AS k#40], functions=[])
+- *(1) LocalTableScan [k#40]
scala> val aggOutput = agg.queryExecution.sparkPlan.collect { case a: HashAggregateExec => a.output.head }
scala> aggOutput.foreach { attr => println(attr.prettyJson) }
### Final Aggregate ###
[ {
"class" : "org.apache.spark.sql.catalyst.expressions.AttributeReference",
"num-children" : 0,
"name" : "k",
"dataType" : {
"type" : "struct",
"fields" : [ {
"name" : "_1",
^^^
"type" : "double",
"nullable" : false,
"metadata" : { }
} ]
},
"nullable" : true,
"metadata" : { },
"exprId" : {
"product-class" : "org.apache.spark.sql.catalyst.expressions.ExprId",
"id" : 40,
"jvmId" : "a824e83f-933e-4b85-a1ff-577b5a0e2366"
},
"qualifier" : [ ]
} ]
### Partial Aggregate ###
[ {
"class" : "org.apache.spark.sql.catalyst.expressions.AttributeReference",
"num-children" : 0,
"name" : "k",
"dataType" : {
"type" : "struct",
"fields" : [ {
"name" : "col1",
^^^^
"type" : "double",
"nullable" : true,
"metadata" : { }
} ]
},
"nullable" : true,
"metadata" : { },
"exprId" : {
"product-class" : "org.apache.spark.sql.catalyst.expressions.ExprId",
"id" : 40,
"jvmId" : "a824e83f-933e-4b85-a1ff-577b5a0e2366"
},
"qualifier" : [ ]
} ]
```
### Why are the changes needed?
bugfix.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests.
Closes apache#29780 from maropu/FixBugInNormalizedFloatingNumbers.
Authored-by: Takeshi Yamamuro <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
lxian
pushed a commit
to lxian/spark
that referenced
this issue
Sep 24, 2021
…query
### What changes were proposed in this pull request?
Remove redundant aliases after `RewritePredicateSubquery`. For example:
```scala
sql("CREATE TABLE t1 USING parquet AS SELECT id AS a, id AS b, id AS c FROM range(10)")
sql("CREATE TABLE t2 USING parquet AS SELECT id AS x, id AS y FROM range(8)")
sql(
"""
|SELECT *
|FROM t1
|WHERE a IN (SELECT x
| FROM (SELECT x AS x,
| Rank() OVER (partition BY x ORDER BY Sum(y) DESC) AS ranking
| FROM t2
| GROUP BY x) tmp1
| WHERE ranking <= 5)
|""".stripMargin).explain
```
Before this PR:
```
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [a#10L], [x#7L], LeftSemi, BuildRight, false
:- FileScan parquet default.t1[a#10L,b#11L,c#12L]
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, true]),false), [id=Kyligence#68]
+- Project [x#7L]
+- Filter (ranking#8 <= 5)
+- Window [rank(_w2#25L) windowspecdefinition(x#15L, _w2#25L DESC NULLS LAST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS ranking#8], [x#15L], [_w2#25L DESC NULLS LAST]
+- Sort [x#15L ASC NULLS FIRST, _w2#25L DESC NULLS LAST], false, 0
+- Exchange hashpartitioning(x#15L, 5), ENSURE_REQUIREMENTS, [id=Kyligence#62]
+- HashAggregate(keys=[x#15L], functions=[sum(y#16L)])
+- Exchange hashpartitioning(x#15L, 5), ENSURE_REQUIREMENTS, [id=Kyligence#59]
+- HashAggregate(keys=[x#15L], functions=[partial_sum(y#16L)])
+- FileScan parquet default.t2[x#15L,y#16L]
```
After this PR:
```
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [a#10L], [x#15L], LeftSemi, BuildRight, false
:- FileScan parquet default.t1[a#10L,b#11L,c#12L]
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, true]),false), [id=Kyligence#67]
+- Project [x#15L]
+- Filter (ranking#8 <= 5)
+- Window [rank(_w2#25L) windowspecdefinition(x#15L, _w2#25L DESC NULLS LAST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS ranking#8], [x#15L], [_w2#25L DESC NULLS LAST]
+- Sort [x#15L ASC NULLS FIRST, _w2#25L DESC NULLS LAST], false, 0
+- HashAggregate(keys=[x#15L], functions=[sum(y#16L)])
+- Exchange hashpartitioning(x#15L, 5), ENSURE_REQUIREMENTS, [id=Kyligence#59]
+- HashAggregate(keys=[x#15L], functions=[partial_sum(y#16L)])
+- FileScan parquet default.t2[x#15L,y#16L]
```
### Why are the changes needed?
Reduce shuffle to improve query performance. This change can benefit TPC-DS q70.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit test.
Closes apache#33509 from wangyum/SPARK-36280.
Authored-by: Yuming Wang <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
https://github.com/Kyligence/KAP/issues/14900
The text was updated successfully, but these errors were encountered: