cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3904577830
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,223 @@ 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)
Review Comment:
**Non-blocking (P2):** Please also cover the opposite precedence orders
introduced by the new evaluator. For `(1, 1)` against `(NULL, 99)`, the later
mismatch must make the candidate FALSE despite the earlier UNKNOWN; against
both `(1, NULL)` and `(1, 1)`, the exact row must make IN TRUE despite the
UNKNOWN row. The current fixtures only put NULL after the decisive mismatch and
never combine UNKNOWN with an exact match, so field-order or candidate-order
regressions can remain green.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,185 @@ 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]
Review Comment:
**Non-blocking (P2):** Could this side be deduplicated with
`multiColRowOrdering` as well? Duplicate multiplicity cannot change the
existential three-valued IN result, but every outer row currently rescans every
duplicate null-containing RHS row. For M outer rows and K copies of `(NULL,
3)`, a miss such as `(7, 2)` performs O(M*K*F) duplicate comparisons even
though one retained copy yields the same result.
--
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]
