Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,6 @@ abstract class Optimizer(catalogManager: CatalogManager)
EliminateAggregateFilter,
ReorderAssociativeOperator,
LikeSimplification,
NotPropagation,
BooleanSimplification,
SimplifyConditionals,
PushFoldableIntoBranches,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -447,53 +447,6 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
}


/**
* Move/Push `Not` operator if it's beneficial.
*/
object NotPropagation extends Rule[LogicalPlan] {
// Given argument x, return true if expression Not(x) can be simplified
// E.g. let x == Not(y), then canSimplifyNot(x) == true because Not(x) == Not(Not(y)) == y
// For the case of x = EqualTo(a, b), recursively check each child expression
// Extra nullable check is required for EqualNullSafe because
// Not(EqualNullSafe(e, null)) is different from EqualNullSafe(e, Not(null))
private def canSimplifyNot(x: Expression): Boolean = x match {
case Literal(_, BooleanType) | Literal(_, NullType) => true
case _: Not | _: IsNull | _: IsNotNull | _: And | _: Or => true
case _: GreaterThan | _: GreaterThanOrEqual | _: LessThan | _: LessThanOrEqual => true
case EqualTo(a, b) if canSimplifyNot(a) || canSimplifyNot(b) => true
case EqualNullSafe(a, b)
if !a.nullable && !b.nullable && (canSimplifyNot(a) || canSimplifyNot(b)) => true
case _ => false
}

def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(NOT), ruleId) {
case q: LogicalPlan => q.transformExpressionsDownWithPruning(_.containsPattern(NOT), ruleId) {
// Move `Not` from one side of `EqualTo`/`EqualNullSafe` to the other side if it's beneficial.
// E.g. `EqualTo(Not(a), b)` where `b = Not(c)`, it will become
// `EqualTo(a, Not(b))` => `EqualTo(a, Not(Not(c)))` => `EqualTo(a, c)`
// In addition, `if canSimplifyNot(b)` checks if the optimization can converge
// that avoids the situation two conditions are returning to each other.
case EqualTo(Not(a), b) if !canSimplifyNot(a) && canSimplifyNot(b) => EqualTo(a, Not(b))
case EqualTo(a, Not(b)) if canSimplifyNot(a) && !canSimplifyNot(b) => EqualTo(Not(a), b)
case EqualNullSafe(Not(a), b) if !canSimplifyNot(a) && canSimplifyNot(b) =>
EqualNullSafe(a, Not(b))
case EqualNullSafe(a, Not(b)) if canSimplifyNot(a) && !canSimplifyNot(b) =>
EqualNullSafe(Not(a), b)

// Push `Not` to one side of `EqualTo`/`EqualNullSafe` if it's beneficial.
// E.g. Not(EqualTo(x, false)) => EqualTo(x, true)
case Not(EqualTo(a, b)) if canSimplifyNot(b) => EqualTo(a, Not(b))
case Not(EqualTo(a, b)) if canSimplifyNot(a) => EqualTo(Not(a), b)
case Not(EqualNullSafe(a, b)) if !a.nullable && !b.nullable && canSimplifyNot(b) =>
EqualNullSafe(a, Not(b))
case Not(EqualNullSafe(a, b)) if !a.nullable && !b.nullable && canSimplifyNot(a) =>
EqualNullSafe(Not(a), b)
}
}
}


/**
* Simplifies binary comparisons with semantically-equal expressions:
* 1) Replace '<=>' with 'true' literal.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,6 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.optimizer.LikeSimplification" ::
"org.apache.spark.sql.catalyst.optimizer.LimitPushDown" ::
"org.apache.spark.sql.catalyst.optimizer.LimitPushDownThroughWindow" ::
"org.apache.spark.sql.catalyst.optimizer.NotPropagation" ::
"org.apache.spark.sql.catalyst.optimizer.NullDownPropagation" ::
"org.apache.spark.sql.catalyst.optimizer.NullPropagation" ::
"org.apache.spark.sql.catalyst.optimizer.ObjectSerializerPruning" ::
Expand Down

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ class NullDownPropagationSuite extends PlanTest with ExpressionEvalHelper {
ConstantFolding,
SimplifyConditionals,
BooleanSimplification,
NotPropagation,
PruneFilters) :: Nil
}

Expand Down
26 changes: 26 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -1956,4 +1956,30 @@ class SubquerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
assert(!nonDeterministicQueryPlan.deterministic)
}

test("SPARK-38132: Not IN subquery correctness checks") {
val t = "test_table"
withTable(t) {
Seq[(Integer, Integer)](
(1, 1),
(2, 2),
(3, 3),
(4, null),
(null, 0))
.toDF("c1", "c2").write.saveAsTable(t)
val df = spark.table(t)

checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t)) = true"), Seq.empty)
checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t WHERE c2 IS NOT NULL)) = true"),
Row(4, null) :: Nil)
checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t)) <=> true"), Seq.empty)
checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t WHERE c2 IS NOT NULL)) <=> true"),
Row(4, null) :: Nil)
checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t)) != false"), Seq.empty)
checkAnswer(df.where(s"(c1 NOT IN (SELECT c2 FROM $t WHERE c2 IS NOT NULL)) != false"),
Row(4, null) :: Nil)
checkAnswer(df.where(s"NOT((c1 NOT IN (SELECT c2 FROM $t)) <=> false)"), Seq.empty)
checkAnswer(df.where(s"NOT((c1 NOT IN (SELECT c2 FROM $t WHERE c2 IS NOT NULL)) <=> false)"),
Row(4, null) :: Nil)
}
}
}