From 2609083cf9532882f080072d64bd767cf1a20ec3 Mon Sep 17 00:00:00 2001
From: ziqi liu <ziqi.liu@databricks.com>
Date: Mon, 1 Dec 2025 16:27:40 -0800
Subject: [PATCH 1/2] fix

---
 .../org/apache/spark/sql/catalyst/plans/QueryPlan.scala  | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index 7801cd347f7d..9d19df04ec62 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -520,9 +520,12 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]]
   def subqueries: Seq[PlanType] = _subqueries()
 
   private val _subqueries = new TransientBestEffortLazyVal(() =>
-    expressions.filter(_.containsPattern(PLAN_EXPRESSION)).flatMap(_.collect {
-      case e: PlanExpression[_] => e.plan.asInstanceOf[PlanType]
-    })
+    expressions
+      .filter(_.containsPattern(PLAN_EXPRESSION))
+      .flatMap(_.collect {
+        case planExpression: PlanExpression[PlanType] =>
+          planExpression.plan
+      })
   )
 
   /**

From 30a40d53a5f1ba0d03ef89a27392cb73393d0ed6 Mon Sep 17 00:00:00 2001
From: ziqi liu <ziqi.liu@databricks.com>
Date: Mon, 1 Dec 2025 20:29:29 -0800
Subject: [PATCH 2/2] update

---
 .../spark/sql/catalyst/plans/QueryPlan.scala  | 41 ++++++++++++++++---
 1 file changed, 35 insertions(+), 6 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index 9d19df04ec62..4a60456984f5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -520,12 +520,41 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]]
   def subqueries: Seq[PlanType] = _subqueries()
 
   private val _subqueries = new TransientBestEffortLazyVal(() =>
-    expressions
-      .filter(_.containsPattern(PLAN_EXPRESSION))
-      .flatMap(_.collect {
-        case planExpression: PlanExpression[PlanType] =>
-          planExpression.plan
-      })
+    {
+      val queryPlanBaseClass = classOf[QueryPlan[_]]
+
+      def directChildQueryPlanClass(clazz: Class[_]): Option[Class[_]] = {
+        var current = clazz
+        while (current != null && queryPlanBaseClass.isAssignableFrom(current)) {
+          val parent = current.getSuperclass
+          if (parent == queryPlanBaseClass) {
+            return Some(current)
+          }
+          current = parent
+        }
+        None
+      }
+
+      val baseQueryPlanClass = directChildQueryPlanClass(getClass)
+
+      val rawSubqueries = expressions
+        .filter(_.containsPattern(PLAN_EXPRESSION))
+        .flatMap(_.collect {
+          case planExpression: PlanExpression[_] =>
+            planExpression.plan
+        })
+
+      baseQueryPlanClass match {
+        case Some(baseClass) =>
+          rawSubqueries.collect {
+            case subquery
+              if directChildQueryPlanClass(subquery.getClass).contains(baseClass) =>
+              subquery.asInstanceOf[PlanType]
+          }
+        case None =>
+          Seq.empty
+      }
+    }
   )
 
   /**