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: [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]
