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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
