You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/07/22 19:41:54 UTC
(spark) branch branch-4.2 updated: [SPARK-58165][SQL][CONNECT] Fix SubqueryExpression dropping nested subquery plans in IN values
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new c19664f2b622 [SPARK-58165][SQL][CONNECT] Fix SubqueryExpression dropping nested subquery plans in IN values
c19664f2b622 is described below
commit c19664f2b622f2a0a87f2ec23f5ed2529b035352
Author: YangJie <ya...@baidu.com>
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 <ya...@baidu.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 5994d4f16a6522ab99b3c1a6bef3ca2ec81e26a9)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../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 79ea389bf5f7..af6745af361f 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 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org