You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/04/06 03:30:39 UTC
[GitHub] [spark] wangyum opened a new pull request, #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
wangyum opened a new pull request, #36080:
URL: https://github.com/apache/spark/pull/36080
### What changes were proposed in this pull request?
This PR makes row-level runtime filtering support pruning side has window. For example:
```sql
SELECT *
FROM (SELECT *,
Row_number() OVER ( partition BY c1 ORDER BY f1) rn
FROM bf1) bf1
JOIN bf2
ON bf1.c1 = bf2.c2
WHERE bf2.a2 = 62
```
After this PR:
```
== Optimized Logical Plan ==
Join Inner, (c1#45922 = c2#45928), Statistics(sizeInBytes=12.3 MiB)
:- Window [row_number() windowspecdefinition(c1#45922, f1#45925 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn#45976], [c1#45922], [f1#45925 ASC NULLS FIRST], Statistics(sizeInBytes=3.7 KiB)
: +- Filter (isnotnull(c1#45922) AND might_contain(scalar-subquery#45993 [], xxhash64(c1#45922, 42))), Statistics(sizeInBytes=3.3 KiB)
: : +- Aggregate [bloom_filter_agg(xxhash64(c2#45928, 42), 1000000, 8388608, 0, 0) AS bloomFilter#45992], Statistics(sizeInBytes=108.0 B, rowCount=1)
: : +- Project [c2#45928], Statistics(sizeInBytes=1278.0 B)
: : +- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928)), Statistics(sizeInBytes=3.3 KiB)
: : +- Relation default.bf2[a2#45926,b2#45927,c2#45928,d2#45929,e2#45930,f2#45931] parquet, Statistics(sizeInBytes=3.3 KiB)
: +- Relation default.bf1[a1#45920,b1#45921,c1#45922,d1#45923,e1#45924,f1#45925] parquet, Statistics(sizeInBytes=3.3 KiB)
+- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928)), Statistics(sizeInBytes=3.3 KiB)
+- Relation default.bf2[a2#45926,b2#45927,c2#45928,d2#45929,e2#45930,f2#45931] parquet, Statistics(sizeInBytes=3.3 KiB)
```
### Why are the changes needed?
Improve query performance.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843531961
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
It is needed because some filters can't push through window, for example:
```sql
SELECT *
FROM (SELECT *,
Row_number() OVER (PARTITION BY d1 ORDER BY f1) rn
FROM bf1) bf1
JOIN bf2 ON bf1.c1 = bf2.c2
WHERE bf2.a2 = 62
```
```
== Optimized Logical Plan ==
Join Inner, (c1#45922 = c2#45928), Statistics(sizeInBytes=12.3 MiB)
:- Filter (isnotnull(c1#45922) AND might_contain(scalar-subquery#45993 [], xxhash64(c1#45922, 42))), Statistics(sizeInBytes=3.7 KiB)
: : +- Aggregate [bloom_filter_agg(xxhash64(c2#45928, 42), 1000000, 8388608, 0, 0) AS bloomFilter#45992], Statistics(sizeInBytes=108.0 B, rowCount=1)
: : +- Project [c2#45928], Statistics(sizeInBytes=1278.0 B)
: : +- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928)), Statistics(sizeInBytes=3.3 KiB)
: : +- Relation default.bf2[a2#45926,b2#45927,c2#45928,d2#45929,e2#45930,f2#45931] parquet, Statistics(sizeInBytes=3.3 KiB)
: +- Window [row_number() windowspecdefinition(d1#45923, f1#45925 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn#45976], [d1#45923], [f1#45925 ASC NULLS FIRST], Statistics(sizeInBytes=3.7 KiB)
: +- Relation default.bf1[a1#45920,b1#45921,c1#45922,d1#45923,e1#45924,f1#45925] parquet, Statistics(sizeInBytes=3.3 KiB)
+- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928)), Statistics(sizeInBytes=3.3 KiB)
+- Relation default.bf2[a2#45926,b2#45927,c2#45928,d2#45929,e2#45930,f2#45931] parquet, Statistics(sizeInBytes=3.3 KiB)
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [c1#45922], [c2#45928], Inner, BuildRight, false
:- Filter (isnotnull(c1#45922) AND might_contain(Subquery subquery#45993, [id=#417], xxhash64(c1#45922, 42)))
: : +- Subquery subquery#45993, [id=#417]
: : +- AdaptiveSparkPlan isFinalPlan=false
: : +- ObjectHashAggregate(keys=[], functions=[bloom_filter_agg(xxhash64(c2#45928, 42), 1000000, 8388608, 0, 0)], output=[bloomFilter#45992])
: : +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#415]
: : +- ObjectHashAggregate(keys=[], functions=[partial_bloom_filter_agg(xxhash64(c2#45928, 42), 1000000, 8388608, 0, 0)], output=[buf#45995])
: : +- Project [c2#45928]
: : +- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928))
: : +- FileScan parquet default.bf2[a2#45926,c2#45928] Batched: true, DataFilters: [isnotnull(a2#45926), (a2#45926 = 62), isnotnull(c2#45928)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/hive-3.1.2/connector/kafka-0-10/spark-wareho..., PartitionFilters: [], PushedFilters: [IsNotNull(a2), EqualTo(a2,62), IsNotNull(c2)], ReadSchema: struct<a2:int,c2:int>
: +- Window [row_number() windowspecdefinition(d1#45923, f1#45925 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn#45976], [d1#45923], [f1#45925 ASC NULLS FIRST]
: +- Sort [d1#45923 ASC NULLS FIRST, f1#45925 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(d1#45923, 5), ENSURE_REQUIREMENTS, [id=#422]
: +- FileScan parquet default.bf1[a1#45920,b1#45921,c1#45922,d1#45923,e1#45924,f1#45925] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/hive-3.1.2/connector/kafka-0-10/spark-wareho..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a1:int,b1:int,c1:int,d1:int,e1:int,f1:int>
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[2, int, false] as bigint)),false), [id=#427]
+- Filter ((isnotnull(a2#45926) AND (a2#45926 = 62)) AND isnotnull(c2#45928))
+- FileScan parquet default.bf2[a2#45926,b2#45927,c2#45928,d2#45929,e2#45930,f2#45931] Batched: true, DataFilters: [isnotnull(a2#45926), (a2#45926 = 62), isnotnull(c2#45928)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/yumwang/spark/hive-3.1.2/connector/kafka-0-10/spark-wareho..., PartitionFilters: [], PushedFilters: [IsNotNull(a2), EqualTo(a2,62), IsNotNull(c2)], ReadSchema: struct<a2:int,b2:int,c2:int,d2:int,e2:int,f2:int>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843469873
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
Review Comment:
Is `expr` needed?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
How about:
case w: Window => true?
I think `findExpressionAndTrackLineageDown` already checked the pushdown-ability?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843469373
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
How about:
case w: Window => true?
I think `findExpressionAndTrackLineageDown` already checked the pushdown-ability to the Scan node?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r845137729
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,38 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ // Make sure injected filters could push through Shuffle, see PushPredicateThroughNonJoin
+ private def probablyPushThroughShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ if (exp.references.isEmpty) return false
Review Comment:
`findExpressionAndTrackLineageDown` is used to check if this Expression is from `LeafNode`, so we don't need to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] singhpk234 commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
singhpk234 commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r846594345
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
Review Comment:
Nice !!!, having this unified it now supports LeftSemi as well, which was added for a follow-up
ref : https://issues.apache.org/jira/browse/SPARK-38565
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843537311
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
Make sense. Is it possible to add the logic for Window into `findExpressionAndTrackLineageDown` since it's a dedicated place to check pushdown-ability?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
Make sense. Is it possible to move the logic for Window into `findExpressionAndTrackLineageDown` since it's a dedicated place to check pushdown-ability?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum closed pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum closed pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
URL: https://github.com/apache/spark/pull/36080
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r845046305
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,38 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ // Make sure injected filters could push through Shuffle, see PushPredicateThroughNonJoin
+ private def probablyPushThroughShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ if (exp.references.isEmpty) return false
+
+ plan match {
+ case Join(left, right, _, _, hint) if isProbablyShuffleJoin(left, right, hint) &&
+ (exp.references.subsetOf(left.outputSet) || exp.references.subsetOf(right.outputSet)) =>
+ true
+ case a: Aggregate if a.aggregateExpressions.forall(_.deterministic) &&
Review Comment:
I thought there would be a scenario: the current Aggregate cannot be pushed down, but the child can be pushed down. But this scenario does not exist. Will remove next Aggregate case branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on PR #36080:
URL: https://github.com/apache/spark/pull/36080#issuecomment-1089750209
cc @sigmod
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r844821659
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,38 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ // Make sure injected filters could push through Shuffle, see PushPredicateThroughNonJoin
+ private def probablyPushThroughShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ if (exp.references.isEmpty) return false
Review Comment:
Just curious: are you saying we can inject a Bloom filter as long as it can be pushed through a shuffle, but not necessarily to immediately after the scan?
Alternatively, I was thinking that `findExpressionAndTrackLineageDown` for `Window`, and add a Window case in the original `probablyHasShuffle`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,38 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ // Make sure injected filters could push through Shuffle, see PushPredicateThroughNonJoin
+ private def probablyPushThroughShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ if (exp.references.isEmpty) return false
+
+ plan match {
+ case Join(left, right, _, _, hint) if isProbablyShuffleJoin(left, right, hint) &&
+ (exp.references.subsetOf(left.outputSet) || exp.references.subsetOf(right.outputSet)) =>
+ true
+ case a: Aggregate if a.aggregateExpressions.forall(_.deterministic) &&
Review Comment:
what's the difference with the next Aggregate case branch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r849133914
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -141,6 +141,7 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
plan.exists {
case Join(left, right, _, _, hint) => isProbablyShuffleJoin(left, right, hint)
case _: Aggregate => true
+ case _: Window => true
Review Comment:
@sigmod Simplified code: just check if filterApplicationSide has Window.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on PR #36080:
URL: https://github.com/apache/spark/pull/36080#issuecomment-1097985700
Test: https://github.com/wangyum/spark/actions/runs/2152299557
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843428746
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
Review Comment:
Move these functions to `JoinSelectionHelper` to make the code easier to maintain.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843535926
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
Review Comment:
See https://github.com/apache/spark/pull/36080#discussion_r843531961
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r846780292
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
Review Comment:
https://github.com/apache/spark/pull/36131
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on PR #36080:
URL: https://github.com/apache/spark/pull/36080#issuecomment-1098601727
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on PR #36080:
URL: https://github.com/apache/spark/pull/36080#issuecomment-1093116663
@andylam-db
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wangyum commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
wangyum commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r843661577
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,21 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ private def probablyHasShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ plan.exists {
+ case Join(left, right, _, _, hint)
+ if isProbablyShuffleJoin(left, right, hint) => true
+ case _: Aggregate => true
+ case w: Window
+ if exp.references.subsetOf(AttributeSet(w.partitionSpec.flatMap(_.references))) => true
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sigmod commented on a diff in pull request #36080: [SPARK-38797][SQL] Runtime Filter supports pruning side has window
Posted by GitBox <gi...@apache.org>.
sigmod commented on code in PR #36080:
URL: https://github.com/apache/spark/pull/36080#discussion_r844821659
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -132,28 +131,38 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with PredicateHelper with J
REGEXP_EXTRACT_FAMILY, REGEXP_REPLACE)
}
- private def canFilterLeft(joinType: JoinType): Boolean = joinType match {
- case Inner | RightOuter => true
- case _ => false
- }
-
- private def canFilterRight(joinType: JoinType): Boolean = joinType match {
- case Inner | LeftOuter => true
- case _ => false
- }
-
private def isProbablyShuffleJoin(left: LogicalPlan,
right: LogicalPlan, hint: JoinHint): Boolean = {
!hintToBroadcastLeft(hint) && !hintToBroadcastRight(hint) &&
!canBroadcastBySize(left, conf) && !canBroadcastBySize(right, conf)
}
- private def probablyHasShuffle(plan: LogicalPlan): Boolean = {
- plan.collectFirst {
- case j@Join(left, right, _, _, hint)
- if isProbablyShuffleJoin(left, right, hint) => j
- case a: Aggregate => a
- }.nonEmpty
+ // Make sure injected filters could push through Shuffle, see PushPredicateThroughNonJoin
+ private def probablyPushThroughShuffle(exp: Expression, plan: LogicalPlan): Boolean = {
+ if (exp.references.isEmpty) return false
Review Comment:
Just curious: are you saying we can inject a Bloom filter as long as it can be pushed through a shuffle, but not necessarily to immediately after the scan?
Alternatively, I was thinking that we can fix `findExpressionAndTrackLineageDown` for `Window`, and add a Window case in the original `probablyHasShuffle`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org