cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3853535838
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +129,24 @@ 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
+ // (a nullable RHS field can produce 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). 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
+ // for the single-column case where child is the bare LHS expression.
+ // LEGACY_IN_SUBQUERY_NULLABILITY suppresses only RHS-derived nullability;
LHS field nullability
+ // is preserved in both modes so that NOT IN on a nullable LHS field always
propagates UNKNOWN.
+ override def nullable: Boolean = {
+ val lhsNullable = child match {
+ case cns: CreateNamedStruct => cns.valExprs.exists(_.nullable)
Review Comment:
**Blocking:**
Gate this field-level nullability branch on `plan.output.length > 1`. A
user-written single struct-valued IN expression is also a `CreateNamedStruct`,
but its top-level value is non-null even when a field is nullable. In legacy
mode, this branch therefore makes the physical expression nullable when logical
`InSubquery` is not, so generated `NOT IN` can propagate NULL instead of
preserving the compatibility result.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +186,174 @@ 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 = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ result.foreach { r =>
+ val row = r.asInstanceOf[InternalRow]
+ if (row.anyNull) withNull += row else nonNull += row
+ }
+ (nonNull.result(), withNull.result())
+ }
+
+ // Three-valued IN semantics for multi-column subqueries.
+ // Result rows are InternalRow objects; InSet's TreeSet uses Catalyst
ordering, but membership
+ // 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). Both
+ // sets of result rows are scanned linearly, stopping once UNKNOWN is
established.
+ //
+ // 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
+ // Materialize LHS fields once before the candidate scans to avoid
repeated get() calls
+ // inside the per-candidate loop. Placed after the TreeSet lookup so
exact-match hits
+ // pay nothing.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i,
fieldTypes(i)))
Review Comment:
**Non-blocking:**
Return `false` before materializing `inputFields` when there are no
candidates to scan. This branch still allocates and fills the array whenever
`multiColNullRows` is empty; the nullable-LHS branch at `subquery.scala:269`
has the same issue when both result collections are empty.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,171 @@ 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))")
+
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ val expected = Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30), // unmatched t1, t0
column null-padded
+ Row(1, null), Row(2, null), Row(3, null)) // unmatched t0, t1
column 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.
+ // Use VALUES-derived temp views: their nullability is inferred from the
literals (no NULL
+ // literal => non-nullable), rather than declared and then widened.
Parquet file-source
+ // analysis applies dataSchema.asNullable regardless of DDL NOT NULL,
which would defeat
+ // the nullability control this test relies on.
+ // Covers both per-candidate cases:
+ // (1,1) vs (99,99): first field differs => definitely FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (2,2) vs either row: both FALSE => NOT IN = TRUE.
+ // Expected: (1,1) gets UNKNOWN => null-padded; (2,2) gets TRUE => joined
with both rhs rows.
+ withSQLConf(
+
"spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled"
-> "false"
+ ) {
+ withTempView("lhs", "rhs") {
+ // VALUES-derived views: the VALUES output schema is inferred as
non-nullable for
+ // columns with no NULLs, while CAST(NULL AS INT) makes that column
nullable. This
+ // preserves the intended nullability without Parquet's asNullable
coercion.
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2)
AS t(a, b)")
+ // (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(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a,
b)""".stripMargin)
+
+ // (1,1): UNKNOWN (indeterminate against (1,NULL)) => null-padded.
+ // (2,2): TRUE (definitively not in set) => joins with both rhs rows.
+ checkAnswer(
+ sql(
+ """SELECT lhs.a, rhs.a FROM lhs FULL OUTER JOIN rhs
+ |ON ((lhs.a, lhs.b) NOT IN (SELECT a, b FROM
rhs))""".stripMargin),
+ Seq(Row(1, null), Row(2, 99), Row(2, 1)))
+ }
+ }
+ }
+
+ test("SPARK-58481: multi-column IN subquery uses Catalyst ordering for
BinaryType fields") {
+ // Object.equals on Array[Byte] compares by identity, not value; Catalyst
ordering compares
+ // by content. A multi-column IN where one field is BinaryType would
incorrectly return FALSE
+ // (no match) with JVM equality even when the bytes are equal. Use an
inner join to keep the
+ // assertion simple: the join condition is TRUE iff the IN match succeeds.
+ withSQLConf(
+
"spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled"
-> "false"
+ ) {
+ withTable("lbin", "rbin") {
+ sql("CREATE TABLE lbin(id INT NOT NULL, b BINARY NOT NULL) USING
PARQUET")
+ sql("INSERT INTO lbin VALUES (1, X'01')")
+ sql("CREATE TABLE rbin(id INT NOT NULL, b BINARY NOT NULL) USING
PARQUET")
+ sql("INSERT INTO rbin VALUES (1, X'01')")
+ // (1, 0x01) IN ((1, 0x01)) must be TRUE; the join should return one
row.
+ checkAnswer(
+ sql(
+ """SELECT lbin.id FROM lbin JOIN rbin
+ |ON ((lbin.id, lbin.b) IN (SELECT id, b FROM
rbin))""".stripMargin),
+ Seq(Row(1)))
+ }
+ }
+ }
+
+ test("SPARK-58481: multi-column NOT IN with nullable LHS and non-nullable
RHS is nullable") {
+ // CreateNamedStruct.nullable is always false, so child.nullable would
return false for a
+ // multi-column LHS even when individual fields are nullable. The
generated NOT IN code
+ // would then suppress null handling and turn UNKNOWN into TRUE, producing
wrong results.
+ // Fixture: lhs.a is nullable; rhs columns are NOT NULL.
Review Comment:
**Non-blocking:**
Use a non-file-source RHS, such as a VALUES-derived temp view, so this
regression actually keeps the RHS attributes non-nullable. With the default
file-source setting, Parquet table analysis applies `dataSchema.asNullable`, so
this test can pass from RHS-derived nullability even if the LHS-field term
regresses.
--
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]
