cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3830205367
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +129,25 @@ case class InSubqueryExec(
@transient private lazy val inSet = InSet(child, result.toSet)
- override def nullable: Boolean = child.nullable
+ // Mirror the logical InSubquery.nullable: nullable when any output column
is nullable
+ // (null in any column position produces UNKNOWN on a miss) or when any LHS
field is nullable.
+ // For multi-column IN the LHS is a CreateNamedStruct whose top-level
nullable is always false
+ // even when individual field expressions are nullable (SPARK-58481).
PlanSubqueries is the only
+ // producer of multi-column InSubqueryExec and always wraps the LHS values
in CreateNamedStruct,
+ // so matching on it here is both precise and exhaustive. The fallback to
child.nullable is safe
Review Comment:
**Nit:**
`PlanAdaptiveSubqueries` also constructs multi-column `InSubqueryExec` with
a `CreateNamedStruct` LHS. Please describe the invariant shared by both
producers.
```suggestion
// even when individual field expressions are nullable (SPARK-58481). Both
PlanSubqueries and
// PlanAdaptiveSubqueries wrap multi-column LHS values in
CreateNamedStruct, so matching on it
// here is precise for the current producers. The fallback to
child.nullable is safe
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +187,152 @@ case class InSubqueryExec(
}
}
+ // Invariant schema/ordering data for the multi-column evaluator, computed
once after the result
+ // is available. @transient so that serialization (result=null) does not
trigger evaluation.
+ @transient private lazy val multiColFieldTypes: Array[DataType] =
+ plan.output.map(_.dataType).toArray
+ @transient private lazy val multiColFieldOrderings: Array[Ordering[Any]] =
+ multiColFieldTypes.map(TypeUtils.getInterpretedOrdering)
+ // Struct-level ordering used to index fully non-null result rows in a
TreeSet.
+ @transient private lazy val multiColRowOrdering: Ordering[InternalRow] =
+
TypeUtils.getInterpretedOrdering(child.dataType).asInstanceOf[Ordering[InternalRow]]
+
+ // Split collected rows into a sorted set of fully non-null rows (O(log n)
membership test)
+ // and an array of rows that contain at least one null field (must be
scanned linearly).
+ // Built once; the TreeSet uses the struct-level Catalyst ordering. See
SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = Array.newBuilder[InternalRow]
+ val nonNull =
result.foldLeft(TreeSet.empty[InternalRow](multiColRowOrdering)) { (s, r) =>
+ val row = r.asInstanceOf[InternalRow]
+ if (row.anyNull) { withNull += row; s } else s + row
+ }
+ (nonNull, withNull.result())
+ }
+
+ // Three-valued IN semantics for multi-column subqueries.
+ // Result rows are InternalRow objects; InSet's TreeSet treats null fields
as non-equal and
+ // cannot distinguish a definitively-false candidate from an indeterminate
one.
Review Comment:
**Nit:**
Catalyst's interpreted struct ordering treats corresponding NULL fields as
equal. `InSet` is unsuitable because membership cannot distinguish an UNKNOWN
row comparison from FALSE.
```suggestion
// Result rows are InternalRow objects; InSet's TreeSet uses Catalyst
ordering, but membership
// cannot distinguish a definitively-false candidate from an indeterminate
one.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,111 @@ class SubquerySuite extends SharedSparkSession
assert(exposedAttribute.exprId == outerReferenceAttribute.exprId)
}
+
+ test("SPARK-58481: InSubqueryExec nullable correctly accounts for subquery
output nullability") {
+ // 5 NOT IN (99, NULL) is UNKNOWN, not TRUE or FALSE. A join condition
that is not TRUE
+ // matches no rows, so a FULL OUTER JOIN must emit null-padded rows for
every row in each
+ // side -- 3 + 3 = 6 null-padded rows -- not the full cross product (9
rows).
+ withTable("t0", "t1", "t3") {
+ sql("CREATE TABLE t0(c0 INT) USING PARQUET")
+ sql("INSERT INTO t0 VALUES (1), (2), (3)")
+ sql("CREATE TABLE t1(c0 INT) USING PARQUET")
+ sql("INSERT INTO t1 VALUES (10), (20), (30)")
+ sql("CREATE TABLE t3(c0 INT) USING PARQUET")
+ sql("INSERT INTO t3 VALUES (99), (CAST(NULL AS INT))")
+
+ // t1 rows are null-padded (no match on left), t0 rows are null-padded
(no match on right).
+ val expected = Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30), // t0 side: null-padded
+ Row(1, null), Row(2, null), Row(3, null)) // t1 side: null-padded
+ checkAnswer(
+ sql("SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN
(SELECT t3.c0 FROM t3))"),
+ expected)
+ }
+ }
+
+ test("SPARK-58481: multi-column IN subquery with nullable non-head output is
nullable") {
+ // Disable the optimizer's join-condition IN rewrite so the query
exercises InSubqueryExec.
+ // Covers both per-candidate cases:
+ // (1,1) vs (99,NULL): first field differs => definitely FALSE (not
UNKNOWN).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (2,2) vs either row: both are FALSE => NOT IN = TRUE.
+ // Expected: (1,1) gets UNKNOWN => null-padded; (2,2) gets TRUE => joined.
+ withSQLConf(
+
"spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled"
-> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (2, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT) USING PARQUET")
+ // (99, 99): definitively not equal to any lhs row (first field
differs from both).
+ // (1, NULL): first field equals lhs(1,1).a; second is null => UNKNOWN
for (1,1).
+ // first field 1 != 2 => FALSE for (2,2).
+ sql("INSERT INTO rhs VALUES (99, 99), (1, CAST(NULL AS INT))")
+
+ // (1,1): UNKNOWN (indeterminate against (1,NULL)) => null-padded on
both sides.
Review Comment:
**Nit:**
Only `(1,1)` is null-padded. Both RHS rows match `(2,2)`, as the three-row
assertion shows.
```suggestion
// (1,1): UNKNOWN (indeterminate against (1,NULL)) => null-padded.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +187,152 @@ case class InSubqueryExec(
}
}
+ // Invariant schema/ordering data for the multi-column evaluator, computed
once after the result
+ // is available. @transient so that serialization (result=null) does not
trigger evaluation.
+ @transient private lazy val multiColFieldTypes: Array[DataType] =
+ plan.output.map(_.dataType).toArray
+ @transient private lazy val multiColFieldOrderings: Array[Ordering[Any]] =
+ multiColFieldTypes.map(TypeUtils.getInterpretedOrdering)
+ // Struct-level ordering used to index fully non-null result rows in a
TreeSet.
+ @transient private lazy val multiColRowOrdering: Ordering[InternalRow] =
+
TypeUtils.getInterpretedOrdering(child.dataType).asInstanceOf[Ordering[InternalRow]]
+
+ // Split collected rows into a sorted set of fully non-null rows (O(log n)
membership test)
+ // and an array of rows that contain at least one null field (must be
scanned linearly).
+ // Built once; the TreeSet uses the struct-level Catalyst ordering. See
SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = Array.newBuilder[InternalRow]
+ val nonNull =
result.foldLeft(TreeSet.empty[InternalRow](multiColRowOrdering)) { (s, r) =>
+ val row = r.asInstanceOf[InternalRow]
+ if (row.anyNull) { withNull += row; s } else s + row
+ }
+ (nonNull, withNull.result())
+ }
+
+ // Three-valued IN semantics for multi-column subqueries.
+ // Result rows are InternalRow objects; InSet's TreeSet treats null fields
as non-equal and
+ // cannot distinguish a definitively-false candidate from an indeterminate
one.
+ //
+ // When the LHS struct has no null fields:
+ // Fast path: O(log n) TreeSet lookup against fully non-null result rows
for TRUE.
+ // Slow path: linear scan over null-containing result rows only for
potential UNKNOWN.
+ //
+ // When the LHS struct has at least one null field, the fast path cannot be
used (a null LHS
+ // field produces UNKNOWN against any non-null RHS row whose non-null fields
all match). In
+ // that case we scan all result rows linearly.
+ //
+ // Per-candidate three-valued logic: TRUE if every field matches; UNKNOWN if
no field is
+ // definitively unequal but at least one comparison involves null; FALSE
otherwise.
+ private def evalMultiColumn(inputRow: InternalRow): Any = {
+ val value = child.eval(inputRow)
+ if (value == null) return null
+ val inputStruct = value.asInstanceOf[InternalRow]
+ val fieldTypes = multiColFieldTypes
+ val orderings = multiColFieldOrderings
+ val numFields = fieldTypes.length
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (multiColNonNullSet.contains(inputStruct)) return true
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ var hasUnknown = false
+ var i = 0
+ while (i < multiColNullRows.length) {
+ val candidate = multiColNullRows(i)
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
+ if (candidateField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputStruct.get(fieldIdx, fieldTypes(fieldIdx)), candidateField)
!= 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ i += 1
+ }
+ if (hasUnknown) null else false
+ } else {
+ // LHS has at least one null field: must scan all result rows because a
null LHS field
+ // produces UNKNOWN against any non-null RHS row whose other fields all
match.
+ var hasUnknown = false
+ // Scan null-containing result rows first.
+ var i = 0
+ while (i < multiColNullRows.length && !hasUnknown) {
+ val candidate = multiColNullRows(i)
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputStruct.get(fieldIdx, fieldTypes(fieldIdx))
+ val candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
+ if (candidateField == null || inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(inputField, candidateField)
!= 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ i += 1
+ }
+ // Scan fully non-null result rows: a null LHS field is UNKNOWN unless a
prior field differs.
Review Comment:
**Nit:**
A mismatch after the NULL field also makes this candidate FALSE because the
loop continues.
```suggestion
// Scan non-null rows: a null LHS comparison is UNKNOWN unless a
non-null field differs.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +187,152 @@ case class InSubqueryExec(
}
}
+ // Invariant schema/ordering data for the multi-column evaluator, computed
once after the result
+ // is available. @transient so that serialization (result=null) does not
trigger evaluation.
+ @transient private lazy val multiColFieldTypes: Array[DataType] =
+ plan.output.map(_.dataType).toArray
+ @transient private lazy val multiColFieldOrderings: Array[Ordering[Any]] =
+ multiColFieldTypes.map(TypeUtils.getInterpretedOrdering)
+ // Struct-level ordering used to index fully non-null result rows in a
TreeSet.
+ @transient private lazy val multiColRowOrdering: Ordering[InternalRow] =
+
TypeUtils.getInterpretedOrdering(child.dataType).asInstanceOf[Ordering[InternalRow]]
+
+ // Split collected rows into a sorted set of fully non-null rows (O(log n)
membership test)
+ // and an array of rows that contain at least one null field (must be
scanned linearly).
+ // Built once; the TreeSet uses the struct-level Catalyst ordering. See
SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = Array.newBuilder[InternalRow]
+ val nonNull =
result.foldLeft(TreeSet.empty[InternalRow](multiColRowOrdering)) { (s, r) =>
+ val row = r.asInstanceOf[InternalRow]
+ if (row.anyNull) { withNull += row; s } else s + row
+ }
+ (nonNull, withNull.result())
+ }
+
+ // Three-valued IN semantics for multi-column subqueries.
+ // Result rows are InternalRow objects; InSet's TreeSet treats null fields
as non-equal and
+ // cannot distinguish a definitively-false candidate from an indeterminate
one.
+ //
+ // When the LHS struct has no null fields:
+ // Fast path: O(log n) TreeSet lookup against fully non-null result rows
for TRUE.
+ // Slow path: linear scan over null-containing result rows only for
potential UNKNOWN.
+ //
+ // When the LHS struct has at least one null field, the fast path cannot be
used (a null LHS
+ // field produces UNKNOWN against any non-null RHS row whose non-null fields
all match). In
+ // that case we scan all result rows linearly.
+ //
+ // Per-candidate three-valued logic: TRUE if every field matches; UNKNOWN if
no field is
+ // definitively unequal but at least one comparison involves null; FALSE
otherwise.
+ private def evalMultiColumn(inputRow: InternalRow): Any = {
+ val value = child.eval(inputRow)
+ if (value == null) return null
+ val inputStruct = value.asInstanceOf[InternalRow]
+ val fieldTypes = multiColFieldTypes
+ val orderings = multiColFieldOrderings
+ val numFields = fieldTypes.length
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (multiColNonNullSet.contains(inputStruct)) return true
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ var hasUnknown = false
+ var i = 0
+ while (i < multiColNullRows.length) {
Review Comment:
**Non-blocking:**
Stop this scan once `hasUnknown` is set. The indexed non-null lookup already
ruled out TRUE, and every row here contains NULL, so later candidates cannot
improve UNKNOWN to TRUE.
```suggestion
while (i < multiColNullRows.length && !hasUnknown) {
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +129,25 @@ case class InSubqueryExec(
@transient private lazy val inSet = InSet(child, result.toSet)
- override def nullable: Boolean = child.nullable
+ // Mirror the logical InSubquery.nullable: nullable when any output column
is nullable
+ // (null in any column position produces UNKNOWN on a miss) or when any LHS
field is nullable.
+ // For multi-column IN the LHS is a CreateNamedStruct whose top-level
nullable is always false
+ // even when individual field expressions are nullable (SPARK-58481).
PlanSubqueries is the only
+ // producer of multi-column InSubqueryExec and always wraps the LHS values
in CreateNamedStruct,
+ // so matching on it here is both precise and exhaustive. The fallback to
child.nullable is safe
+ // for the single-column case where child is the bare LHS expression.
+ // Respects LEGACY_IN_SUBQUERY_NULLABILITY to stay in sync with the logical
node.
+ override def nullable: Boolean = {
+ if (!SQLConf.get.getConf(SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY)) {
Review Comment:
**Non-blocking:**
Please add a focused regression with
`spark.sql.legacy.inSubqueryNullability=true`. This branch intentionally
restores child-only nullability, but all changed tests use the default mode, so
the compatibility contract can drift without detection.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
