Skip to content

[SPARK-33847][SQL] Simplify CaseWhen if elseValue is None #30852

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
wants to merge 10 commits into from
Closed

[SPARK-33847][SQL] Simplify CaseWhen if elseValue is None #30852

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
09edff5
Replace None of elseValue inside CaseWhen if all branches are FalseLi…
wangyum Dec 19, 2020
ca2a224
Merge remote-tracking branch 'upstream/master' into SPARK-33847
wangyum Dec 21, 2020
3c5f3da
Fix
wangyum Dec 21, 2020
b837e37
fix
wangyum Dec 21, 2020
d07344f
Merge remote-tracking branch 'upstream/master' into SPARK-33847
wangyum Dec 22, 2020
81c38f8
fix
wangyum Dec 22, 2020
7f3529d
fix
wangyum Dec 23, 2020
1684d67
Merge remote-tracking branch 'upstream/master' into SPARK-33847
wangyum Dec 23, 2020
0933019
fix
wangyum Dec 23, 2020
d20fede
remove
wangyum Dec 23, 2020
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
1 change: 1 addition & 0 deletions .../main/scala/org/apache/spark/sql/catalyst/optimizer/ReplaceNullWithFalseInPredicate.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ object ReplaceNullWithFalseInPredicate extends Rule[LogicalPlan] {
replaceNullWithFalse(cond) -> replaceNullWithFalse(value)
}
val newElseValue = cw.elseValue.map(replaceNullWithFalse)
.orElse(if (newBranches.forall(_._2 == FalseLiteral)) Some(FalseLiteral) else None)
Copy link
Contributor

@cloud-fan cloud-fan Dec 22, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change assumes that other rules can optimize CASE WHEN with false in all branches, which violates the catalyst principle that rules should be orthogonal. Can we think of a better way?

Copy link
Contributor

@cloud-fan cloud-fan Dec 22, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here we do need the logic from 2 optimizer rules:

  1. null means false in predicates
  2. eliminate CASE WHEN if all branches return the same value

We can probably include logic 2 in this rule

...
if (newBranches.forall(_._2 == FalseLiteral) && cw.elseValue.isEmpty) {
  FalseLiteral
} else {
  val newElseValue = cw.elseValue.map(replaceNullWithFalse)
  CaseWhen(newBranches, newElseValue)
}

Copy link
Member Author

@wangyum wangyum Dec 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. This change assumes CaseWhen has optimized by PushFoldableIntoBranches, SimplifyConditionals and ConstantFolding.

CaseWhen(newBranches, newElseValue)
case i @ If(pred, trueVal, falseVal) if i.dataType == BooleanType =>
If(replaceNullWithFalse(pred), replaceNullWithFalse(trueVal), replaceNullWithFalse(falseVal))
Expand Down
83 changes: 70 additions & 13 deletions .../scala/org/apache/spark/sql/catalyst/optimizer/ReplaceNullWithFalseInPredicateSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions.{And, ArrayExists, ArrayFilter, ArrayTransform, CaseWhen, Expression, GreaterThan, If, LambdaFunction, Literal, MapFilter, NamedExpression, Or, UnresolvedNamedLambdaVariable}
import org.apache.spark.sql.catalyst.expressions.{And, ArrayExists, ArrayFilter, ArrayTransform, CaseWhen, EqualTo, Expression, GreaterThan, If, LambdaFunction, Literal, MapFilter, NamedExpression, Or, UnresolvedNamedLambdaVariable}
import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral, TrueLiteral}
import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
import org.apache.spark.sql.catalyst.plans.logical.{DeleteFromTable, LocalRelation, LogicalPlan, UpdateTable}
Expand All @@ -38,6 +38,7 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
ConstantFolding,
BooleanSimplification,
SimplifyConditionals,
PushFoldableIntoBranches,
ReplaceNullWithFalseInPredicate) :: Nil
}

Expand Down Expand Up @@ -222,10 +223,17 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
Literal(null, IntegerType),
Literal(3)),
FalseLiteral)
testFilter(originalCond = condition, expectedCond = condition)
testJoin(originalCond = condition, expectedCond = condition)
testDelete(originalCond = condition, expectedCond = condition)
testUpdate(originalCond = condition, expectedCond = condition)
val expectedCond = If(
Copy link
Contributor

@cloud-fan cloud-fan Dec 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to update the test name. Now it's inability to replace null in non-boolean branches of If

Copy link
Contributor

@cloud-fan cloud-fan Dec 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

or we change this test to preserve the original purpose.

Copy link
Member Author

@wangyum wangyum Dec 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I removed PushFoldableIntoBranches from this test suite, and add integration test to ReplaceNullWithFalseInPredicateEndToEndSuite.

UnresolvedAttribute("i") > Literal(10),
If(
UnresolvedAttribute("i") === Literal(15),
FalseLiteral,
TrueLiteral),
FalseLiteral)
testFilter(originalCond = condition, expectedCond = expectedCond)
testJoin(originalCond = condition, expectedCond = expectedCond)
testDelete(originalCond = condition, expectedCond = expectedCond)
testUpdate(originalCond = condition, expectedCond = expectedCond)
}

