This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new d52a8e13efde [SPARK-58165][SQL][CONNECT] Fix SubqueryExpression
dropping nested subquery plans in IN values
d52a8e13efde is described below
commit d52a8e13efde20fb9ae0e36b116516b61e279eaa
Author: YangJie <[email protected]>
AuthorDate: Wed Jul 22 12:41:19 2026 -0700
[SPARK-58165][SQL][CONNECT] Fix SubqueryExpression dropping nested subquery
plans in IN values
### What changes were proposed in this pull request?
`SubqueryExpression` (in `sql/api`'s `internal/columnNodes.scala`) was the
only `ColumnNode` that returned `children = Seq.empty` even when its
`SubqueryType` was `IN(values)`, whose `values: Seq[ColumnNode]` can hold
nested column nodes -- including another `SubqueryExpression`. It also did not
override `normalize()`, unlike every sibling node.
This PR makes `SubqueryExpression` consistent with the other nodes:
- `children` now exposes `IN.values` (and stays empty for
`SCALAR`/`EXISTS`). The subquery plan `ds` is intentionally not a child -- it
is a DataFrame plan, not a column node, and the converters handle it separately.
- `normalize()` is overridden to strip `origin` and normalize the
`IN.values`, mirroring sibling nodes.
### Why are the changes needed?
A nested subquery is reachable through the public 4.1 API, e.g.
```scala
t1.select(max($"v")).scalar().isin(t2.select($"c"))
// WHERE (SELECT max(v) FROM t1) IN (SELECT c FROM t2)
```
On Spark Connect, `SparkSession.newDataset` collects subquery plan
references via `node.collect { case _: SubqueryExpression => ... }`, which
recurses through `children`. Because `children` was empty, the inner scalar
subquery's plan was never added to `WithRelations`, while the proto converter
still emitted a planId reference to it. The server then failed resolving that
planId:
```
[CONNECT_INVALID_PLAN.ASSERTION_FAILURE] Missing relation in WithRelations:
(N) not in (M)
```
Spark Classic is unaffected: its converter embeds the subquery's logical
plan directly and does not rely on `children`. The missing `normalize()`
override is a separate, lower-impact inconsistency: two Columns built from the
same `ds` at different call sites compared unequal purely due to retained
`origin`.
### Does this PR introduce _any_ user-facing change?
Yes. On Spark Connect, a nested subquery inside `isin(ds)` -- e.g.
`df1.scalar().isin(df2)` -- previously failed with an internal `Missing
relation in WithRelations` error. After this fix it works, matching Spark
Classic and the equivalent SQL. This is a bug fix relative to released 4.1.x.
### How was this patch tested?
New cases in `DataFrameSubquerySuite` (`sql/core` Classic +
`sql/connect/client/jvm` Connect) exercise a scalar subquery nested inside an
IN subquery's values. Verified reproduce-then-fix: the Connect case fails with
`Missing relation in WithRelations` on the unfixed tree and passes after the
fix; both full suites pass with no regressions. Direct normalization coverage
for `SubqueryExpression` (SCALAR / EXISTS / IN(values)) was added to
`ColumnNodeSuite`, verifying origin stripping an [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57308 from LuciferYang/SPARK-subquery-columnnode.
Authored-by: YangJie <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 5994d4f16a6522ab99b3c1a6bef3ca2ec81e26a9)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/sql/internal/columnNodes.scala | 13 ++++++++++-
.../apache/spark/sql/DataFrameSubquerySuite.scala | 25 ++++++++++++++++++++++
.../apache/spark/sql/DataFrameSubquerySuite.scala | 21 ++++++++++++++++++
.../spark/sql/internal/ColumnNodeSuite.scala | 7 ++++++
4 files changed, 65 insertions(+), 1 deletion(-)
diff --git
a/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala
b/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala
index 4a7339165cb5..b7c5c85caeea 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala
@@ -666,10 +666,21 @@ case class SubqueryExpression(
subqueryType: SubqueryType,
override val origin: Origin = CurrentOrigin.get)
extends ColumnNode {
+ override private[internal] def normalize(): SubqueryExpression = copy(
+ subqueryType = subqueryType match {
+ case SubqueryType.IN(values) =>
SubqueryType.IN(ColumnNode.normalize(values))
+ case other => other
+ },
+ origin = NO_ORIGIN)
override def sql: String = subqueryType match {
case SubqueryType.SCALAR => s"($ds)"
case SubqueryType.IN(values) => s"(${values.map(_.sql).mkString(",")}) IN
($ds)"
case _ => s"$subqueryType ($ds)"
}
- override private[internal] def children: Seq[ColumnNodeLike] = Seq.empty
+ // The subquery plan `ds` is intentionally not a child: it is a DataFrame
plan, not a
+ // ColumnNode, and the converters handle it separately. Only IN's values are
child nodes.
+ override private[internal] def children: Seq[ColumnNodeLike] = subqueryType
match {
+ case SubqueryType.IN(values) => values
+ case _ => Seq.empty
+ }
}
diff --git
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
index 9e93cd4442d5..61342faac51e 100644
---
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
+++
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
@@ -339,6 +339,31 @@ class DataFrameSubquerySuite extends QueryTest with
RemoteSparkSession {
sql("select * from l where l.a in (select c from r) and l.a > 2 and l.b
is not null"))
}
+ test("SPARK-58165: scalar subquery nested in IN subquery values") {
+ // A scalar subquery nested inside the IN subquery's values. The nested
SubqueryExpression
+ // must still contribute its plan to WithRelations; otherwise the server
fails with
+ // "Missing relation in WithRelations". Regression guard for the Connect
path.
+ checkAnswer(
+ spark
+ .table("l")
+
.where(spark.table("r").select(max($"c")).scalar().isin(spark.table("r").select($"c"))),
+ sql("select * from l where (select max(c) from r) in (select c from r)"))
+
+ // Correlated variant: the nested scalar subquery references the outer row
via outer(), so it
+ // is recomputed per row instead of being a constant.
+ checkAnswer(
+ spark
+ .table("l")
+ .where(
+ spark
+ .table("r")
+ .where($"c" === $"a".outer())
+ .select(max($"d"))
+ .scalar()
+ .isin(spark.table("r").select($"c"))),
+ sql("select * from l where (select max(d) from r where c = a) in (select
c from r)"))
+ }
+
test("IN predicate subquery with struct") {
withTempView("ll", "rr") {
spark.table("l").select($"*", struct("a",
"b").alias("sab")).createOrReplaceTempView("ll")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
index 6e900c3c8b5f..b7bd9da027ea 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
@@ -399,6 +399,27 @@ class DataFrameSubquerySuite extends QueryTest with
SharedSparkSession {
sql("select * from l where l.a in (select c from r) and l.a > 2 and l.b
is not null"))
}
+ test("SPARK-58165: scalar subquery nested in IN subquery values") {
+ // Classic parity for a scalar subquery nested inside the IN subquery's
values. Classic
+ // embeds the subquery plan directly, so it is unaffected by the Connect
WithRelations bug;
+ // this guards against a future regression and keeps the two suites in
sync.
+ checkAnswer(
+ spark
+ .table("l")
+
.where(spark.table("r").select(max($"c")).scalar().isin(spark.table("r").select($"c"))),
+ sql("select * from l where (select max(c) from r) in (select c from r)"))
+
+ // Correlated variant: the nested scalar subquery references the outer row
via outer(), so it
+ // is recomputed per row instead of being a constant.
+ checkAnswer(
+ spark
+ .table("l")
+ .where(
+ spark.table("r").where($"c" ===
$"a".outer()).select(max($"d")).scalar()
+ .isin(spark.table("r").select($"c"))),
+ sql("select * from l where (select max(d) from r where c = a) in (select
c from r)"))
+ }
+
test("IN predicate subquery with struct") {
withTempView("ll", "rr") {
spark.table("l").select($"*", struct("a",
"b").alias("sab")).createOrReplaceTempView("ll")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/internal/ColumnNodeSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/internal/ColumnNodeSuite.scala
index 8376433e4427..c32c57fc1443 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/internal/ColumnNodeSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/internal/ColumnNodeSuite.scala
@@ -187,6 +187,13 @@ class ColumnNodeSuite extends SparkFunSuite {
testNormalization(InvokeInlineUserDefinedFunction(
simpleUdf,
Seq(attribute("a", 2))))
+ // `ds` is not part of the normalization contract (it is a plan, not a
ColumnNode, and is
+ // compared by reference), so a stable null reference is enough to
exercise origin stripping
+ // for SCALAR/EXISTS and the recursive normalization of IN(values).
+ testNormalization(SubqueryExpression(null, SubqueryType.SCALAR))
+ testNormalization(SubqueryExpression(null, SubqueryType.EXISTS))
+ testNormalization(
+ SubqueryExpression(null, SubqueryType.IN(Seq(attribute("a", 2),
attribute("b", 3)))))
}
private def testNormalization(generate: => ColumnNode): Unit = {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
