Skip to content

[SPARK-41162][SQL] Fix anti- and semi-join for self-join with aggregations #39131

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.

Sign up for GitHub

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

Closed
EnricoMi wants to merge 5 commits into from
Closed

[SPARK-41162][SQL] Fix anti- and semi-join for self-join with aggregations #39131

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
eebeb34
Use canPushThroughCondition for LeftSemi and LeftAnti joins
EnricoMi Dec 19, 2022
76dcd1e
Fix LeftSemiPushdownSuite class name (match file name)
EnricoMi Dec 19, 2022
ade61b8
Rename LeftSemiAnti to LeftSemi in LeftSemiAntiJoinPushDownSuite
EnricoMi Dec 19, 2022
cc28b21
Adjust test names to one naming scheme
EnricoMi Dec 19, 2022
2ae3a41
Add test to LeftSemiAntiJoinPushDownSuite
EnricoMi Dec 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Adjust test names to one naming scheme
  • Loading branch information
@EnricoMi
EnricoMi committed Dec 19, 2022
commit cc28b21414d5c86753db365c3c6eebb53dea00ab
22 changes: 11 additions & 11 deletions ...rc/test/scala/org/apache/spark/sql/catalyst/optimizer/LeftSemiAntiJoinPushDownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("Project: LeftSemi join no pushdown because of non-deterministic proj exprs") {
test("Project: LeftSemi join no pushdown - non-deterministic proj exprs") {
val originalQuery = testRelation
.select(Rand(1), $"b", $"c")
.join(testRelation1, joinType = LeftSemi, condition = Some($"b" === $"d"))
Expand All @@ -68,7 +68,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Project: LeftSemi join non correlated scalar subq") {
test("Project: LeftSemi join pushdown - non-correlated scalar subq") {
val subq = ScalarSubquery(testRelation.groupBy($"b")(sum($"c").as("sum")).analyze)
val originalQuery = testRelation
.select(subq.as("sum"))
Expand Down Expand Up @@ -109,7 +109,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("Aggregate: LeftSemi join no pushdown due to non-deterministic aggr expressions") {
test("Aggregate: LeftSemi join no pushdown - non-deterministic aggr expressions") {
val originalQuery = testRelation
.groupBy($"b")($"b", Rand(10).as("c"))
.join(testRelation1, joinType = LeftSemi, condition = Some($"b" === $"d"))
Expand Down Expand Up @@ -142,7 +142,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("LeftSemi join over aggregate - no pushdown") {
test("Aggregate: LeftSemi join no pushdown") {
val originalQuery = testRelation
.groupBy($"b")($"b", sum($"c").as("sum"))
.join(testRelation1, joinType = LeftSemi, condition = Some($"b" === $"d" && $"sum" === $"d"))
Expand All @@ -151,7 +151,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Aggregate: LeftSemi join non-correlated scalar subq aggr exprs") {
test("Aggregate: LeftSemi join pushdown - non-correlated scalar subq aggr exprs") {
val subq = ScalarSubquery(testRelation.groupBy($"b")(sum($"c").as("sum")).analyze)
val originalQuery = testRelation
.groupBy($"a") ($"a", subq.as("sum"))
Expand All @@ -166,7 +166,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("LeftSemi join over Window") {
test("Window: LeftSemi join pushdown") {
val winExpr = windowExpr(count($"b"),
windowSpec($"a" :: Nil, $"b".asc :: Nil, UnspecifiedFrame))

Expand All @@ -185,7 +185,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("Window: LeftSemi partial pushdown") {
test("Window: LeftSemi join partial pushdown") {
// Attributes from join condition which does not refer to the window partition spec
// are kept up in the plan as a Filter operator above Window.
val winExpr = windowExpr(count($"b"),
Expand Down Expand Up @@ -292,7 +292,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("Unary: LeftSemi join pushdown - partial pushdown") {
test("Unary: LeftSemi join partial pushdown") {
val testRelationWithArrayType = LocalRelation($"a".int, $"b".int, $"c_arr".array(IntegerType))
val originalQuery = testRelationWithArrayType
.generate(Explode($"c_arr"), alias = Some("arr"), outputNames = Seq("out_col"))
Expand All @@ -309,7 +309,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("Unary: LeftAnti join pushdown - no pushdown") {
test("Unary: LeftAnti join no pushdown") {
val testRelationWithArrayType = LocalRelation($"a".int, $"b".int, $"c_arr".array(IntegerType))
val originalQuery = testRelationWithArrayType
.generate(Explode($"c_arr"), alias = Some("arr"), outputNames = Seq("out_col"))
Expand All @@ -320,7 +320,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Unary: LeftSemi join pushdown - no pushdown") {
test("Unary: LeftSemi join - no pushdown") {
val testRelationWithArrayType = LocalRelation($"a".int, $"b".int, $"c_arr".array(IntegerType))
val originalQuery = testRelationWithArrayType
.generate(Explode($"c_arr"), alias = Some("arr"), outputNames = Seq("out_col"))
Expand All @@ -331,7 +331,7 @@ class LeftSemiAntiJoinPushDownSuite extends PlanTest {
comparePlans(optimized, originalQuery.analyze)
}

test("Unary: LeftSemi join push down through Expand") {
test("Unary: LeftSemi join pushdown through Expand") {
val expand = Expand(Seq(Seq($"a", $"b", "null"), Seq($"a", "null", $"c")),
Seq($"a", $"b", $"c"), testRelation)
val originalQuery = expand
Expand Down