test("inability to replace null in non-boolean values of CaseWhen") {
Expand All @@ -238,10 +246,18 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
FalseLiteral)
val branches = Seq((UnresolvedAttribute("i") > Literal(10)) -> branchValue)
val condition = CaseWhen(branches)
testFilter(originalCond = condition, expectedCond = condition)
testJoin(originalCond = condition, expectedCond = condition)
testDelete(originalCond = condition, expectedCond = condition)
testUpdate(originalCond = condition, expectedCond = condition)

val expectedCond = CaseWhen(Seq((UnresolvedAttribute("i") > Literal(10)) -> If(
CaseWhen(
Seq((UnresolvedAttribute("i") > Literal(20)) -> TrueLiteral),
FalseLiteral),
TrueLiteral,
FalseLiteral)))

testFilter(originalCond = condition, expectedCond = expectedCond)
testJoin(originalCond = condition, expectedCond = expectedCond)
testDelete(originalCond = condition, expectedCond = expectedCond)
testUpdate(originalCond = condition, expectedCond = expectedCond)
}

test("inability to replace null in non-boolean branches of If inside another If") {
Expand All @@ -252,10 +268,17 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
Literal(3)),
TrueLiteral,
FalseLiteral)
testFilter(originalCond = condition, expectedCond = condition)
testJoin(originalCond = condition, expectedCond = condition)
testDelete(originalCond = condition, expectedCond = condition)
testUpdate(originalCond = condition, expectedCond = condition)
val expectedCond = If(
If(
UnresolvedAttribute("i") === Literal(15),
FalseLiteral,
TrueLiteral),
TrueLiteral,
FalseLiteral)
testFilter(originalCond = condition, expectedCond = expectedCond)
testJoin(originalCond = condition, expectedCond = expectedCond)
testDelete(originalCond = condition, expectedCond = expectedCond)
testUpdate(originalCond = condition, expectedCond = expectedCond)
}

test("replace null in If used as a join condition") {
Expand Down Expand Up @@ -375,6 +398,40 @@ class ReplaceNullWithFalseInPredicateSuite extends PlanTest {
testProjection(originalExpr = column, expectedExpr = column)
}

test("replace None of elseValue inside CaseWhen if all branches are FalseLiteral") {
val allFalseBranches = Seq(
(UnresolvedAttribute("i") < Literal(10)) -> FalseLiteral,
(UnresolvedAttribute("i") > Literal(40)) -> FalseLiteral)
val allFalseCond = CaseWhen(allFalseBranches)

val nonAllFalseBranches = Seq(
(UnresolvedAttribute("i") < Literal(10)) -> FalseLiteral,
(UnresolvedAttribute("i") > Literal(40)) -> TrueLiteral)
val nonAllFalseCond = CaseWhen(nonAllFalseBranches)

testFilter(allFalseCond, FalseLiteral)
testJoin(allFalseCond, FalseLiteral)
testDelete(allFalseCond, FalseLiteral)
testUpdate(allFalseCond, FalseLiteral)

testFilter(nonAllFalseCond, nonAllFalseCond)
testJoin(nonAllFalseCond, nonAllFalseCond)
testDelete(nonAllFalseCond, nonAllFalseCond)
testUpdate(nonAllFalseCond, nonAllFalseCond)
}

test("replace None of elseValue inside CaseWhen with PushFoldableIntoBranches") {
val allFalseBranches = Seq(
(UnresolvedAttribute("i") < Literal(10)) -> Literal("a"),
(UnresolvedAttribute("i") > Literal(40)) -> Literal("b"))
val allFalseCond = EqualTo(CaseWhen(allFalseBranches), "c")

testFilter(allFalseCond, FalseLiteral)
testJoin(allFalseCond, FalseLiteral)
testDelete(allFalseCond, FalseLiteral)
testUpdate(allFalseCond, FalseLiteral)
}

private def testFilter(originalCond: Expression, expectedCond: Expression): Unit = {
test((rel, exp) => rel.where(exp), originalCond, expectedCond)
}
Expand Down