You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anshulbaliga7 (via GitHub)" <gi...@apache.org> on 2026/08/18 10:04:03 UTC
[PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
anshulbaliga7 opened a new pull request, #58077:
URL: https://github.com/apache/spark/pull/58077
### What changes were proposed in this pull request?
Fix `InSubqueryExec.nullable` to account for whether the subquery output can produce NULLs, not just whether the child expression is nullable.
### Why are the changes needed?
`x IN (subquery)` evaluates to UNKNOWN when no match is found and the subquery result contains NULL. `InSubqueryExec.nullable` was declared as `child.nullable`, ignoring the subquery output nullability. This caused codegen to emit `isNull = false` unconditionally, computing `!false = true` instead of UNKNOWN for `NOT IN` expressions, returning wrong results.
The interpreted path was correct because it always checks the runtime null value regardless of the declared nullability, which is why `spark.sql.codegen.factoryMode=NO_CODEGEN` returned correct results.
### Does this PR introduce any user-facing change?
Yes. Queries using `NOT IN (subquery)` with NULLs in the subquery result on FULL OUTER JOIN conditions now return correct results.
### How was this patch tested?
Added regression test in `SubquerySuite` using `checkAnswer` with actual null-padded rows, and SQL golden file test in `in-joins.sql`.
### Was this patch authored or co-authored using generative AI tooling?
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5498929750
Thanks @cloud-fan for the review, the nondeterministic regression now constructs `InSubqueryExec` directly with a two-column subquery output, so `doGenCode` takes the multi-column fallback branch (not `InSet.doGenCode`, where Rand would self-register regardless of this fix).
Confirmed by temporarily removing the registration loop and verifying the test fails without it.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3933752087
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -188,6 +267,161 @@ case class InSubqueryExec(
copy(child = newChild)
}
+/**
+ * Lightweight, serializable evaluator for multi-column IN subqueries.
+ * Holds only what executors need: the LHS expression, collected result rows, field metadata,
+ * and the legacy empty-list behavior flag. Excludes plan (BaseSubqueryExec) so that
+ * task-closure serialization does not carry the driver's physical plan subtree.
+ * All @transient lazy fields are rebuilt from the serialized fieldTypes and resultRows
+ * after deserialization, mirroring InSubqueryExec's own pattern. See SPARK-58481.
+ */
+private[sql] class MultiColumnInSubqueryEvaluator(
+ val child: Expression,
+ val fieldTypes: Array[DataType],
+ val resultRows: Array[InternalRow],
+ val legacyNullInEmptyBehavior: Boolean) extends Serializable {
+
+ @transient private lazy val fieldOrderings: Array[Ordering[Any]] =
+ fieldTypes.map(TypeUtils.getInterpretedOrdering)
+
+ @transient private lazy val rowOrdering: Ordering[InternalRow] =
+ TypeUtils.getInterpretedOrdering(child.dataType).asInstanceOf[Ordering[InternalRow]]
+
+ // Split collected rows using the struct-level Catalyst ordering so that duplicates are
+ // deduplicated. Fully non-null rows go into nonNullSet (TreeSet) for O(log n)
+ // membership tests. Null-containing rows are deduplicated via a temporary TreeSet and
+ // then stored as nullRows (Array) for linear scanning; each distinct null-containing row
+ // is thus scanned at most once per outer row regardless of RHS duplicate multiplicity.
+ @transient private lazy val (nonNullSet, nullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](rowOrdering)
+ val nonNull = TreeSet.newBuilder[InternalRow](rowOrdering)
+ resultRows.foreach { r => if (r.anyNull) withNull += r else nonNull += r }
+ (nonNull.result(), withNull.result().toArray)
+ }
+
+ // 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 set.
+ //
+ // 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.
+ def eval(inputRow: InternalRow): Any = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is
Review Comment:
**Nit (P3):** ANSI controls only the default here; an explicit spark.sql.legacy.nullInEmptyListBehavior=true still evaluates the LHS when ANSI is enabled, as the test below demonstrates. Please describe this branch as legacyNullInEmptyBehavior=false and mention ANSI only as the defaulting rule.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,61 @@ case class InSubqueryExec(
}
}
+ // Memoized lightweight evaluator instance. @transient so it is rebuilt (from the
+ // serializable fields) when InSubqueryExec is deserialized in an executor; constructed
+ // only for the multi-column path after prepareResult() has populated result.
+ @transient private lazy val multiColEvaluator: MultiColumnInSubqueryEvaluator = {
+ val rows = result.map(_.asInstanceOf[InternalRow])
+ val fTypes = plan.output.map(_.dataType).toArray
+ new MultiColumnInSubqueryEvaluator(child, fTypes, rows, legacyNullInEmptyBehavior)
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ multiColEvaluator.eval(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path via the lightweight evaluator.
+ // Register the evaluator (not the full InSubqueryExec) so that task closures do not
Review Comment:
**Non-blocking (P2):** The evaluator still captures the entire child expression. For a multi-column LHS containing a scalar subquery, PlanSubqueries embeds an execution ScalarSubquery whose non-transient BaseSubqueryExec remains reachable through this reference, so driver plan state is still serialized with generated tasks. Please remove prepared scalar-subquery plans from the captured child and add a regression that proves no BaseSubqueryExec remains reachable.
**Recommended change:** Sanitize the evaluator child by literalizing prepared execution ScalarSubquery nodes before it enters CodegenContext.references.
**Why this works:** Transform the prepared child to replace each execution ScalarSubquery with its toLiteral result, then discover and register nondeterministic descendants from the transformed child.
**Scope:** InSubqueryExec generated evaluation and SubquerySuite serialization coverage for multi-column left-hand expressions containing scalar subqueries.
**Compatibility:** Preserve SQL semantics, legacy branches, and the single-column InSet path; only remove already-evaluated scalar-subquery plan objects from the serialized evaluator graph.
**Risks:** Nondeterministic expression identity must remain consistent with partition initialization. Literal replacement must occur only after scalar subquery results are available.
**Constraints:** No BaseSubqueryExec may remain reachable through CodegenContext.references. Existing nullability and empty-result compatibility behavior must remain unchanged.
**Success:** A reachable multi-column scalar-subquery LHS serializes and executes with no BaseSubqueryExec reachable from task references, while the existing evaluator behavior remains intact.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -124,8 +128,36 @@ case class InSubqueryExec(
extends ExecSubqueryExpression with UnaryLike[Expression] with Predicate {
@transient private lazy val inSet = InSet(child, result.toSet)
+ // Captured at construction time to mirror InSet's empty-result short-circuit (SPARK-44550).
+ private val legacyNullInEmptyBehavior = SQLConf.get.legacyNullInEmptyBehavior
- 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 a nullable LHS field can contribute UNKNOWN
+ // when no field differs.
+ // For the multi-column case (plan.output.length > 1) the LHS is always a generated
+ // CreateNamedStruct; a user-written single struct-valued IN also produces a CreateNamedStruct,
Review Comment:
**Nit (P3):** A single struct-valued operand does not necessarily produce CreateNamedStruct: both planners use values.head unchanged when values.length == 1, so a struct attribute remains an AttributeReference. Please say that only multi-value IN is wrapped and that the single-value case explains the plan.output.length guard.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,61 @@ case class InSubqueryExec(
}
}
+ // Memoized lightweight evaluator instance. @transient so it is rebuilt (from the
+ // serializable fields) when InSubqueryExec is deserialized in an executor; constructed
+ // only for the multi-column path after prepareResult() has populated result.
+ @transient private lazy val multiColEvaluator: MultiColumnInSubqueryEvaluator = {
+ val rows = result.map(_.asInstanceOf[InternalRow])
Review Comment:
**Nit (P3):** updateResult's normal paths already retain a runtime Array[InternalRow], so this map allocates and copies an extra n-element reference array before the index builders traverse it again. Please reuse result.asInstanceOf[Array[InternalRow]] on that path, with a conversion only for any supported Array[Any] fallback caller.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2682,421 @@ 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).
+ //
+ // The predicate 5 NOT IN (...) is a constant that references neither join input, so
+ // RewritePredicateSubquery returns the original Join unchanged and PlanSubqueries always
+ // constructs InSubqueryExec here regardless of any optimizer config.
+ 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))")
+
+ val query = sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0" +
+ " ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ // Verify the physical plan contains InSubqueryExec so the nullability fix is
+ // actually exercised and not silently bypassed by an optimizer path.
+ // Use AdaptiveSparkPlanHelper.find to traverse AQE wrapper nodes, and recurse
+ // into expression subtrees via Expression.exists to find the wrapped InSubqueryExec.
+ assert(
+ find(query.queryExecution.executedPlan) { p =>
+ p.expressions.exists(_.exists {
+ case _: InSubqueryExec => true
+ case _ => false
+ })
+ }.isDefined,
+ "Expected InSubqueryExec in the physical plan")
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ checkAnswer(query, Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30),
+ Row(1, null), Row(2, null), Row(3, null)))
+ }
+ }
+
+ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: RHS is fully non-null (stored in nonNullSet). LHS (NULL, 2) vs RHS (99, 2):
+ // second fields equal, first field null => UNKNOWN; exercises the nonNullSet scan with
+ // a nullable LHS. Fixture: lhs.a is nullable; rhs columns are NOT NULL (VALUES-derived
+ // to avoid Parquet dataSchema.asNullable widening that would defeat the RHS control).
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+
+ // Case B: RHS is fully non-null (goes into nonNullSet, not nullRows). LHS (NULL, 1) vs
+ // RHS (99, 2): first field null, but second field 1 != 2 is a definitive mismatch that
+ // makes the candidate FALSE. An implementation that stops at the first UNKNOWN instead
+ // of continuing to check later fields would return UNKNOWN here instead of FALSE, causing
+ // NOT IN to be UNKNOWN and the inner join to return no rows.
+ withTempView("lhs_nn", "rhs_nn") {
+ sql("CREATE TEMPORARY VIEW lhs_nn AS " +
+ "SELECT * FROM VALUES (CAST(NULL AS INT), 1) AS t(a, b)")
+ sql("CREATE TEMPORARY VIEW rhs_nn AS " +
+ "SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 1) NOT IN ((99, 2)): second field 1 != 2 => candidate FALSE => NOT IN TRUE.
+ // Inner join returns the single lhs row.
+ checkAnswer(
+ sql(
+ """SELECT lhs_nn.b FROM lhs_nn JOIN (SELECT 1 AS x)
+ |ON ((lhs_nn.a, lhs_nn.b) NOT IN (SELECT a, b FROM rhs_nn))
+ |""".stripMargin),
+ Seq(Row(1)))
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields " +
+ "for multi-column IN subqueries") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: InSubqueryExec.doGenCode registers Nondeterministic children " +
+ "for partition-level initialization") {
+ // A join-condition IN subquery with a nondeterministic LHS is rejected by CheckAnalysis,
+ // so this direct-construction test provides a focused unit check: it constructs
+ // InSubqueryExec directly (bypassing analysis) to verify that doGenCode correctly
+ // registers each Nondeterministic descendant of the LHS child for partition-level
+ // initialization. Without that registration, a Rand node's eval() would throw
+ // IllegalArgumentException (via require(initialized, ...)) because initialize() was
+ // never called.
+ //
+ // Two output columns force plan.output.length > 1, taking the multi-column fallback
+ // branch (with the explicit foreach loop), not the single-column InSet path where Rand
+ // would register itself through ordinary child codegen traversal.
+ //
+ // The test compiles the expression to bytecode via GeneratePredicate and executes it,
+ // so a missing initialize() call causes an actual failure rather than just a missing
+ // entry in partitionInitializationStatements.
+ val subplanOutput = spark
+ .range(0)
+ .select($"id".cast(IntegerType).as("a"), $"id".cast(IntegerType).as("b"))
+ .queryExecution.executedPlan.output
+ val subplan = SubqueryExec(
+ "test_subquery",
+ LocalTableScanExec(subplanOutput, Nil, None))
+ // LHS: struct of (Literal(1), Rand(42)>=0.0). Rand is always non-negative so the struct
+ // always equals the single RHS row (1, true), making IN = true.
+ val lhs = CreateNamedStruct(Seq(
+ Literal("a"), Literal(1),
+ Literal("b"), GreaterThanOrEqual(Rand(42L), Literal(0.0))))
+ val expr = InSubqueryExec(
+ lhs, subplan, ExprId(99),
+ isDynamicPruning = false,
+ result = Array(InternalRow(1, true)))
+ // Compile to bytecode, initialize (seeds the Rand RNG), and evaluate.
+ // If initialize() were skipped the Rand eval() would throw IllegalArgumentException.
+ val pred = GeneratePredicate.generate(expr, useSubexprElimination = false)
+ pred.initialize(0)
+ assert(pred.eval(InternalRow.empty) === true,
+ "Expected IN to be TRUE: (1, rand()>=0.0) IN ((1, true))")
+ }
+
+ test("SPARK-58481: MultiColumnInSubqueryEvaluator preserves shared Nondeterministic " +
+ "identity after Java serialization round-trip") {
+ // In doGenCode, ctx.references receives both the MultiColumnInSubqueryEvaluator (slot 0)
+ // and the bare Nondeterministic node (slot 1) for partition initialization. On the driver
+ // both slots point to the same Rand instance (evaluator.child embeds it). The risk is that
+ // Java serialization could write two independent copies, causing initialize() called on
+ // references[1] to seed a different instance than eval() uses inside the evaluator.
+ //
+ // Java's ObjectOutputStream tracks shared references within a single graph (via its
+ // internal handle table), so a single serialize/deserialize round-trip of the references
+ // array as one object produces back-references rather than copies. This test verifies
+ // that invariant explicitly using the same JavaSerializer Spark uses for task closures.
+ val rand = Rand(42L)
+ val lhs = CreateNamedStruct(Seq(
+ Literal("a"), Literal(1),
+ Literal("b"), GreaterThanOrEqual(rand, Literal(0.0))))
+ val evaluator = new MultiColumnInSubqueryEvaluator(
+ lhs,
+ Array(IntegerType, IntegerType),
+ Array(InternalRow(1, true)),
+ legacyNullInEmptyBehavior = false)
+ // Simulate what doGenCode does: put the evaluator and the bare Nondeterministic into
+ // the references array as separate slots (as WholeStageCodegenEvaluatorFactory would).
+ val references: Array[Any] = Array(evaluator, rand)
+ // Round-trip through JavaSerializer -- the same serializer Spark uses for task closures
+ // (SparkEnv always sets closureSerializer = new JavaSerializer(...)).
+ // Serializing the array directly is equivalent to serializing it as a field of
+ // WholeStageCodegenEvaluatorFactory: Java's ObjectOutputStream maintains a handle table
+ // across the entire stream, so shared-reference tracking is independent of nesting depth.
+ val ser = new JavaSerializer(new SparkConf()).newInstance()
+ val rt = ser.deserialize[Array[Any]](ser.serialize(references))
+ val rtEvaluator = rt(0).asInstanceOf[MultiColumnInSubqueryEvaluator]
+ val rtRand = rt(1).asInstanceOf[Rand]
+ // Find the Rand inside the deserialized evaluator's child expression tree.
+ var embeddedRand: Rand = null
+ rtEvaluator.child.foreach {
+ case n: Rand => embeddedRand = n
+ case _ =>
+ }
+ assert(embeddedRand ne null,
+ "Expected a Rand node inside the evaluator's child after round-trip")
+ // The critical identity check: the Rand node that would receive initialize(partitionIndex)
+ // must be the same instance eval() uses inside the evaluator -- if Java serialization wrote
+ // two independent copies, initialize() would seed a different instance than eval() uses.
+ assert(embeddedRand eq rtRand,
+ "Rand inside evaluator and separately-registered Rand must be the same instance " +
+ "after serialization round-trip")
+ }
+
+ test("SPARK-58481: multi-column IN with empty subquery short-circuits before evaluating LHS") {
+ // Under ANSI semantics, IN (empty subquery) is always FALSE regardless of the LHS value.
+ // evalMultiColumn must return false before calling child.eval so that a LHS expression
Review Comment:
**Nit (P3):** evalMultiColumn no longer exists after the evaluator extraction. Please update both occurrences in this test area to MultiColumnInSubqueryEvaluator.eval, or simply call it the multi-column evaluator.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3868490742
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,177 @@ 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
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (multiColNullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ 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 candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
+ if (candidateField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (multiColNullRows.isEmpty && multiColNonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ 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 = inputFields(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 non-null rows: a null LHS comparison is UNKNOWN unless a non-null field differs.
+ val nonNullIter = multiColNonNullSet.iterator
+ while (nonNullIter.hasNext && !hasUnknown) {
+ val candidate = nonNullIter.next()
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ if (inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputField, candidate.get(fieldIdx, fieldTypes(fieldIdx))) != 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ }
+ if (hasUnknown) null else false
+ }
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ evalMultiColumn(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path via eval().
+ // Register any Nondeterministic descendants (e.g. rand() in the LHS) for partition-level
+ // initialization, mirroring CodegenFallback's protocol.
+ val resultIdx = ctx.references.length
+ ctx.references += this
+ child.foreach {
+ case n: expressions.Nondeterministic =>
Review Comment:
**Non-blocking:**
Add a codegen-enabled multi-column `IN` regression whose LHS contains a nondeterministic expression such as `rand()`. This hand-written fallback owns partition initialization, but none of the added tests reaches this branch, so a future reference-index or initialization regression could recreate the pre-initialize runtime failure unnoticed.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +129,32 @@ 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.
Review Comment:
**Nit:**
A nullable LHS field makes UNKNOWN possible, not inevitable: `(NULL, 2)` compared with `(99, 3)` is FALSE because the second field differs.
```suggestion
// is preserved in both modes so a nullable LHS field can contribute UNKNOWN when no field differs.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,177 @@ 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
Review Comment:
**Non-blocking:**
Bind `multiColNullRows` and `multiColNonNullSet` to locals here, then use those aliases in both branches. Because these fields are lazy vals, each current loop condition and candidate lookup re-enters an accessor for every candidate of every input row.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5352071816
Thanks for the detailed review @cloud-fan , have addressed all three points:
- Replaced `equals` with `TypeUtils.getInterpretedOrdering(dt).compare(...)` for correct Catalyst equality
- Hoisted output schema and per-field orderings/types out of the candidate loop
- Capitalized "Result" in the comment
- Added binary-column regression test
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3815458758
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +176,76 @@ case class InSubqueryExec(
}
}
+ // Three-valued IN semantics for multi-column subqueries (SPARK-58481).
+ // result rows are InternalRow objects; InSet uses TreeSet ordering which treats null fields as
Review Comment:
**Nit:**
Please capitalize `Result` at the start of this sentence.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +176,76 @@ case class InSubqueryExec(
}
}
+ // Three-valued IN semantics for multi-column subqueries (SPARK-58481).
+ // result rows are InternalRow objects; InSet uses TreeSet ordering which treats null fields as
+ // non-equal and therefore cannot distinguish a definitively-false candidate (a non-null field
+ // differs) from an indeterminate one (all non-null fields match but some fields are null).
+ // We replicate In.eval's per-candidate logic: TRUE if any candidate matches exactly,
+ // UNKNOWN if no TRUE and at least one candidate is indeterminate, FALSE otherwise.
+ private def evalMultiColumn(inputRow: InternalRow): Any = {
+ val value = child.eval(inputRow)
+ if (value == null) return null
+ val inputStruct = value.asInstanceOf[InternalRow]
+ val numFields = plan.output.length
+ var hasUnknown = false
+ var i = 0
+ while (i < result.length) {
+ val candidate = result(i).asInstanceOf[InternalRow]
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputStruct.get(fieldIdx, plan.output(fieldIdx).dataType)
+ val candidateField = candidate.get(fieldIdx, plan.output(fieldIdx).dataType)
+ if (candidateField == null || inputField == null) {
+ candidateIsUnknown = true
+ } else if (!inputField.equals(candidateField)) {
Review Comment:
**Blocking:**
Use Catalyst's interpreted ordering for each field here. `Object.equals` is not SQL equality for supported values such as `BinaryType` byte arrays, so equal binary fields from separately materialized rows compare unequal and a multi-column `IN` can incorrectly return FALSE. Please use the field data type's Catalyst ordering and add a binary-column regression case.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +176,76 @@ case class InSubqueryExec(
}
}
+ // Three-valued IN semantics for multi-column subqueries (SPARK-58481).
+ // result rows are InternalRow objects; InSet uses TreeSet ordering which treats null fields as
+ // non-equal and therefore cannot distinguish a definitively-false candidate (a non-null field
+ // differs) from an indeterminate one (all non-null fields match but some fields are null).
+ // We replicate In.eval's per-candidate logic: TRUE if any candidate matches exactly,
+ // UNKNOWN if no TRUE and at least one candidate is indeterminate, FALSE otherwise.
+ private def evalMultiColumn(inputRow: InternalRow): Any = {
+ val value = child.eval(inputRow)
+ if (value == null) return null
+ val inputStruct = value.asInstanceOf[InternalRow]
+ val numFields = plan.output.length
+ var hasUnknown = false
+ var i = 0
+ while (i < result.length) {
+ val candidate = result(i).asInstanceOf[InternalRow]
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputStruct.get(fieldIdx, plan.output(fieldIdx).dataType)
Review Comment:
**Non-blocking:**
Capture `plan.output` before the candidate loop and the current field's data type once per field. Both values are invariant, but these lookups currently run twice for every field of every result row.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5359886726
Thanks for the review @cloud-fan , have addressed all three points:
- Derive LHS nullability from `valExprs` for `CreateNamedStruct` so nullable LHS fields are correctly propagated through generated NOT IN code
- Split result rows into a TreeSet of fully non-null rows (O(log n) indexed lookup) and an array of null-containing rows (linear scan only when needed)
- Fixed comment to 5+5=10 rows
- Added nullable-LHS/non-nullable-RHS codegen regression test
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5692319344
merged to master/4.x
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3849443918
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +186,173 @@ 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). In
+ // that case we scan all result rows linearly.
Review Comment:
**Nit:**
Please describe this as a linear scan that stops once UNKNOWN is established. Both loops short-circuit on `!hasUnknown`, so 'all result rows' is also inaccurate in the repeated comment at `subquery.scala:265`.
##########
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
+ // (null in any column position produces UNKNOWN on a miss) or when any LHS field is nullable.
Review Comment:
**Nit:**
Describe this as a possible outcome rather than an unconditional one. `(1,1)` versus `(99,NULL)` is definitively FALSE because the first field differs, even though another comparison is UNKNOWN.
```suggestion
// (a nullable RHS field can produce UNKNOWN on a miss) or when any LHS field is nullable.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +186,173 @@ 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). 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
+ // Materialize LHS fields once before the candidate scans. For types like BinaryType,
+ // InternalRow.get copies the payload on every call; caching avoids one copy per
Review Comment:
**Nit:**
Please remove the BinaryType copy-avoidance rationale here and at `subquery.scala:267`. The LHS is produced by `CreateNamedStruct` as a `GenericInternalRow`, and `BaseGenericInternalRow.get` returns the stored field reference rather than copying a binary payload.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,164 @@ 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")
Review Comment:
**Non-blocking:**
Use non-nullable `VALUES`-derived inputs for the LHS and the RHS first column, leaving only a later RHS column nullable. File-source analysis applies `dataSchema.asNullable`, so both this LHS and the RHS head analyze as nullable despite the DDL; the test therefore still passes with the old `plan.output.head.nullable` check and does not isolate the behavior named by the test.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,164 @@ 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).
Review Comment:
**Nit:**
The relation labels are reversed: `Row(null, 10)` retains the unmatched t1 row and pads t0, while `Row(1, null)` retains t0 and pads t1.
```suggestion
// Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5413398465
Thanks for the review @cloud-fan , have addressed both:
- Gated the `CreateNamedStruct/valExprs` nullability path on plan.output.length > 1, so a user-written single struct-valued IN correctly falls through to `child.nullable` instead of picking up field-level nullability meant only for the generated multi-column carrier
- Added early returns before `inputFields` materialization when there are no candidates to scan (empty `multiColNullRows` in the non-null-LHS path; both collections empty in the nullable-LHS path)
- Switched rhs to a VALUES-derived temp view in the nullable-LHS regression so it actually stays non-nullable instead of being widened by Parquet's dataSchema.asNullable
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3920856006
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,188 @@ 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 using the struct-level Catalyst ordering so that duplicates are
+ // deduplicated. Fully non-null rows go into multiColNonNullSet (TreeSet) for O(log n)
+ // membership tests. Null-containing rows are deduplicated via a temporary TreeSet and then
+ // stored as multiColNullRows (Array) for linear scanning; each distinct null-containing row
+ // is thus scanned at most once per outer row regardless of RHS duplicate multiplicity.
+ // See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
+ 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
+ // Cache lazy accessors in locals so the loop bodies do not re-enter them on every iteration.
+ val nullRows = multiColNullRows
+ val nonNullSet = multiColNonNullSet
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (nonNullSet.contains(inputStruct)) return true
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (nullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(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(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (nullRows.isEmpty && nonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ var hasUnknown = false
+ // Scan null-containing result rows first.
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(i)
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(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 non-null rows: a null LHS comparison is UNKNOWN unless a non-null field differs.
+ val nonNullIter = nonNullSet.iterator
+ while (nonNullIter.hasNext && !hasUnknown) {
+ val candidate = nonNullIter.next()
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ if (inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputField, candidate.get(fieldIdx, fieldTypes(fieldIdx))) != 0) {
+ candidateIsFalse = true
Review Comment:
**Non-blocking (P2):** This fully non-null RHS loop is distinct from the nullRows loop, but the current late-mismatch regression exercises only nullRows. Please extend `SPARK-58481: multi-column NOT IN with nullable LHS and non-nullable RHS is nullable` with a fully non-null RHS such as `(99, 2)` and a nullable LHS `(NULL, 1)`, then assert through the forced InSubqueryExec join path that NOT IN is TRUE (the inner join returns exactly `Row(1)`). This catches a regression that stops at the first UNKNOWN instead of allowing the later mismatch to make the candidate FALSE.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,361 @@ 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).
+ //
+ // Disable the optimizer rewrite so the query exercises InSubqueryExec.nullable directly;
+ // with the default enabled the IN is folded away before PlanSubqueries constructs
+ // InSubqueryExec, making the nullability fix unreachable by this test.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ val query = sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0" +
+ " ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ // Verify the physical plan contains InSubqueryExec so the nullability fix is
+ // actually exercised and not silently bypassed by an optimizer path.
+ // Use AdaptiveSparkPlanHelper.find to traverse AQE wrapper nodes, and recurse
+ // into expression subtrees via Expression.exists to find the wrapped InSubqueryExec.
+ assert(
+ find(query.queryExecution.executedPlan) { p =>
+ p.expressions.exists(_.exists {
+ case _: InSubqueryExec => true
+ case _ => false
+ })
+ }.isDefined,
+ "Expected InSubqueryExec in the physical plan")
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ checkAnswer(query, Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30),
+ Row(1, null), Row(2, null), Row(3, null)))
+ }
+ }
+ }
+
+ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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 (VALUES-derived to avoid Parquet
+ // dataSchema.asNullable widening that would defeat the RHS non-nullability control).
+ // (NULL, 2) vs (99, 2): second fields equal (2=2), first field is null => UNKNOWN.
+ // (1, 1) vs (99, 2): first field 1!=99 => FALSE => NOT IN = TRUE => matches all rhs rows.
+ // FULL OUTER JOIN: (NULL,2) gets null-padded (UNKNOWN condition); (1,1) joins with (99,2);
+ // since (1,1) matched rhs(99,2), rhs(99,2) is not null-padded.
+ // Pre-fix: (NULL,2) NOT IN is wrongly TRUE (null suppressed) => emits (null,99); no
+ // null-padded rows. Post-fix: UNKNOWN propagated => emits (null,null) for (NULL,2).
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs(99,2) is matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields " +
+ "for multi-column IN subqueries") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: InSubqueryExec.doGenCode registers Nondeterministic children " +
+ "for partition-level initialization") {
+ // A nondeterministic expression in the LHS of a multi-column IN subquery is unreachable
Review Comment:
**Non-blocking (P2):** This multi-column fallback is reachable from SQL in a projection: RewritePredicateSubquery does not rewrite Project expressions, CheckAnalysis permits nondeterministic Project expressions, and PlanSubqueries then creates InSubqueryExec. Please add an end-to-end projection regression such as `SELECT (id, rand(42)) IN (SELECT id, score FROM rhs) FROM lhs` and correct the comment. That verifies the analysis/planning/codegen path in addition to the focused direct-construction check.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,188 @@ 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 using the struct-level Catalyst ordering so that duplicates are
+ // deduplicated. Fully non-null rows go into multiColNonNullSet (TreeSet) for O(log n)
+ // membership tests. Null-containing rows are deduplicated via a temporary TreeSet and then
+ // stored as multiColNullRows (Array) for linear scanning; each distinct null-containing row
+ // is thus scanned at most once per outer row regardless of RHS duplicate multiplicity.
+ // See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
+ 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
+ // Cache lazy accessors in locals so the loop bodies do not re-enter them on every iteration.
+ val nullRows = multiColNullRows
+ val nonNullSet = multiColNonNullSet
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (nonNullSet.contains(inputStruct)) return true
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (nullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(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(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (nullRows.isEmpty && nonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ var hasUnknown = false
+ // Scan null-containing result rows first.
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(i)
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ val candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
Review Comment:
**Nit (P3):** When `inputField` is null, this comparison is already UNKNOWN regardless of the RHS value, so `candidate.get` is unnecessary and repeats for every scanned null-containing row. Please check `inputField == null` first and retrieve `candidateField` only in the non-null branch; this preserves the truth table while removing cardinality-scaled row accesses.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,361 @@ 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).
+ //
+ // Disable the optimizer rewrite so the query exercises InSubqueryExec.nullable directly;
+ // with the default enabled the IN is folded away before PlanSubqueries constructs
Review Comment:
**Nit (P3):** This rationale is not accurate for the constant predicate used below. `5 NOT IN (...)` references neither join input, and RewritePredicateSubquery returns the original Join when both reference groups are empty, so PlanSubqueries still constructs InSubqueryExec with the default setting. Please remove the unnecessary override and update the comment to describe the actual reachability check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5692303443
Thanks, I re-reviewed the latest changes. The value-only evaluator boundary, bound LHS codegen, serialization behavior, and new regressions look good to me; I found no new issues.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5668488066","thread_id":"conversation:5668488066","verdict_sha256":"b57d1c28db544159b1abc8eab7a9d9d0ef2865e2b67c52e0ef1fdd0eee6c3e23"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3869684827
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +195,180 @@ 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)
Review Comment:
**Blocking:**
Preserve the current empty-set short circuit before evaluating the LHS. Under ANSI behavior, `InSet.eval` returns `false` for an empty candidate set before calling `child.eval`, but this path evaluates `child` first; an otherwise irrelevant `(1 / zero_col, other_col) IN (empty subquery)` now throws instead of returning `false`. Please mirror `InSet`'s config-gated empty-result check before this call and add a regression that forces this physical path.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3926343023
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,193 @@ 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 using the struct-level Catalyst ordering so that duplicates are
+ // deduplicated. Fully non-null rows go into multiColNonNullSet (TreeSet) for O(log n)
+ // membership tests. Null-containing rows are deduplicated via a temporary TreeSet and then
+ // stored as multiColNullRows (Array) for linear scanning; each distinct null-containing row
+ // is thus scanned at most once per outer row regardless of RHS duplicate multiplicity.
+ // See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
+ 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
+ // Cache lazy accessors in locals so the loop bodies do not re-enter them on every iteration.
+ val nullRows = multiColNullRows
+ val nonNullSet = multiColNonNullSet
+
+ if (!inputStruct.anyNull) {
Review Comment:
**Nit (P3):** Legacy mode must evaluate `child`, but once `value != null` and `result.isEmpty`, the result is definitely false. This path currently initializes both empty TreeSets and scans `inputStruct.anyNull` for every outer row. Please add `if (result.isEmpty) return false` immediately after the `value == null` guard; that preserves legacy evaluation semantics while bypassing the redundant index and per-field work.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,193 @@ 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 using the struct-level Catalyst ordering so that duplicates are
+ // deduplicated. Fully non-null rows go into multiColNonNullSet (TreeSet) for O(log n)
+ // membership tests. Null-containing rows are deduplicated via a temporary TreeSet and then
+ // stored as multiColNullRows (Array) for linear scanning; each distinct null-containing row
+ // is thus scanned at most once per outer row regardless of RHS duplicate multiplicity.
+ // See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
+ 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
+ // Cache lazy accessors in locals so the loop bodies do not re-enter them on every iteration.
+ val nullRows = multiColNullRows
+ val nonNullSet = multiColNonNullSet
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (nonNullSet.contains(inputStruct)) return true
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (nullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(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(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (nullRows.isEmpty && nonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ var hasUnknown = false
+ // Scan null-containing result rows first.
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(i)
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ if (inputField == null) {
+ // LHS field is null: UNKNOWN regardless of the RHS value; skip candidate.get.
+ candidateIsUnknown = true
+ } else {
+ val candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
+ if (candidateField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(inputField, candidateField) != 0) {
+ candidateIsFalse = true
+ }
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ i += 1
+ }
+ // Scan non-null rows: a null LHS comparison is UNKNOWN unless a non-null field differs.
+ val nonNullIter = nonNullSet.iterator
+ while (nonNullIter.hasNext && !hasUnknown) {
+ val candidate = nonNullIter.next()
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ if (inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputField, candidate.get(fieldIdx, fieldTypes(fieldIdx))) != 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ }
+ if (hasUnknown) null else false
+ }
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ evalMultiColumn(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path via eval().
+ // Register any Nondeterministic descendants (e.g. rand() in the LHS) for partition-level
+ // initialization, mirroring CodegenFallback's protocol.
+ val resultIdx = ctx.references.length
+ ctx.references += this
Review Comment:
**Non-blocking (P2):** `ctx.references += this` puts the whole `InSubqueryExec` into `WholeStageCodegenEvaluatorFactory.references`. Although `result` is transient, `plan` is not, so task serialization retains the `BaseSubqueryExec` and its physical subtree after the compact result has already been broadcast. A large file-backed subquery can therefore bloat task closures or exceed task-size limits. Please register a lightweight evaluator containing only the LHS expression, output types/orderings, and broadcast result, so the driver plan does not cross the executor boundary.
**Recommended change:** Reference a lightweight serializable multi-column evaluator from generated code instead of the full InSubqueryExec.
**Why this works:** Separate the child, collected result, schema/orderings, and multi-column evaluation logic into executor state that excludes BaseSubqueryExec; register that object while preserving explicit Nondeterministic initialization.
**Scope:** SQL core subquery execution and focused codegen or serialization coverage.
**Compatibility:** Keep SQL three-valued results, legacy empty-RHS behavior, the broadcast lifecycle, Dynamic Partition Pruning behavior, and nondeterministic initialization unchanged.
**Risks:** Separating evaluator state could omit required output ordering information. Moving the child expression could accidentally bypass partition-level initialization for nondeterministic descendants.
**Constraints:** Do not serialize BaseSubqueryExec or its driver SparkPlan into task references. Keep the single-column InSet path unchanged.
**Success:** Whole-stage evaluator references contain only compact subquery value/evaluator state and no BaseSubqueryExec, while multi-column and nondeterministic codegen behavior remains unchanged.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,395 @@ 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).
+ //
+ // The predicate 5 NOT IN (...) is a constant that references neither join input, so
+ // RewritePredicateSubquery returns the original Join unchanged and PlanSubqueries always
+ // constructs InSubqueryExec here regardless of any optimizer config.
+ 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))")
+
+ val query = sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0" +
+ " ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ // Verify the physical plan contains InSubqueryExec so the nullability fix is
+ // actually exercised and not silently bypassed by an optimizer path.
+ // Use AdaptiveSparkPlanHelper.find to traverse AQE wrapper nodes, and recurse
+ // into expression subtrees via Expression.exists to find the wrapped InSubqueryExec.
+ assert(
+ find(query.queryExecution.executedPlan) { p =>
+ p.expressions.exists(_.exists {
+ case _: InSubqueryExec => true
+ case _ => false
+ })
+ }.isDefined,
+ "Expected InSubqueryExec in the physical plan")
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ checkAnswer(query, Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30),
+ Row(1, null), Row(2, null), Row(3, null)))
+ }
+ }
+
+ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: RHS has a null-containing row. LHS (NULL, 2) vs RHS (99, 2): second fields
+ // equal, first field null => UNKNOWN; exercises the nullRows scan in evalMultiColumn.
+ // Fixture: lhs.a is nullable; rhs columns are NOT NULL (VALUES-derived to avoid Parquet
+ // dataSchema.asNullable widening that would defeat the RHS non-nullability control).
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+
+ // Case B: RHS is fully non-null (goes into nonNullSet, not nullRows). LHS (NULL, 1) vs
+ // RHS (99, 2): first field null, but second field 1 != 2 is a definitive mismatch that
+ // makes the candidate FALSE. An implementation that stops at the first UNKNOWN instead
+ // of continuing to check later fields would return UNKNOWN here instead of FALSE, causing
+ // NOT IN to be UNKNOWN and the inner join to return no rows.
+ withTempView("lhs_nn", "rhs_nn") {
+ sql("CREATE TEMPORARY VIEW lhs_nn AS " +
+ "SELECT * FROM VALUES (CAST(NULL AS INT), 1) AS t(a, b)")
+ sql("CREATE TEMPORARY VIEW rhs_nn AS " +
+ "SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 1) NOT IN ((99, 2)): second field 1 != 2 => candidate FALSE => NOT IN TRUE.
+ // Inner join returns the single lhs row.
+ checkAnswer(
+ sql(
+ """SELECT lhs_nn.b FROM lhs_nn JOIN (SELECT 1 AS x)
+ |ON ((lhs_nn.a, lhs_nn.b) NOT IN (SELECT a, b FROM rhs_nn))
+ |""".stripMargin),
+ Seq(Row(1)))
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields " +
+ "for multi-column IN subqueries") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: nondeterministic LHS in a Project IN subquery initializes via codegen") {
+ // RewritePredicateSubquery does not rewrite Project expressions, and CheckAnalysis permits
Review Comment:
**Non-blocking (P2):** This premise is reversed: `Project` is a `UnaryNode`, and `RewritePredicateSubquery` sends any matching unary node through `handleUnaryNode`, which replaces the IN expression with an `ExistenceJoin`. This query can pass without constructing `InSubqueryExec`, so it does not verify `doGenCode` or nondeterministic initialization. Please remove this projection test and its end-to-end coverage claim; the focused direct-construction `GeneratePredicate` test below is the regression that actually executes this codegen branch.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,395 @@ 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).
+ //
+ // The predicate 5 NOT IN (...) is a constant that references neither join input, so
+ // RewritePredicateSubquery returns the original Join unchanged and PlanSubqueries always
+ // constructs InSubqueryExec here regardless of any optimizer config.
+ 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))")
+
+ val query = sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0" +
+ " ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ // Verify the physical plan contains InSubqueryExec so the nullability fix is
+ // actually exercised and not silently bypassed by an optimizer path.
+ // Use AdaptiveSparkPlanHelper.find to traverse AQE wrapper nodes, and recurse
+ // into expression subtrees via Expression.exists to find the wrapped InSubqueryExec.
+ assert(
+ find(query.queryExecution.executedPlan) { p =>
+ p.expressions.exists(_.exists {
+ case _: InSubqueryExec => true
+ case _ => false
+ })
+ }.isDefined,
+ "Expected InSubqueryExec in the physical plan")
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ checkAnswer(query, Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30),
+ Row(1, null), Row(2, null), Row(3, null)))
+ }
+ }
+
+ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: RHS has a null-containing row. LHS (NULL, 2) vs RHS (99, 2): second fields
Review Comment:
**Nit (P3):** The RHS is `(99, 2)`, so it is fully non-null and is stored in `multiColNonNullSet`. The UNKNOWN result comes from scanning `nonNullSet` with a nullable LHS, not from `nullRows`. Please update these two sentences to say `fully non-null RHS` and `nonNullSet scan`; a genuinely null-containing case can carry the `nullRows` coverage claim.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3812018372
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,52 @@ 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") {
+ // The logical InSubquery.nullable checks ALL query outputs, not just the first.
+ // The physical InSubqueryExec must do the same: for multi-column IN, results are
+ // stored as InternalRow objects and InSet.hasNull only detects top-level null elements,
+ // not null fields inside a row. If only plan.output.head.nullable were checked, a
+ // subquery with a non-nullable first column but a nullable second column would be
+ // incorrectly declared non-nullable, causing codegen to skip the null-result path and
+ // evaluate NOT IN as !false = true instead of UNKNOWN.
+ // Here: lhs has (a NOT NULL, b NOT NULL); rhs has (a NOT NULL, b nullable).
+ // (1,1) NOT IN ((99,99),(NULL_a,NULL_b)): no exact match and the set contains a row
+ // with null fields => UNKNOWN => no row satisfies the ON condition
+ // => FULL OUTER JOIN emits null-padded rows from both sides (2 + 2 = 4 rows).
+ 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")
+ sql("INSERT INTO rhs VALUES (99, 99), (CAST(NULL AS INT), CAST(NULL AS INT))")
+
+ val result = 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)
+ assert(result.count() === 4,
Review Comment:
**Non-blocking:**
Please assert the actual rows here. The current broken runtime path makes this join condition TRUE and returns a 2 x 2 cross product, which also has count 4; the intended UNKNOWN path returns four null-padded rows. A row-level assertion will distinguish those outputs and fail until nested-NULL propagation is fixed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5515124425
Thanks for the review @cloud-fan , have addressed all six:
- Forced the primary nullability regression through `InSubqueryExec` by disabling the join-condition rewrite and added an assertion on the executed plan (via `Expression.exists` recursing into the expression tree) confirming `InSubqueryExec` is actually present before checking results
- Added a legacy empty-RHS regression: forces the physical path with ANSI + `LEGACY_NULL_IN_EMPTY_LIST_BEHAVIOR=true`, and asserts a `DIVIDE_BY_ZERO` LHS actually throws proving the legacy branch evaluates the LHS rather than short-circuiting
- Added a null-vs-null regression distinguishing (NULL,1) IN ((NULL,1)) (UNKNOWN) from (NULL,1) IN ((NULL,2)) (FALSE via a later definitive mismatch)
- Fixed the `multiColNullRows` comment to accurately describe it as an Array (deduplicated via a temporary TreeSet at construction), not a TreeSet
- Fixed the subject-verb agreement nit ("columns force")
- Renamed the multi-column legacy test to complete the sentence
10/10 tests pass. Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5381704607
Thanks for the detailed review @cloud-fan , have addressed all three:
- Nondeterministic descendants (eg: rand() in LHS) now registered for partition-level init via a fresh index per append, mirroring CodegenFallback's exact pattern
- lhsNullable now computed identically in both branches; legacy flag only gates rhsNullable
- Non-null row index now built with TreeSet.newBuilder instead of foldLeft
Added a legacy multi-column regression covering the nullable-LHS-field case the old formula got wrong.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3807100136
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +125,10 @@ case class InSubqueryExec(
@transient private lazy val inSet = InSet(child, result.toSet)
- override def nullable: Boolean = child.nullable
+ // x IN (subquery) evaluates to UNKNOWN when no match is found and the subquery result
+ // contains NULL, so nullable must account for whether the subquery output can produce NULLs,
+ // not just whether the child expression is nullable. See SPARK-58481.
+ override def nullable: Boolean = child.nullable || plan.output.head.nullable
Review Comment:
**Blocking:**
Please preserve UNKNOWN semantics for every subquery output here. A multi-column `IN` can have a non-nullable first output and a nullable later output, so checking only `head` still declares this expression non-nullable; those results are struct rows, and `InSet.hasNull` also misses nested NULL fields. Please propagate nested-field UNKNOWN at runtime, mirror the logical all-output nullability contract (including the legacy config), and add a focused multi-column regression.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3910711535
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,187 @@ 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 two sorted sets, both using the struct-level Catalyst ordering
+ // so that duplicate rows are deduplicated. Fully non-null rows go into multiColNonNullSet
+ // for O(log n) membership tests; rows with at least one null field go into multiColNullRows
+ // (also a TreeSet, not an Array) so each distinct null-containing row is scanned at most
+ // once per outer row regardless of RHS duplicate multiplicity. See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Mirror InSet.eval: if the candidate set is empty, return false before evaluating the LHS.
+ // Under legacy behavior (ANSI off by default), the SQL standard says IN (empty set) is
Review Comment:
**Non-blocking (P2):** The compatibility description is reversed here. `spark.sql.legacy.nullInEmptyListBehavior` preserves the old NULL result; current behavior returns FALSE for an empty set. Also, only the current branch skips `child.eval` - with the legacy flag enabled, the LHS is still evaluated. Please describe those two branches explicitly so this comment matches the guard below.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2682,275 @@ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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 (VALUES-derived to avoid Parquet
+ // dataSchema.asNullable widening that would defeat the RHS non-nullability control).
+ // (NULL, 2) vs (99, 2): second fields equal (2=2), first field is null => UNKNOWN.
+ // (1, 1) vs (99, 2): first field 1!=99 => FALSE => NOT IN = TRUE => matches all rhs rows.
+ // FULL OUTER JOIN: (NULL,2) gets null-padded (UNKNOWN condition); (1,1) joins with (99,2);
+ // since (1,1) matched rhs(99,2), rhs(99,2) is not null-padded.
+ // Pre-fix: (NULL,2) NOT IN is wrongly TRUE (null suppressed) => emits (null,99); no
+ // null-padded rows. Post-fix: UNKNOWN propagated => emits (null,null) for (NULL,2).
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs(99,2) is matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields multi-column") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: InSubqueryExec.doGenCode registers Nondeterministic children " +
+ "for partition-level initialization") {
+ // CheckAnalysis rejects nondeterministic expressions in any join or filter IN subquery
Review Comment:
**Non-blocking (P2):** This rationale names both the wrong phase and the wrong failure type. CheckAnalysis exempts Filter; filter IN subqueries disappear later in RewritePredicateSubquery, while nondeterministic joins are rejected. Also, `Nondeterministic.eval` uses `require(initialized)`, so a pre-initialize call raises IllegalArgumentException, not IllegalStateException. Please update or simplify the comment.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5692448819
Thank you @cloud-fan for the detailed review!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3966363951
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,79 @@ case class InSubqueryExec(
}
}
+ // Memoized lightweight evaluator instance. @transient so it is rebuilt (from the
+ // serializable fields) when InSubqueryExec is deserialized in an executor; constructed
+ // only for the multi-column path after prepareResult() has populated result.
+ // The child expression is transformed to replace every prepared ScalarSubquery with its
+ // already-evaluated Literal, so that no nested BaseSubqueryExec crosses the serialization
+ // boundary with generated task closures. See SPARK-58481.
+ @transient private[sql] lazy val multiColEvaluator: MultiColumnInSubqueryEvaluator = {
+ // On the SQL execution path, updateResult stores rows.asInstanceOf[Array[Any]] where rows is
+ // Array[InternalRow], so result is always Array[InternalRow] at runtime. Cast directly to
+ // avoid allocating an extra reference array.
+ val rows = result.asInstanceOf[Array[InternalRow]]
+ val fTypes = plan.output.map(_.dataType).toArray
+ // Replace each prepared ScalarSubquery in the LHS with its literal result so the captured
+ // child expression carries no BaseSubqueryExec references into executor task closures.
+ // Ordering is safe: SparkPlan.executeQuery calls waitForSubqueries() -- which calls
+ // updateResult() on every ExecSubqueryExpression found in this plan's expressions, including
+ // nested ScalarSubquery nodes inside child -- before doExecute()/doGenCode() runs and
+ // accesses this lazy val. So every s.toLiteral call below finds a fully-evaluated result.
+ val literalizedChild = child.transformUp {
+ case s: ScalarSubquery => s.toLiteral
+ }
+ new MultiColumnInSubqueryEvaluator(literalizedChild, fTypes, rows, legacyNullInEmptyBehavior)
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ multiColEvaluator.eval(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path via the lightweight evaluator.
+ // Register the evaluator (not the full InSubqueryExec) so that task closures do not
+ // carry plan (BaseSubqueryExec) or any other driver-side SparkPlan state.
+ val evalIdx = ctx.references.length
+ ctx.references += multiColEvaluator
+ // Register any Nondeterministic descendants for partition-level initialization,
+ // mirroring CodegenFallback's protocol. Walk the evaluator's literalized child so that
+ // initialize() is called on the same expression instances eval() uses -- if we walked
+ // the original child instead, a ScalarSubquery that was replaced by a Literal would
+ // leave its Nondeterministic descendants unregistered (harmless here, since ScalarSubquery
+ // results are constants, but walking the same child preserves the identity guarantee).
+ multiColEvaluator.child.foreach {
+ case n: expressions.Nondeterministic =>
+ val idx = ctx.references.length
+ ctx.references += n
+ ctx.addPartitionInitializationStatement(
+ s"((${classOf[expressions.Nondeterministic].getName}) references[$idx])" +
+ s".initialize(partitionIndex);")
+ case _ =>
+ }
+ val evalClass = classOf[MultiColumnInSubqueryEvaluator].getName
+ val tmp = ctx.freshName("inSubqueryTmp")
+ ev.copy(code =
+ code"""
+ Object $tmp =
+ (($evalClass) references[$evalIdx]).eval(${ctx.INPUT_ROW});
Review Comment:
**Blocking (P1):** This fallback interprets the bound child against `ctx.INPUT_ROW`, but residual join codegen binds the condition to streamed-plus-build `ctx.currentVars` while `INPUT_ROW` is only the build-side row. A multi-column IN/NOT IN whose LHS references both sides can therefore read the wrong ordinal or fail the task. Please generate the LHS through the active codegen inputs and pass its evaluated value to the lightweight evaluator, with a codegen-enabled residual-join regression that references both sides.
**Recommended change:** Move LHS evaluation back across the InSubqueryExec-to-evaluator boundary and keep only evaluated values and membership state in the task reference.
**Why this works:** Generate the multi-column LHS through Spark's normal expression codegen while the operator's currentVars are active, then pass only that evaluated value and the prepared membership state to a plan-free evaluator instead of letting the evaluator interpret and serialize the child expression.
**Scope:** InSubqueryExec's multi-column codegen branch, the MultiColumnInSubqueryEvaluator interface, and focused residual-join and nested-IN regressions in SubquerySuite.
**Compatibility:** Preserve the single-column InSet path, interpreted multi-column truth table, legacy empty-subquery behavior, and exactly-once initialization and evaluation of nondeterministic children.
**Risks:** The generated value representation must preserve struct-level and field-level nulls. Moving child evaluation must not duplicate nondeterministic evaluation or bypass partition initialization.
**Constraints:** Generated child evaluation must consume the bound ctx.currentVars layout for its operator. The evaluator stored in task references must not retain any BaseSubqueryExec or other driver-side SparkPlan. Nullable structs and fields must retain the existing three-valued IN and NOT IN semantics.
**Success:** A code-generated residual join predicate can read columns from both join sides without wrong ordinals or task failure, and a nested IN expression leaves no BaseSubqueryExec reachable from generated task references.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,79 @@ case class InSubqueryExec(
}
}
+ // Memoized lightweight evaluator instance. @transient so it is rebuilt (from the
+ // serializable fields) when InSubqueryExec is deserialized in an executor; constructed
+ // only for the multi-column path after prepareResult() has populated result.
+ // The child expression is transformed to replace every prepared ScalarSubquery with its
+ // already-evaluated Literal, so that no nested BaseSubqueryExec crosses the serialization
+ // boundary with generated task closures. See SPARK-58481.
+ @transient private[sql] lazy val multiColEvaluator: MultiColumnInSubqueryEvaluator = {
+ // On the SQL execution path, updateResult stores rows.asInstanceOf[Array[Any]] where rows is
+ // Array[InternalRow], so result is always Array[InternalRow] at runtime. Cast directly to
+ // avoid allocating an extra reference array.
+ val rows = result.asInstanceOf[Array[InternalRow]]
+ val fTypes = plan.output.map(_.dataType).toArray
+ // Replace each prepared ScalarSubquery in the LHS with its literal result so the captured
+ // child expression carries no BaseSubqueryExec references into executor task closures.
+ // Ordering is safe: SparkPlan.executeQuery calls waitForSubqueries() -- which calls
+ // updateResult() on every ExecSubqueryExpression found in this plan's expressions, including
+ // nested ScalarSubquery nodes inside child -- before doExecute()/doGenCode() runs and
+ // accesses this lazy val. So every s.toLiteral call below finds a fully-evaluated result.
+ val literalizedChild = child.transformUp {
Review Comment:
**Non-blocking (P2):** This transform removes only `ScalarSubquery`; a supported nested `InSubqueryExec` remains in `literalizedChild` together with its non-transient `BaseSubqueryExec` plan. Registering the evaluator in generated references can therefore serialize a full physical subquery tree into every task closure, contradicting the plan-free guarantee and potentially failing on driver-only state. Please make the evaluator consume an already generated LHS value instead of owning the child expression, and add a nested-IN closure regression.
**Recommended change:** Move LHS evaluation back across the InSubqueryExec-to-evaluator boundary and keep only evaluated values and membership state in the task reference.
**Why this works:** Generate the multi-column LHS through Spark's normal expression codegen while the operator's currentVars are active, then pass only that evaluated value and the prepared membership state to a plan-free evaluator instead of letting the evaluator interpret and serialize the child expression.
**Scope:** InSubqueryExec's multi-column codegen branch, the MultiColumnInSubqueryEvaluator interface, and focused residual-join and nested-IN regressions in SubquerySuite.
**Compatibility:** Preserve the single-column InSet path, interpreted multi-column truth table, legacy empty-subquery behavior, and exactly-once initialization and evaluation of nondeterministic children.
**Risks:** The generated value representation must preserve struct-level and field-level nulls. Moving child evaluation must not duplicate nondeterministic evaluation or bypass partition initialization.
**Constraints:** Generated child evaluation must consume the bound ctx.currentVars layout for its operator. The evaluator stored in task references must not retain any BaseSubqueryExec or other driver-side SparkPlan. Nullable structs and fields must retain the existing three-valued IN and NOT IN semantics.
**Success:** A code-generated residual join predicate can read columns from both join sides without wrong ordinals or task failure, and a nested IN expression leaves no BaseSubqueryExec reachable from generated task references.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,79 @@ case class InSubqueryExec(
}
}
+ // Memoized lightweight evaluator instance. @transient so it is rebuilt (from the
+ // serializable fields) when InSubqueryExec is deserialized in an executor; constructed
+ // only for the multi-column path after prepareResult() has populated result.
+ // The child expression is transformed to replace every prepared ScalarSubquery with its
+ // already-evaluated Literal, so that no nested BaseSubqueryExec crosses the serialization
+ // boundary with generated task closures. See SPARK-58481.
+ @transient private[sql] lazy val multiColEvaluator: MultiColumnInSubqueryEvaluator = {
+ // On the SQL execution path, updateResult stores rows.asInstanceOf[Array[Any]] where rows is
+ // Array[InternalRow], so result is always Array[InternalRow] at runtime. Cast directly to
+ // avoid allocating an extra reference array.
+ val rows = result.asInstanceOf[Array[InternalRow]]
+ val fTypes = plan.output.map(_.dataType).toArray
+ // Replace each prepared ScalarSubquery in the LHS with its literal result so the captured
+ // child expression carries no BaseSubqueryExec references into executor task closures.
+ // Ordering is safe: SparkPlan.executeQuery calls waitForSubqueries() -- which calls
+ // updateResult() on every ExecSubqueryExpression found in this plan's expressions, including
+ // nested ScalarSubquery nodes inside child -- before doExecute()/doGenCode() runs and
+ // accesses this lazy val. So every s.toLiteral call below finds a fully-evaluated result.
+ val literalizedChild = child.transformUp {
+ case s: ScalarSubquery => s.toLiteral
+ }
+ new MultiColumnInSubqueryEvaluator(literalizedChild, fTypes, rows, legacyNullInEmptyBehavior)
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
Review Comment:
**Nit (P3):** This dispatch recomputes `plan.output.length` for every outer row even though the plan width is stable for the expression instance. For a common `ProjectExec` subquery, `output` maps the projection list on each call, adding avoidable traversal and allocation to the interpreted hot path. Please cache the multi-column decision on `InSubqueryExec` and reuse it in evaluation and codegen; `withNewPlan` already creates a new instance for a replacement plan.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2682,476 @@ 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).
+ //
+ // The predicate 5 NOT IN (...) is a constant that references neither join input, so
+ // RewritePredicateSubquery returns the original Join unchanged and PlanSubqueries always
+ // constructs InSubqueryExec here regardless of any optimizer config.
+ 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))")
+
+ val query = sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0" +
+ " ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ // Verify the physical plan contains InSubqueryExec so the nullability fix is
+ // actually exercised and not silently bypassed by an optimizer path.
+ // Use AdaptiveSparkPlanHelper.find to traverse AQE wrapper nodes, and recurse
+ // into expression subtrees via Expression.exists to find the wrapped InSubqueryExec.
+ assert(
+ find(query.queryExecution.executedPlan) { p =>
+ p.expressions.exists(_.exists {
+ case _: InSubqueryExec => true
+ case _ => false
+ })
+ }.isDefined,
+ "Expected InSubqueryExec in the physical plan")
+ // Unmatched t1 rows null-pad t0; unmatched t0 rows null-pad t1.
+ checkAnswer(query, Seq(
+ Row(null, 10), Row(null, 20), Row(null, 30),
+ Row(1, null), Row(2, null), Row(3, null)))
+ }
+ }
+
+ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: RHS is fully non-null (stored in nonNullSet). LHS (NULL, 2) vs RHS (99, 2):
+ // second fields equal, first field null => UNKNOWN; exercises the nonNullSet scan with
+ // a nullable LHS. Fixture: lhs.a is nullable; rhs columns are NOT NULL (VALUES-derived
+ // to avoid Parquet dataSchema.asNullable widening that would defeat the RHS control).
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+
+ // Case B: RHS is fully non-null (goes into nonNullSet, not nullRows). LHS (NULL, 1) vs
+ // RHS (99, 2): first field null, but second field 1 != 2 is a definitive mismatch that
+ // makes the candidate FALSE. An implementation that stops at the first UNKNOWN instead
+ // of continuing to check later fields would return UNKNOWN here instead of FALSE, causing
+ // NOT IN to be UNKNOWN and the inner join to return no rows.
+ withTempView("lhs_nn", "rhs_nn") {
+ sql("CREATE TEMPORARY VIEW lhs_nn AS " +
+ "SELECT * FROM VALUES (CAST(NULL AS INT), 1) AS t(a, b)")
+ sql("CREATE TEMPORARY VIEW rhs_nn AS " +
+ "SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 1) NOT IN ((99, 2)): second field 1 != 2 => candidate FALSE => NOT IN TRUE.
+ // Inner join returns the single lhs row.
+ checkAnswer(
+ sql(
+ """SELECT lhs_nn.b FROM lhs_nn JOIN (SELECT 1 AS x)
+ |ON ((lhs_nn.a, lhs_nn.b) NOT IN (SELECT a, b FROM rhs_nn))
+ |""".stripMargin),
+ Seq(Row(1)))
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields " +
+ "for multi-column IN subqueries") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: InSubqueryExec.doGenCode registers Nondeterministic children " +
+ "for partition-level initialization") {
+ // A join-condition IN subquery with a nondeterministic LHS is rejected by CheckAnalysis,
+ // so this direct-construction test provides a focused unit check: it constructs
+ // InSubqueryExec directly (bypassing analysis) to verify that doGenCode correctly
+ // registers each Nondeterministic descendant of the LHS child for partition-level
+ // initialization. Without that registration, a Rand node's eval() would throw
+ // IllegalArgumentException (via require(initialized, ...)) because initialize() was
+ // never called.
+ //
+ // Two output columns force plan.output.length > 1, taking the multi-column fallback
+ // branch (with the explicit foreach loop), not the single-column InSet path where Rand
+ // would register itself through ordinary child codegen traversal.
+ //
+ // The test compiles the expression to bytecode via GeneratePredicate and executes it,
+ // so a missing initialize() call causes an actual failure rather than just a missing
+ // entry in partitionInitializationStatements.
+ val subplanOutput = spark
+ .range(0)
+ .select($"id".cast(IntegerType).as("a"), $"id".cast(IntegerType).as("b"))
+ .queryExecution.executedPlan.output
+ val subplan = SubqueryExec(
+ "test_subquery",
+ LocalTableScanExec(subplanOutput, Nil, None))
+ // LHS: struct of (Literal(1), Rand(42)>=0.0). Rand is always non-negative so the struct
+ // always equals the single RHS row (1, true), making IN = true.
+ val lhs = CreateNamedStruct(Seq(
+ Literal("a"), Literal(1),
+ Literal("b"), GreaterThanOrEqual(Rand(42L), Literal(0.0))))
+ val expr = InSubqueryExec(
+ lhs, subplan, ExprId(99),
+ isDynamicPruning = false,
+ result = Array[InternalRow](InternalRow(1, true)).asInstanceOf[Array[Any]])
+ // Compile to bytecode, initialize (seeds the Rand RNG), and evaluate.
+ // If initialize() were skipped the Rand eval() would throw IllegalArgumentException.
+ val pred = GeneratePredicate.generate(expr, useSubexprElimination = false)
+ pred.initialize(0)
+ assert(pred.eval(InternalRow.empty) === true,
+ "Expected IN to be TRUE: (1, rand()>=0.0) IN ((1, true))")
+ }
+
+ test("SPARK-58481: MultiColumnInSubqueryEvaluator preserves shared Nondeterministic " +
+ "identity after Java serialization round-trip") {
+ // In doGenCode, ctx.references receives both the MultiColumnInSubqueryEvaluator (slot 0)
+ // and the bare Nondeterministic node (slot 1) for partition initialization. On the driver
+ // both slots point to the same Rand instance (evaluator.child embeds it). The risk is that
+ // Java serialization could write two independent copies, causing initialize() called on
+ // references[1] to seed a different instance than eval() uses inside the evaluator.
+ //
+ // Java's ObjectOutputStream tracks shared references within a single graph (via its
+ // internal handle table), so a single serialize/deserialize round-trip of the references
+ // array as one object produces back-references rather than copies. This test verifies
+ // that invariant explicitly using the same JavaSerializer Spark uses for task closures.
+ val rand = Rand(42L)
+ val lhs = CreateNamedStruct(Seq(
+ Literal("a"), Literal(1),
+ Literal("b"), GreaterThanOrEqual(rand, Literal(0.0))))
+ val evaluator = new MultiColumnInSubqueryEvaluator(
+ lhs,
+ Array(IntegerType, IntegerType),
+ Array(InternalRow(1, true)),
+ legacyNullInEmptyBehavior = false)
+ // Simulate what doGenCode does: put the evaluator and the bare Nondeterministic into
+ // the references array as separate slots (as WholeStageCodegenEvaluatorFactory would).
+ val references: Array[Any] = Array(evaluator, rand)
+ // Round-trip through JavaSerializer -- the same serializer Spark uses for task closures
+ // (SparkEnv always sets closureSerializer = new JavaSerializer(...)).
+ // Serializing the array directly is equivalent to serializing it as a field of
+ // WholeStageCodegenEvaluatorFactory: Java's ObjectOutputStream maintains a handle table
+ // across the entire stream, so shared-reference tracking is independent of nesting depth.
+ val ser = new JavaSerializer(new SparkConf()).newInstance()
+ val rt = ser.deserialize[Array[Any]](ser.serialize(references))
+ val rtEvaluator = rt(0).asInstanceOf[MultiColumnInSubqueryEvaluator]
+ val rtRand = rt(1).asInstanceOf[Rand]
+ // Find the Rand inside the deserialized evaluator's child expression tree.
+ var embeddedRand: Rand = null
+ rtEvaluator.child.foreach {
+ case n: Rand => embeddedRand = n
+ case _ =>
+ }
+ assert(embeddedRand ne null,
+ "Expected a Rand node inside the evaluator's child after round-trip")
+ // The critical identity check: the Rand node that would receive initialize(partitionIndex)
+ // must be the same instance eval() uses inside the evaluator -- if Java serialization wrote
+ // two independent copies, initialize() would seed a different instance than eval() uses.
+ assert(embeddedRand eq rtRand,
+ "Rand inside evaluator and separately-registered Rand must be the same instance " +
+ "after serialization round-trip")
+ }
+
+ test("SPARK-58481: scalar subquery in multi-column LHS is literalized before capture") {
+ // If the multi-column LHS contains a nested scalar subquery (e.g. (col, (SELECT max(x)
+ // FROM t)) IN (...)), PlanSubqueries embeds an executable ScalarSubquery whose non-transient
+ // BaseSubqueryExec would otherwise remain reachable through evaluator.child in generated task
+ // closures. MultiColumnInSubqueryEvaluator replaces each ScalarSubquery with its already-
+ // evaluated Literal before capturing child, so no BaseSubqueryExec crosses the boundary.
+ //
+ // Verify via a direct-construction InSubqueryExec (same pattern as the nondeterministic test)
+ // that after a JavaSerializer round-trip the evaluator's child contains only a Literal where
+ // the scalar subquery was, with no ScalarSubquery or BaseSubqueryExec reachable.
+ val scalarSubplanOutput = spark
+ .range(0).select($"id".cast(IntegerType).as("v"))
+ .queryExecution.executedPlan.output
+ val scalarSubplan = SubqueryExec(
+ "scalar_subquery",
+ LocalTableScanExec(scalarSubplanOutput, Seq(InternalRow(42)), None))
+ // Build the scalar subquery expression, pre-update it with its result (42).
+ val scalarSub = ScalarSubquery(scalarSubplan, ExprId(77))
+ scalarSub.updateResult()
+ // Outer subplan for the IN clause: two-column output so plan.output.length > 1.
+ val outerSubplanOutput = spark
+ .range(0)
+ .select($"id".cast(IntegerType).as("a"), $"id".cast(IntegerType).as("b"))
+ .queryExecution.executedPlan.output
+ val outerSubplan = SubqueryExec(
+ "outer_subquery",
+ LocalTableScanExec(outerSubplanOutput, Nil, None))
+ // LHS: (scalarSub, Literal(1)) -- the scalar subquery is the first field.
+ val lhs = CreateNamedStruct(Seq(
+ Literal("a"), scalarSub,
+ Literal("b"), Literal(1)))
+ val expr = InSubqueryExec(
+ lhs, outerSubplan, ExprId(88),
+ isDynamicPruning = false,
+ result = Array[InternalRow](InternalRow(42, 1)).asInstanceOf[Array[Any]])
+ // Access multiColEvaluator to trigger literalization (normally done in doGenCode/eval).
+ val evaluator = expr.multiColEvaluator
+ // Serialize and deserialize the evaluator via JavaSerializer -- the task-closure serializer.
+ val ser = new JavaSerializer(new SparkConf()).newInstance()
+ val rtEvaluator = ser.deserialize[MultiColumnInSubqueryEvaluator](
+ ser.serialize(evaluator))
+ // The deserialized child must contain no ScalarSubquery or BaseSubqueryExec.
+ var foundScalarSubquery = false
+ rtEvaluator.child.foreach {
+ case _: ScalarSubquery => foundScalarSubquery = true
+ case _ =>
+ }
+ assert(!foundScalarSubquery,
+ "ScalarSubquery (and thus BaseSubqueryExec) must not be reachable from the " +
+ "deserialized evaluator's child -- scalar subqueries must be literalized first")
+ // The evaluator must still produce the correct result: (42, 1) IN ((42, 1)) = true.
+ assert(evaluator.eval(InternalRow.empty) === true,
Review Comment:
**Non-blocking (P2):** The test serializes `evaluator` into `rtEvaluator` and inspects the round-tripped child, but this final behavior assertion calls the original object. A deserialization-only failure in the transient orderings, sets, null rows, or child evaluation would still leave the test green. Please call `rtEvaluator.eval(InternalRow.empty)` here so the regression exercises the executor-side instance it claims to protect.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5668488066
Thanks for the review @cloud-fan, have addressed all four:
- LHS binding (P1): the multi-column codegen branch now generates the LHS through `child.genCode(ctx)` like every other expression so it consumes the operator's bound `ctx.currentVars` (e.g. streamed ++ build variables of a residual join condition) instead of interpreting the bound child against ctx.INPUT_ROW. Only the evaluated struct value is passed to the evaluator. Added an E2E regression with a whole-stage code-generated hash join whose NOT IN LHS references both join sides, asserting the join runs inside `WholeStageCodegenExec` with `InSubqueryExec` in its condition.
- Nested plans in references (P2): `MultiColumnInSubqueryEvaluator` no longer owns any expression. It holds only the result rows, their field types and the legacy empty-list flag, so nothing reachable through the LHS (scalar subquery, nested `InSubqueryExec` or anything added later) can enter task closures. The `ScalarSubquery` literalization and the manual Nondeterministic registration loop are gone; those nodes now register their own state through normal codegen. Added a regression with both a `ScalarSubquery` and a nested `InSubqueryExec` in the LHS that asserts no SparkPlan or expression holding a PlanExpression lands in `ctx.references` and executes the generated predicate.
- Serialization test (P2): the round-trip test now calls eval on the deserialized instance across TRUE / UNKNOWN / FALSE / null-field / null-struct cases.
- Dispatch caching (P3): `isMultiColumn` is computed once per instance and used in nullable, updateResult, eval and `doGenCode`.
Two additional hardenings while re-verifying the earlier rounds: the SQL cases that used an inner join with a one-sided IN were being pushed into a Filter by `PushPredicateThroughJoin` and rewritten to a semi/anti join so they never reached `InSubqueryExec`. Converted them to FULL OUTER JOIN and added a shared plan assertion to every SQL case. Also added a regression with implicitly widened LHS/RHS types (SMALLINT/INT vs INT/BIGINT), since the evaluator now derives its orderings from the subquery output types.
14/14 SPARK-58481 tests, the full SubquerySuite, both dpp suites and the in-subquery golden files pass locally.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5530942568
Thanks for the review @cloud-fan , have addressed the architectural concern by extracting a lightweight `MultiColumnInSubqueryEvaluator` that excludes `plan/BaseSubqueryExec` from codegen references, so task closures no longer carry the driver's physical subquery plan.
Verified the shared-Nondeterministic-identity risk you flagged with a Java serialization round-trip test confirming `ctx.references` preserves object identity across the driver/executor boundary. Also added the fully-non-null-RHS late-mismatch regression, fixed the early-return in the legacy empty-RHS path, and corrected the Case A comment.
11/11 tests pass. Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5589560101
Gentle ping @cloud-fan , can you please take a look? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5339048701
Thanks for the review @cloud-fan , have changed to `plan.output.exists(_.nullable)` to mirror the logical `InSubquery.nullable` contract for all output columns, including the legacy config. Added a focused multi-column regression where the first column is NOT NULL but the second is nullable.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3914041627
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,277 @@ class SubquerySuite extends SharedSparkSession
assert(exposedAttribute.exprId == outerReferenceAttribute.exprId)
}
+
+ test("SPARK-58481: InSubqueryExec nullable correctly accounts for subquery output nullability") {
Review Comment:
**Non-blocking (P2):** This test does not currently exercise `InSubqueryExec.nullable`: the default `spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled=true` rewrites this join-condition IN before `PlanSubqueries`. Removing the RHS-derived nullability would still leave the six-row assertion green. Please disable that rewrite here and assert that the executed condition contains `InSubqueryExec` before checking the result.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,187 @@ 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 two sorted sets, both using the struct-level Catalyst ordering
+ // so that duplicate rows are deduplicated. Fully non-null rows go into multiColNonNullSet
+ // for O(log n) membership tests; rows with at least one null field go into multiColNullRows
+ // (also a TreeSet, not an Array) so each distinct null-containing row is scanned at most
+ // once per outer row regardless of RHS duplicate multiplicity. See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
+ 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
+ // Cache lazy accessors in locals so the loop bodies do not re-enter them on every iteration.
+ val nullRows = multiColNullRows
+ val nonNullSet = multiColNonNullSet
+
+ if (!inputStruct.anyNull) {
+ // Fast path: indexed lookup among fully non-null candidates.
+ if (nonNullSet.contains(inputStruct)) return true
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (nullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
+ val candidate = nullRows(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(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (nullRows.isEmpty && nonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ var hasUnknown = false
+ // Scan null-containing result rows first.
+ var i = 0
+ while (i < nullRows.length && !hasUnknown) {
Review Comment:
**Non-blocking (P2):** No added test reaches this loop with null-containing rows on both sides. Please add a forced physical multi-column case that distinguishes `(NULL, 1) IN ((NULL, 1))` (UNKNOWN) from `(NULL, 1) IN ((NULL, 2))` (FALSE). An implementation that returns UNKNOWN at the first null would otherwise pass every current case while missing the later definitive mismatch.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,187 @@ 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 two sorted sets, both using the struct-level Catalyst ordering
+ // so that duplicate rows are deduplicated. Fully non-null rows go into multiColNonNullSet
+ // for O(log n) membership tests; rows with at least one null field go into multiColNullRows
+ // (also a TreeSet, not an Array) so each distinct null-containing row is scanned at most
+ // once per outer row regardless of RHS duplicate multiplicity. See SPARK-58481.
+ @transient private lazy val (multiColNonNullSet, multiColNullRows) = {
+ val withNull = TreeSet.newBuilder[InternalRow](multiColRowOrdering)
+ 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().toArray)
+ }
+
+ // 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 = {
+ // Current behavior (ANSI on, or legacyNullInEmptyBehavior=false): IN (empty set) is always
+ // FALSE without evaluating the LHS. Legacy behavior (ANSI off by default) returns NULL when
+ // the LHS is null and FALSE otherwise, requiring the LHS to be evaluated. Mirror InSet.eval's
+ // guard exactly: skip child.eval only when legacyNullInEmptyBehavior is false (SPARK-44550).
+ if (result.isEmpty && !legacyNullInEmptyBehavior) return false
Review Comment:
**Non-blocking (P2):** The new empty-RHS regression covers only the non-legacy branch that skips `child.eval`. Please add the complementary forced-`InSubqueryExec` case with ANSI enabled and `spark.sql.legacy.nullInEmptyListBehavior=true`, and verify that a division-by-zero LHS raises `DIVIDE_BY_ZERO`. Otherwise an unconditional early FALSE return would pass the current test while breaking the compatibility branch.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +197,187 @@ 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 two sorted sets, both using the struct-level Catalyst ordering
+ // so that duplicate rows are deduplicated. Fully non-null rows go into multiColNonNullSet
+ // for O(log n) membership tests; rows with at least one null field go into multiColNullRows
+ // (also a TreeSet, not an Array) so each distinct null-containing row is scanned at most
Review Comment:
**Nit (P3):** `multiColNullRows` is an `Array[InternalRow]`, not a `TreeSet`: the initializer ends with `withNull.result().toArray`, and the evaluator uses array indexing. Please update this comment so it describes the deduplication step without claiming the stored representation remains a sorted set.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,277 @@ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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 (VALUES-derived to avoid Parquet
+ // dataSchema.asNullable widening that would defeat the RHS non-nullability control).
+ // (NULL, 2) vs (99, 2): second fields equal (2=2), first field is null => UNKNOWN.
+ // (1, 1) vs (99, 2): first field 1!=99 => FALSE => NOT IN = TRUE => matches all rhs rows.
+ // FULL OUTER JOIN: (NULL,2) gets null-padded (UNKNOWN condition); (1,1) joins with (99,2);
+ // since (1,1) matched rhs(99,2), rhs(99,2) is not null-padded.
+ // Pre-fix: (NULL,2) NOT IN is wrongly TRUE (null suppressed) => emits (null,99); no
+ // null-padded rows. Post-fix: UNKNOWN propagated => emits (null,null) for (NULL,2).
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs(99,2) is matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields multi-column") {
Review Comment:
**Nit (P3):** This test name is missing a connector before `multi-column`. For example, `LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields for multi-column IN subqueries` states the scenario clearly.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2681,277 @@ 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.
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ // Case A: NULL after a definitive match (null in non-head position after matching head).
+ // RHS: (99,99) and (1,NULL).
+ // (1,1) vs (99,99): first field 1!=99 => FALSE.
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // Overall for (1,1): UNKNOWN => NOT IN = null-padded.
+ // (2,2) vs both: all FALSE => NOT IN = TRUE => joins with both rhs rows.
+ withTempView("lhs", "rhs") {
+ sql("CREATE TEMPORARY VIEW lhs AS SELECT * FROM VALUES (1, 1), (2, 2) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs AS
+ |SELECT * FROM VALUES (99, 99), (1, CAST(NULL AS INT)) AS t(a, b)""".stripMargin)
+ 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)))
+ }
+
+ // Case B: NULL in head position followed by a definitive mismatch in a later field.
+ // RHS: (NULL, 99).
+ // (1,1) vs (NULL,99): first field null => UNKNOWN so far; second field 1!=99 => FALSE.
+ // A later definitive mismatch must override the earlier UNKNOWN: result is FALSE,
+ // NOT IN = TRUE. A field-order regression would leave (1,1) as UNKNOWN instead.
+ withTempView("lhs2", "rhs2") {
+ sql("CREATE TEMPORARY VIEW lhs2 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs2 AS
+ |SELECT * FROM VALUES (CAST(NULL AS INT), 99) AS t(a, b)""".stripMargin)
+ // (1,1) NOT IN ((NULL,99)): second field 1!=99 makes the candidate FALSE =>
+ // NOT IN = TRUE => inner join returns the single matching row.
+ checkAnswer(
+ sql(
+ """SELECT lhs2.a FROM lhs2 JOIN (SELECT 1 AS a)
+ |ON ((lhs2.a, lhs2.b) NOT IN (SELECT a, b FROM rhs2))""".stripMargin),
+ Seq(Row(1)))
+ }
+
+ // Case C: UNKNOWN candidate followed by an exact-match candidate => IN = TRUE.
+ // RHS: (1,NULL) and (1,1).
+ // (1,1) vs (1,NULL): first fields equal, second null => UNKNOWN.
+ // (1,1) vs (1,1): exact match => TRUE.
+ // The exact match must dominate the UNKNOWN: IN = TRUE, NOT IN = FALSE.
+ // A candidate-order regression would short-circuit on UNKNOWN and miss the TRUE.
+ withTempView("lhs3", "rhs3") {
+ sql("CREATE TEMPORARY VIEW lhs3 AS SELECT * FROM VALUES (1, 1) AS t(a, b)")
+ sql(
+ """CREATE TEMPORARY VIEW rhs3 AS
+ |SELECT * FROM VALUES (1, CAST(NULL AS INT)), (1, 1) AS t(a, b)""".stripMargin)
+ // (1,1) IN ((1,NULL),(1,1)): exact match exists => IN = TRUE => inner join returns row.
+ checkAnswer(
+ sql(
+ """SELECT lhs3.a FROM lhs3 JOIN (SELECT 1 AS a)
+ |ON ((lhs3.a, lhs3.b) IN (SELECT a, b FROM rhs3))""".stripMargin),
+ Seq(Row(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 (VALUES-derived to avoid Parquet
+ // dataSchema.asNullable widening that would defeat the RHS non-nullability control).
+ // (NULL, 2) vs (99, 2): second fields equal (2=2), first field is null => UNKNOWN.
+ // (1, 1) vs (99, 2): first field 1!=99 => FALSE => NOT IN = TRUE => matches all rhs rows.
+ // FULL OUTER JOIN: (NULL,2) gets null-padded (UNKNOWN condition); (1,1) joins with (99,2);
+ // since (1,1) matched rhs(99,2), rhs(99,2) is not null-padded.
+ // Pre-fix: (NULL,2) NOT IN is wrongly TRUE (null suppressed) => emits (null,99); no
+ // null-padded rows. Post-fix: UNKNOWN propagated => emits (null,null) for (NULL,2).
+ withSQLConf(
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs") {
+ withTempView("rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ // rhs as VALUES view: both columns inferred non-nullable from all-literal rows.
+ sql("CREATE TEMPORARY VIEW rhs AS SELECT * FROM VALUES (99, 2) AS t(a, b)")
+ // (NULL, 2) NOT IN ((99,2)): second fields match, first is null => UNKNOWN
+ // => join condition not TRUE => (NULL,2) is null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): first field 1!=99 => FALSE => NOT IN = TRUE
+ // => (1,1) joins with rhs(99,2): Row(1, 99). rhs(99,2) is matched; no null-padded rhs.
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY suppresses RHS-only nullability") {
+ // Legacy mode suppresses only RHS-derived nullability (plan.output nullable).
+ // A non-nullable scalar LHS (Literal 5) has lhsNullable=false; with RHS suppressed,
+ // nullable=false. The generated code omits null handling and NOT IN on a subquery that
+ // returns NULL evaluates to TRUE -- the pre-fix single-column behaviour the flag preserves.
+ // Note: intentionally codegen-specific. The interpreted path correctly returns UNKNOWN
+ // regardless of nullable (6 rows); the assertion of 9 verifies codegen ran.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ 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))")
+
+ // Legacy: lhsNullable=false (Literal 5), rhsNullable suppressed => nullable=false.
+ // Generated code suppresses null; 5 NOT IN (99, NULL) evaluates to TRUE.
+ // FULL OUTER JOIN condition is TRUE => full cross product of 3 x 3 = 9 rows.
+ assert(sql(
+ "SELECT t0.c0, t1.c0 FROM t1 FULL OUTER JOIN t0 ON (5 NOT IN (SELECT t3.c0 FROM t3))")
+ .count() === 9)
+ }
+ }
+ }
+
+ test("SPARK-58481: LEGACY_IN_SUBQUERY_NULLABILITY preserves nullable LHS fields multi-column") {
+ // Legacy mode suppresses RHS nullability but preserves LHS field nullability.
+ // With a nullable LHS field, lhsNullable=true even in legacy mode, so nullable=true.
+ // Generated NOT IN code propagates UNKNOWN correctly; result is identical to non-legacy.
+ withSQLConf(
+ SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY.key -> "true",
+ "spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled" -> "false"
+ ) {
+ withTable("lhs", "rhs") {
+ sql("CREATE TABLE lhs(a INT, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO lhs VALUES (1, 1), (NULL, 2)")
+ sql("CREATE TABLE rhs(a INT NOT NULL, b INT NOT NULL) USING PARQUET")
+ sql("INSERT INTO rhs VALUES (99, 2)")
+ // (NULL, 2) NOT IN ((99,2)): first field null => UNKNOWN => null-padded: Row(null, null).
+ // (1, 1) NOT IN ((99,2)): 1!=99 => FALSE => NOT IN=TRUE => joins: Row(1, 99).
+ 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, 99), Row(null, null)))
+ }
+ }
+ }
+
+ test("SPARK-58481: InSubqueryExec.doGenCode registers Nondeterministic children " +
+ "for partition-level initialization") {
+ // A nondeterministic expression in the LHS of a multi-column IN subquery is unreachable
+ // via SQL: filter IN subqueries are rewritten to LeftSemi by RewritePredicateSubquery
+ // before PlanSubqueries runs, and join-condition IN subqueries with nondeterministic
+ // operands are rejected by CheckAnalysis. Construct InSubqueryExec directly to verify
+ // that doGenCode correctly registers each Nondeterministic descendant of the LHS child
+ // for partition-level initialization. Without that registration, a Rand node's eval()
+ // would throw IllegalArgumentException (via require(initialized, ...)) because
+ // initialize() was never called.
+ //
+ // Two output columns forces plan.output.length > 1, taking the multi-column fallback
Review Comment:
**Nit (P3):** Minor grammar fix: `Two output columns` is plural, so this should read `Two output columns force plan.output.length > 1`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #58077: [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability
URL: https://github.com/apache/spark/pull/58077
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3841252831
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +186,168 @@ 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). 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.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ 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 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))
Review Comment:
**Non-blocking:**
Cache the LHS field values once before entering the candidate scans, then reuse them at `subquery.scala:252`, `subquery.scala:273`, and `subquery.scala:293`. For an `UnsafeRow` BinaryType field, `InternalRow.get` reaches `getBinary`, which allocates and copies the full byte array on every candidate even though the LHS is invariant.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3813494004
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -162,17 +179,44 @@ case class InSubqueryExec(
require(result != null || resultBroadcast != null, s"$this has not finished")
if (result == null && resultBroadcast != null) {
result = resultBroadcast.value
+ // Recompute after broadcast deserialisation: @transient fields are not serialised.
+ if (plan.output.length > 1 && !isResultUnavailable) {
+ hasNullInMultiColumnResult =
+ result.exists(_.asInstanceOf[InternalRow].anyNull)
+ }
}
}
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else {
+ val r = inSet.eval(input)
+ // InSet.hasNull misses null fields inside rows; propagate UNKNOWN on a miss manually.
+ if (r == false && hasNullInMultiColumnResult) null else r
Review Comment:
**Blocking:**
**Blocking:** This global flag is too coarse for row-valued `IN`. For `(1,1)` versus `(99,NULL)`, the first field comparison is FALSE, so the whole candidate is FALSE even though the second comparison is UNKNOWN; this branch instead returns NULL for every miss once any result row contains NULL. Please preserve per-result-row three-valued comparison state, returning UNKNOWN only when no candidate is TRUE and at least one candidate remains UNKNOWN, and update the `SubquerySuite` fixture to cover both definitely-unequal and potentially-equal nullable rows.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5345616791
Thanks for the review @cloud-fan , replaced the coarse hasNullInMultiColumnResult global flag with a per-candidate three-valued comparison in evalMultiColumn: the inner loop short-circuits to FALSE on the first non-null field mismatch, sets UNKNOWN only when a null field is reached after all non-null fields matched and returns TRUE immediately on an exact match.
The test fixture is updated to (99,99),(1,NULL) to cover both paths: (1,1) vs (99,NULL) → FALSE (first field differs); (1,1) vs (1,NULL) → UNKNOWN; (2,2) → TRUE (both candidates definitively unequal).
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5505135700
Thanks for the review @cloud-fan , have addressed all three:
- Corrected the empty-set comment to accurately describe both branches (current behavior skips `child.eval` entirely and legacy behavior still evaluates it and can return NULL)
- Fixed the nondeterministic-test rationale: filter-IN is unreachable via `RewritePredicateSubquery`'s rewrite to LeftSemi (not `CheckAnalysis`), and a pre-initialize() eval throws `IllegalArgumentException` (not `IllegalStateException`)
- Replaced the `partitionInitializationStatements` string check with an actual compile-and-execute test via `GeneratePredicate.generate`, confirmed with a negative control and removing the registration loop makes the test fail with exactly that `IllegalArgumentException`
8/8 tests pass. Can you PTALa again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5393075067
Thanks for the review @cloud-fan , moved inputFields materialization after the TreeSet fast-path check so exact-match hits skip it entirely, and cache it before both slow-path scans (avoids repeated BinaryType copies via InternalRow.get).
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5406353514
Thanks for the review @cloud-fan , have addressed all five:
- Reworded the RHS-nullability comment to describe a possible outcome, not unconditional
- Clarified both "scan all result rows" comments to reflect the !hasUnknown early-exit that's already in the code
- Removed the `BinaryType` copy-avoidance rationale for `inputFields` (LHS is a `GenericInternalRow` from `CreateNamedStruct`, so .get() doesn't copy)
- Fixed the reversed null-pad labels in the first test's comment
- Replaced Parquet-backed tables with VALUES-derived temp views in the non-head-nullability test, since Parquet file-source analysis applies `dataSchema.asNullable` and was silently defeating the test's nullability control
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5550833437
Thanks @cloud-fan for the review, have addressed the architectural gap: the evaluator's captured child is now transformed via `transformUp` to replace every prepared `ScalarSubquery` with its already-evaluated Literal before entering `ctx.references`, so no nested `BaseSubqueryExec` (from a scalar subquery embedded in the multi-column LHS) crosses the task-closure serialization boundary.
Verified the ordering is safe by tracing `SparkPlan.executeQuery` - `waitForSubqueries()` (which resolves all nested `ScalarSubquery` results) always completes before `doExecute()/doGenCode()` can force the evaluator's lazy construction. Added a regression that serializes the evaluator via `JavaSerializer` and confirms no `ScalarSubquery/BaseSubqueryExec` remains reachable in the deserialized child while `eval()` still produces the correct result.
Also folded in the three P3 fixes (accurate `CreateNamedStruct` comment, accurate legacy/ANSI wording, avoided the extra array copy in `multiColEvaluator`).
12/12 tests pass locally. Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5370195834
Thanks for the review @cloud-fan , CI is running now and have added a focused regression for `LEGACY_IN_SUBQUERY_NULLABILITY=true` that verifies the intentional pre-fix behavior (5 NOT IN (99, NULL) → TRUE, 9-row cross product) is preserved. All 5 SPARK-58481 tests pass too.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3836466082
##########
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). 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.
+ // 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)) {
+ val lhsNullable = child match {
+ case cns: CreateNamedStruct => cns.valExprs.exists(_.nullable)
+ case _ => child.nullable
+ }
+ lhsNullable || plan.output.exists(_.nullable)
+ } else {
+ child.nullable
Review Comment:
**Blocking:**
Preserve nullable fields from a row-valued LHS in the legacy branch. The flag removes only RHS-derived nullability, but `child` is a non-nullable `CreateNamedStruct`, so this returns false even when one of its fields is nullable and generated `NOT IN` code can turn UNKNOWN into TRUE. Compute LHS field nullability in both modes, apply the flag only to `plan.output` nullability, and add a legacy multi-column regression.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +187,154 @@ 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 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). 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.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ 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 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 non-null rows: a null LHS comparison is UNKNOWN unless a non-null field differs.
+ val nonNullIter = multiColNonNullSet.iterator
+ while (nonNullIter.hasNext && !hasUnknown) {
+ val candidate = nonNullIter.next()
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputStruct.get(fieldIdx, fieldTypes(fieldIdx))
+ if (inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputField, candidate.get(fieldIdx, fieldTypes(fieldIdx))) != 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ }
+ if (hasUnknown) null else false
+ }
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ evalMultiColumn(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path.
+ val resultRef = ctx.addReferenceObj("inSubqueryResult", this)
Review Comment:
**Blocking:**
Reuse `CodegenFallback`'s initialization protocol before invoking `eval` here. This hand-written path does not initialize nondeterministic LHS descendants, so a row-valued `IN` such as `(rand(), id) IN (...)` can reach `Nondeterministic.eval` before `initialize()` and fail at runtime.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +187,154 @@ 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) =>
Review Comment:
**Non-blocking:**
Use `TreeSet.newBuilder` to construct the non-null index. `s + row` creates a new immutable red-black path for every result row, while the builder preserves the same ordering and final immutable set with mutable construction.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5436915120
Thanks for the review @cloud-fan , have addressed the blocking item: captured `legacyNullInEmptyBehavior `at construction (mirroring `InSet`) and added the empty-result short-circuit before `child.eval` in `evalMultiColumn`.
Confirmed a plain WHERE IN gets rewritten to LeftSemi before `PlanSubqueries` runs, so the regression forces the `InSubqueryExec` path via a FULL OUTER JOIN ON condition (same pattern as the other tests here) and verified it reaches `evalMultiColumn` and that the division-by-zero LHS doesn't throw under ANSI.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5435441612
Thanks for the review @cloud-fan , have addressed all three:
- Fixed the nullability comment to describe `UNKNOWN` as possible, not inevitable, when the LHS field is null
- Cached `multiColNullRows/multiColNonNullSet` as locals (`nullRows/nonNullSet`) before the scan loops, avoiding repeated lazy-val accessor overhead per candidate
- Added a codegen-enabled regression with rand() in a multi-column IN LHS to exercise the nondeterministic partition-initialization path end-to-end
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5521527103
Thanks for the review @cloud-fan , have addressed all four:
- Extended the nullable-LHS/non-nullable-RHS test with Case B: a fully non-null RHS (99, 2) against nullable `LHS (NULL, 1)`, asserting NOT IN is TRUE via the second field's definitive mismatch which exercises the `nonNullSet` scan specifically, distinct from the nullRows loop the existing case covered
- Added an end-to-end SQL regression `(SELECT (id, rand(42) >= 0.0) IN (...) FROM lhs)` proving the nondeterministic codegen fallback is reachable via a Project, and corrected the prior test's comment such that it's not unreachable via SQL, just that the direct-construction test is a more targeted check of the registration logic
- Removed the unnecessary `withSQLConf` override on the primary nullability test and fixed the comment: 5 NOT IN (...) is a constant referencing neither join input, so `RewritePredicateSubquery` leaves the Join unchanged regardless of the optimizer conf
- Reordered the null-check in the candidate scan to test `inputField == null` before calling `candidate.get(...)`, avoiding an unnecessary row access when the LHS field alone already determines UNKNOWN
11/11 tests pass. Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5327117530
Thank you @anshulbaliga7!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3823269824
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -125,7 +127,17 @@ 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 the child is nullable.
+ // Respects LEGACY_IN_SUBQUERY_NULLABILITY to stay in sync with the logical node.
+ // See SPARK-58481.
+ override def nullable: Boolean = {
+ if (!SQLConf.get.getConf(SQLConf.LEGACY_IN_SUBQUERY_NULLABILITY)) {
+ child.nullable || plan.output.exists(_.nullable)
Review Comment:
**Blocking:**
Row-valued `IN` must remain nullable when any left-hand field is nullable. For multi-column inputs, `child` is the `CreateNamedStruct` from `PlanSubqueries`, whose top-level `nullable` is always false; with a non-nullable RHS, this declares the expression non-nullable even though `evalMultiColumn` returns UNKNOWN for a null LHS field. Generated `NOT IN` can then turn UNKNOWN into TRUE. Please derive LHS nullability from the struct fields or underlying values and add a nullable-LHS/non-nullable-RHS codegen regression.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +177,81 @@ case class InSubqueryExec(
}
}
+ // Three-valued IN semantics for multi-column subqueries.
+ // Result rows are InternalRow objects; InSet uses TreeSet ordering which treats null fields as
+ // non-equal and therefore cannot distinguish a definitively-false candidate (a non-null field
+ // differs) from an indeterminate one (all non-null fields match but some fields are null).
+ // We replicate In.eval's per-candidate logic: TRUE if any candidate matches exactly,
+ // UNKNOWN if no TRUE and at least one candidate is indeterminate, FALSE otherwise.
+ private def evalMultiColumn(inputRow: InternalRow): Any = {
+ val value = child.eval(inputRow)
+ if (value == null) return null
+ val inputStruct = value.asInstanceOf[InternalRow]
+ val output = plan.output
+ val numFields = output.length
+ // Hoist per-field orderings: invariant across all candidate rows.
+ val orderings = Array.tabulate(numFields)(i =>
+ TypeUtils.getInterpretedOrdering(output(i).dataType))
+ val fieldTypes = Array.tabulate(numFields)(i => output(i).dataType)
+ var hasUnknown = false
+ var i = 0
+ while (i < result.length) {
Review Comment:
**Blocking:**
Please avoid scanning the entire collected subquery result on every predicate evaluation. The generated path calls this same loop, while `InSet` already provides indexed struct membership; on a nested-loop join this adds another factor of subquery cardinality to every candidate pair. Keep the indexed fast path for provably non-null rows and precompute the schema, orderings, and any null-aware candidate index outside `evalMultiColumn`.
##########
sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-joins.sql:
##########
@@ -389,6 +389,20 @@ ON s1.id = s2.id
AND s1.id NOT IN (SELECT id FROM s3);
+-- SPARK-58481: NOT IN subquery with NULL in result and non-nullable child on FULL OUTER JOIN.
+-- 5 NOT IN (99, NULL) evaluates to UNKNOWN; a join condition that is not TRUE matches no rows,
+-- so FULL OUTER JOIN must emit null-padded rows for every row from each side (3+3=6 rows).
Review Comment:
**Nit:**
`s1` and `s2` contain five rows each, and the generated result contains ten null-padded rows.
```suggestion
-- so FULL OUTER JOIN must emit null-padded rows for every row from each side (5+5=10 rows).
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5490085086
@cloud-fan gentle ping, can you please take a look? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58077:
URL: https://github.com/apache/spark/pull/58077#discussion_r3904588400
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -165,14 +194,177 @@ 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
+ // No null-containing candidates: no path to UNKNOWN, result is FALSE.
+ if (multiColNullRows.isEmpty) return false
+ // Materialize LHS fields once before the candidate scans to avoid repeated get() calls
+ // inside the per-candidate loop.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ // Slow path: scan null-containing candidates for potential UNKNOWN.
+ // Stop early once hasUnknown is set: the indexed lookup already ruled out TRUE,
+ // and every row here contains NULL, so no later candidate can improve UNKNOWN to TRUE.
+ var hasUnknown = false
+ 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 candidateField = candidate.get(fieldIdx, fieldTypes(fieldIdx))
+ if (candidateField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(inputFields(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 both result sets, stopping once UNKNOWN
+ // is established (a null LHS field can produce UNKNOWN against any non-null RHS row
+ // whose other fields all match).
+ // No candidates at all: result is FALSE (no match possible).
+ if (multiColNullRows.isEmpty && multiColNonNullSet.isEmpty) return false
+ // Materialize LHS fields once before the scans to avoid repeated get() calls.
+ val inputFields = Array.tabulate(numFields)(i => inputStruct.get(i, fieldTypes(i)))
+ 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 = inputFields(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 non-null rows: a null LHS comparison is UNKNOWN unless a non-null field differs.
+ val nonNullIter = multiColNonNullSet.iterator
+ while (nonNullIter.hasNext && !hasUnknown) {
+ val candidate = nonNullIter.next()
+ var fieldIdx = 0
+ var candidateIsUnknown = false
+ var candidateIsFalse = false
+ while (fieldIdx < numFields && !candidateIsFalse) {
+ val inputField = inputFields(fieldIdx)
+ if (inputField == null) {
+ candidateIsUnknown = true
+ } else if (orderings(fieldIdx).compare(
+ inputField, candidate.get(fieldIdx, fieldTypes(fieldIdx))) != 0) {
+ candidateIsFalse = true
+ }
+ fieldIdx += 1
+ }
+ if (!candidateIsFalse && candidateIsUnknown) hasUnknown = true
+ }
+ if (hasUnknown) null else false
+ }
+ }
+
override def eval(input: InternalRow): Any = {
prepareResult()
- if (isResultUnavailable) true else inSet.eval(input)
+ if (isResultUnavailable) {
+ true
+ } else if (plan.output.length > 1) {
+ evalMultiColumn(input)
+ } else {
+ inSet.eval(input)
+ }
}
override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
prepareResult()
- if (isResultUnavailable) Literal.TrueLiteral.doGenCode(ctx, ev) else inSet.doGenCode(ctx, ev)
+ if (isResultUnavailable) {
+ Literal.TrueLiteral.doGenCode(ctx, ev)
+ } else if (plan.output.length > 1) {
+ // Multi-column: per-candidate three-valued comparison cannot be expressed with InSet's
+ // generated code. Fall back to the interpreted path via eval().
+ // Register any Nondeterministic descendants (e.g. rand() in the LHS) for partition-level
+ // initialization, mirroring CodegenFallback's protocol.
+ val resultIdx = ctx.references.length
+ ctx.references += this
+ child.foreach {
+ case n: expressions.Nondeterministic =>
Review Comment:
The new rand() regression still does not exercise this branch: its Filter IN predicate is rewritten to a LeftSemi join before PlanSubqueries, so the executed plan has no InSubqueryExec. Please force a physical multi-column InSubqueryExec and assert that its generated fallback evaluates after partition initialization; a direct expression-level test avoids the analyzer restrictions on nondeterministic join conditions.
<!-- SPARK_DEV_REVIEW_REPLY {"finding_id":"finding:b1e87d31cc02e5ae06b8","thread_id":"inline:3868490742","verdict_sha256":"488117f430da2ac6a7f4ce254ddc1887f0bbbb003880b9c3050c23998e76f23d"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58481][SQL] Fix InSubqueryExec.nullable to account for subquery output nullability [spark]
Posted by "anshulbaliga7 (via GitHub)" <gi...@apache.org>.
anshulbaliga7 commented on PR #58077:
URL: https://github.com/apache/spark/pull/58077#issuecomment-5341330947
Thanks for the review @cloud-fan , have addressed both points: added `hasNullInMultiColumnResult` (scanned via `rows.exists(_.anyNull)`, recomputed after broadcast deserialisation) and promote a `false` miss to `null` in both `eval` and `doGenCode`. Codegen falls back to the interpreted path since `InSet`'s generated `isNull` is the literal `false` in this case which is not assignable. Also replaced the count assertion with `checkAnswer` on actual null-padded rows, and added `withSQLConf` to bypass the optimizer's ExistenceJoin rewrite so the test actually exercises `InSubqueryExec`.
Can you PTAL again? Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org