You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ur...@apache.org on 2026/10/22 22:05:30 UTC
(spark) branch branch-4.x updated: [SPARK-60096][4.X][CONNECT][SQL] Raise a user-facing error instead of INTERNAL_ERROR for an unresolved DataFrame subquery plan id
This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 57b39f428167 [SPARK-60096][4.X][CONNECT][SQL] Raise a user-facing error instead of INTERNAL_ERROR for an unresolved DataFrame subquery plan id
57b39f428167 is described below
commit 57b39f42816708f48c30ed4dcdab93bfb3644223
Author: Ivan <77...@users.noreply.github.com>
AuthorDate: Thu Oct 22 22:05:16 2026 +0000
[SPARK-60096][4.X][CONNECT][SQL] Raise a user-facing error instead of INTERNAL_ERROR for an unresolved DataFrame subquery plan id
### What changes were proposed in this pull request?
Backport for branch-4.x. Original PR https://github.com/apache/spark/pull/59310.
Spark Connect represents DataFrame subqueries such as `col.isin(df)`, `df.scalar()`, `df.exists()`, and table arguments as `UnresolvedPlanId` expressions (`UnresolvedInSubqueryPlanId`, `UnresolvedScalarSubqueryPlanId`,
`UnresolvedExistsPlanId`, `UnresolvedTableArgPlanId`). These expressions point to the subquery plan by plan id. Only `SparkConnectPlanner.transformWithRelations` replaces them with the actual plan, using the enclosing `WithRelations`
node. No analyzer rule resolves them.
This PR:
- Adds a new sub-condition `CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED`. It inherits SQLSTATE `56K00` from `CONNECT_INVALID_PLAN`, the same classification `transformWithRelations` already uses for a missing plan id inside a `WithRelations`.
- Adds an `UnresolvedPlanId` case to the expression-level checks in `CheckAnalysis.checkAnalysis0`. The case throws the new `AnalysisException`, which carries the plan id and the expression's origin.
- Adds `QueryCompilationErrors.subqueryPlanNotAttached`.
### Why are the changes needed?
If the client never attaches the referenced plan (for example, when the root relation is not wrapped in a `WithRelations` that contains it), the `UnresolvedPlanId` survives analysis. CheckAnalysis has no specific case for it, so it reaches
the generic unresolved-operator check and fails with an internal error:
[INTERNAL_ERROR] Found the unresolved operator: ... SQLSTATE: XX000
This means the Connect client sent an invalid plan. It can come from a non-conforming third-party client, or from the Scala Connect client itself: the typed `Dataset.select[U1](TypedColumn)` and `Dataset.joinWith` do not collect subquery expressions, so they send plan ids without `WithRelations`. It is not a Spark server bug, so it should be reported as an invalid Spark Connect plan rather than `INTERNAL_ERROR`.
### Does this PR introduce _any_ user-facing change?
Yes, but only for queries that already fail. A Spark Connect DataFrame subquery whose plan id is never attached used to fail with `[INTERNAL_ERROR] ... SQLSTATE: XX000`. It now fails with an `AnalysisException`:
[CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED] The Spark Connect plan is invalid. The subquery plan referenced by plan id 42 is not attached to the request. SQLSTATE: 56K00
Valid queries are not affected.
### How was this patch tested?
Added new tests to `SparkConnectProtoSuite`.
Ran locally:
- `build/sbt "connect/testOnly *SparkConnectProtoSuite -- -z SPARK-60096"`: passed
- `build/sbt "core/testOnly *SparkThrowableSuite"`: 38 passed (validates `error-conditions.json`)
### Was this patch authored or co-authored using generative AI tooling?
Yes
Closes #59687 from sadikovi/SPARK-59988_4.x.
Authored-by: Ivan <77...@users.noreply.github.com>
Signed-off-by: Uros Bojanic <22...@users.noreply.github.com>
---
.../src/main/resources/error/error-conditions.json | 5 +
.../sql/catalyst/analysis/CheckAnalysis.scala | 8 +
.../spark/sql/errors/QueryCompilationErrors.scala | 10 +-
.../sql/connect/planner/SparkConnectPlanner.scala | 96 ++++++---
.../connect/planner/SparkConnectProtoSuite.scala | 222 ++++++++++++++++++++-
5 files changed, 311 insertions(+), 30 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 30f785bbed11..71ef75a4b712 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1663,6 +1663,11 @@
"Struct literal is not set."
]
},
+ "SUBQUERY_PLAN_NOT_ATTACHED" : {
+ "message" : [
+ "The subquery plan referenced by plan id <planId> is not attached to the request."
+ ]
+ },
"UDT_JVM_CLASS_NOT_UDT" : {
"message" : [
"The 'jvm_class' <jvmClass> in the UDT is not a subclass of UserDefinedType."
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index f567268d964e..a6b3185beda6 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -646,6 +646,14 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
s"Cannot resolve the runtime replaceable expression ${toSQLExpr(e)}. " +
s"The replacement is unresolved: ${toSQLExpr(e.replacement)}.")
+ // Only the Spark Connect planner creates UnresolvedPlanId (a DataFrame subquery by plan
+ // id), and only its WithRelations transform resolves it. A survivor usually means the
+ // client never attached the referenced plan to the request, so report an invalid Spark
+ // Connect plan. If the planner analyzes the plan before substituting an attached plan
+ // id, it rethrows this error as an INTERNAL_ERROR.
+ case u: UnresolvedPlanId =>
+ throw QueryCompilationErrors.subqueryPlanNotAttached(u)
+
// `Grouping` and `GroupingID` are considered as of having lower priority than the other
// nodes which cause errors.
case g: Grouping =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 4b0a5cfc1222..f91fde372a20 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -24,7 +24,7 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.{SPARK_DOC_ROOT, SparkException, SparkIllegalArgumentException, SparkThrowable, SparkUnsupportedOperationException}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.{ExtendedAnalysisException, FunctionIdentifier, InternalRow, QualifiedTableName, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, FunctionAlreadyExistsException, NamedRelation, NamespaceAlreadyExistsException, NoSuchFunctionException, NoSuchNamespaceException, NoSuchPartitionException, NoSuchTableException, Star, TableAlreadyExistsException, UnresolvedRegex}
+import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, FunctionAlreadyExistsException, NamedRelation, NamespaceAlreadyExistsException, NoSuchFunctionException, NoSuchNamespaceException, NoSuchPartitionException, NoSuchTableException, Star, TableAlreadyExistsException, UnresolvedPlanId, UnresolvedRegex}
import org.apache.spark.sql.catalyst.catalog.{CatalogTable, InvalidUDFClassException}
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, AttributeSet, CreateMap, CreateStruct, Expression, GroupingID, NamedExpression, SortOrder, SpecifiedWindowFrame, WindowFrame, WindowFunction, WindowSpecDefinition}
@@ -4878,6 +4878,14 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
)
}
+ def subqueryPlanNotAttached(u: UnresolvedPlanId): Throwable = {
+ new AnalysisException(
+ errorClass = "CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED",
+ messageParameters = Map("planId" -> u.planId.toString),
+ origin = u.origin
+ )
+ }
+
def ambiguousColumnReferences(e: Expression): Throwable = {
new AnalysisException(
errorClass = "AMBIGUOUS_COLUMN_REFERENCE",
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index f18e953a6f7d..a387409d532e 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -1310,18 +1310,24 @@ class SparkConnectPlanner(
}
val child = transformRelation(rel.getInput)
- try {
- // Try the eager analysis first.
- Dataset
- .ofRows(session, child)
- .withColumnsRenamed(colNames, newColNames)
- .logicalPlan
- } catch {
- case _: AnalysisException | _: SparkException =>
- Project(
- Seq(UnresolvedStarWithColumnsRenames(existingNames = colNames, newNames = newColNames)),
- child)
+ // Try the eager analysis first.
+ val eagerPlan = if (hasUnsubstitutedPlanId(child)) {
+ None
+ } else {
+ try {
+ Some(
+ Dataset
+ .ofRows(session, child)
+ .withColumnsRenamed(colNames, newColNames)
+ .logicalPlan)
+ } catch {
+ case _: AnalysisException | _: SparkException => None
+ }
}
+ eagerPlan.getOrElse(
+ Project(
+ Seq(UnresolvedStarWithColumnsRenames(existingNames = colNames, newNames = newColNames)),
+ child))
}
private def transformWithColumns(rel: proto.WithColumns): LogicalPlan = {
@@ -1341,22 +1347,39 @@ class SparkConnectPlanner(
}.unzip3
val child = transformRelation(rel.getInput)
- try {
- // Try the eager analysis first.
- Dataset
- .ofRows(session, child)
- .withColumns(colNames, exprs.map(expr => Column(expr)), metadata)
- .logicalPlan
- } catch {
- case _: AnalysisException | _: SparkException =>
- Project(
- Seq(
- UnresolvedStarWithColumns(
- colNames = colNames,
- exprs = exprs,
- explicitMetadata = Some(metadata))),
- child)
+ // Try the eager analysis first.
+ val eagerPlan = if (hasUnsubstitutedPlanId(child, exprs)) {
+ None
+ } else {
+ try {
+ Some(
+ Dataset
+ .ofRows(session, child)
+ .withColumns(colNames, exprs.map(expr => Column(expr)), metadata)
+ .logicalPlan)
+ } catch {
+ case _: AnalysisException | _: SparkException => None
+ }
}
+ eagerPlan.getOrElse(
+ Project(
+ Seq(
+ UnresolvedStarWithColumns(
+ colNames = colNames,
+ exprs = exprs,
+ explicitMetadata = Some(metadata))),
+ child))
+ }
+
+ /**
+ * Returns true if the plan or expressions still reference a subquery plan by plan id, which
+ * [[transformWithRelations]] substitutes only after the root relation is transformed. Eager
+ * analysis cannot succeed before that, and its failure would be posted as a failed SQL
+ * execution, so callers should build a lazy plan instead.
+ */
+ private def hasUnsubstitutedPlanId(plan: LogicalPlan, exprs: Seq[Expression] = Nil): Boolean = {
+ plan.containsPattern(TreePattern.UNRESOLVED_PLAN_ID) ||
+ exprs.exists(_.containsPattern(TreePattern.UNRESOLVED_PLAN_ID))
}
private def transformWithWatermark(rel: proto.WithWatermark): LogicalPlan = {
@@ -4596,8 +4619,27 @@ class SparkConnectPlanner(
if (isValidSQLWithRefs(getWithRelations)) {
transformSqlWithRefs(getWithRelations)
} else {
+ val referencePlanIds = getWithRelations.getReferencesList.asScala.collect {
+ case ref if ref.hasCommon && ref.getCommon.hasPlanId => ref.getCommon.getPlanId
+ }.toSet
+
// Wrap the plan to keep the original planId.
- val plan = Project(Seq(UnresolvedStar(None)), transformRelation(getWithRelations.getRoot))
+ val plan =
+ try {
+ Project(Seq(UnresolvedStar(None)), transformRelation(getWithRelations.getRoot))
+ } catch {
+ // Some transforms analyze the root eagerly, before the referenced plans are substituted
+ // below. A plan id that is attached here is then a server-side issue, not a client error.
+ case e: AnalysisException
+ if e.getCondition == "CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED" &&
+ e.getMessageParameters.asScala.get("planId").exists { id =>
+ referencePlanIds.contains(id.toLong)
+ } =>
+ throw SparkException.internalError(
+ s"The subquery plan referenced by plan id ${e.getMessageParameters.get("planId")} " +
+ "is attached in WithRelations but was analyzed before it was substituted.",
+ e)
+ }
val relations = getWithRelations.getReferencesList.asScala.map { ref =>
if (ref.hasCommon && ref.getCommon.hasPlanId) {
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
index 0f23b50e17f1..087f09a63296 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectProtoSuite.scala
@@ -17,18 +17,20 @@
package org.apache.spark.sql.connect.planner
import java.nio.file.{Files, Paths}
+import java.util.concurrent.ConcurrentLinkedQueue
import scala.jdk.CollectionConverters._
import com.google.protobuf.ByteString
-import org.apache.spark.{SparkClassNotFoundException, SparkIllegalArgumentException}
+import org.apache.spark.{SparkClassNotFoundException, SparkException, SparkIllegalArgumentException}
import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.Expression
import org.apache.spark.connect.proto.Join.JoinType
+import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
import org.apache.spark.sql.{AnalysisException, Column, Observation, Row, SaveMode}
import org.apache.spark.sql.catalyst.analysis
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow, UnresolvedInSubqueryPlanId, UnsafeProjection}
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftAnti, LeftOuter, LeftSemi, PlanTest, RightOuter}
import org.apache.spark.sql.catalyst.plans.logical.{CollectMetrics, Deduplicate, DeduplicateWithinWatermark, Distinct, LocalRelation, LogicalPlan}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -43,6 +45,7 @@ import org.apache.spark.sql.connect.dsl.plans._
import org.apache.spark.sql.connector.catalog.{Column => ColumnV2, Identifier, InMemoryTableCatalog, TableCatalog}
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
import org.apache.spark.sql.execution.arrow.ArrowConverters
+import org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionStart
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -976,6 +979,221 @@ class SparkConnectProtoSuite extends PlanTest with SparkConnectPlanTest {
sparkTestRelation.withColumns(Map("id" -> lit(1024), "col_not_exist" -> lit(2048))))
}
+ test("SPARK-60096: IN-subquery with an unattached plan id is an invalid Connect plan") {
+ // A Connect `col.isin(<subquery>)` inside `when(...)` whose plan id is not attached by any
+ // WithRelations leaves an UnresolvedInSubqueryPlanId in the plan.
+ val unattachedPlanId = 42L
+ val inSubquery = Expression
+ .newBuilder()
+ .setSubqueryExpression(
+ proto.SubqueryExpression
+ .newBuilder()
+ .setPlanId(unattachedPlanId)
+ .setSubqueryType(proto.SubqueryExpression.SubqueryType.SUBQUERY_TYPE_IN)
+ .addInSubqueryValues("id".protoAttr))
+ .build()
+ val caseWhen = Expression
+ .newBuilder()
+ .setUnresolvedFunction(
+ Expression.UnresolvedFunction
+ .newBuilder()
+ .setFunctionName("when")
+ .addArguments(inSubquery)
+ .addArguments(Expression.newBuilder().setLiteral(toLiteralProto(1)).build())
+ .addArguments(Expression.newBuilder().setLiteral(toLiteralProto(0)).build()))
+ .build()
+ val connectPlan = connectTestRelation.select(caseWhen.as("x"))
+
+ val transformed = transform(connectPlan)
+
+ // The planner leaves an UnresolvedInSubqueryPlanId in the tree: no analyzer rule resolves
+ // it (plan-id references are wired up only at proto-transform time via WithRelations).
+ val survivingInSubqueries = transformed.collect { case plan =>
+ plan.expressions.flatMap(_.collect { case u: UnresolvedInSubqueryPlanId => u })
+ }.flatten
+ assert(survivingInSubqueries.map(_.planId) == Seq(unattachedPlanId))
+ assert(!transformed.resolved)
+
+ // CheckAnalysis now rejects the surviving plan-id subquery as an invalid Spark Connect plan
+ // instead of the previous INTERNAL_ERROR / XX000 "Found the unresolved operator".
+ checkError(
+ exception = intercept[AnalysisException] {
+ analyzePlan(transformed)
+ },
+ condition = "CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED",
+ sqlState = Some("56K00"),
+ parameters = Map("planId" -> unattachedPlanId.toString))
+ }
+
+ // `id IN <plan id>`: a Connect IN-subquery whose plan is referenced by plan id.
+ private def inSubqueryProto(planId: Long): Expression = {
+ Expression
+ .newBuilder()
+ .setSubqueryExpression(
+ proto.SubqueryExpression
+ .newBuilder()
+ .setPlanId(planId)
+ .setSubqueryType(proto.SubqueryExpression.SubqueryType.SUBQUERY_TYPE_IN)
+ .addInSubqueryValues("id".protoAttr))
+ .build()
+ }
+
+ // Wraps `root` in a WithRelations that attaches a range plan with the given plan id.
+ private def withAttachedRange(root: proto.Relation, attachedPlanId: Long): proto.Relation = {
+ val reference = proto.Relation
+ .newBuilder()
+ .setCommon(proto.RelationCommon.newBuilder().setPlanId(attachedPlanId))
+ .setRange(proto.Range.newBuilder().setStart(0).setEnd(3).setStep(1))
+ .build()
+ proto.Relation
+ .newBuilder()
+ .setWithRelations(proto.WithRelations.newBuilder().setRoot(root).addReferences(reference))
+ .build()
+ }
+
+ // `rootWithSubquery` builds a root that analyzes an IN-subquery on the given plan id eagerly,
+ // before WithRelations substitutes the referenced plans.
+ private def checkEagerlyAnalyzedSubquery(rootWithSubquery: Long => proto.Relation): Unit = {
+ // The plan id is attached, so the failure is a server-side issue.
+ val attachedPlanId = 7L
+ val e = intercept[SparkException] {
+ transform(withAttachedRange(rootWithSubquery(attachedPlanId), attachedPlanId))
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains(s"plan id $attachedPlanId is attached in WithRelations"))
+ assert(
+ e.getCause.asInstanceOf[AnalysisException].getCondition ==
+ "CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED")
+
+ // The plan id is not attached, so the failure remains an invalid Spark Connect plan.
+ val unattachedPlanId = 8L
+ checkError(
+ exception = intercept[AnalysisException] {
+ transform(withAttachedRange(rootWithSubquery(unattachedPlanId), attachedPlanId))
+ },
+ condition = "CONNECT_INVALID_PLAN.SUBQUERY_PLAN_NOT_ATTACHED",
+ sqlState = Some("56K00"),
+ parameters = Map("planId" -> unattachedPlanId.toString))
+ }
+
+ test("SPARK-60096: sampleBy with an attached IN-subquery is an internal error") {
+ // sampleBy(when(id IN <plan id>, 'a').otherwise('b'), {'a': 0.5}) runs a filter, which
+ // analyzes the column eagerly.
+ checkEagerlyAnalyzedSubquery { planId =>
+ val caseWhen = Expression
+ .newBuilder()
+ .setUnresolvedFunction(
+ Expression.UnresolvedFunction
+ .newBuilder()
+ .setFunctionName("when")
+ .addArguments(inSubqueryProto(planId))
+ .addArguments(Expression.newBuilder().setLiteral(toLiteralProto("a")).build())
+ .addArguments(Expression.newBuilder().setLiteral(toLiteralProto("b")).build()))
+ .build()
+ proto.Relation
+ .newBuilder()
+ .setSampleBy(
+ proto.StatSampleBy
+ .newBuilder()
+ .setInput(connectTestRelation)
+ .setCol(caseWhen)
+ .setSeed(1)
+ .addFractions(proto.StatSampleBy.Fraction
+ .newBuilder()
+ .setStratum(toLiteralProto("a"))
+ .setFraction(0.5)))
+ .build()
+ }
+ }
+
+ test("SPARK-60096: joinAsOf with an attached IN-subquery is an internal error") {
+ // joinAsOf(right, id, rid, joinExprs = id IN <plan id>) analyzes the join condition eagerly.
+ val right = createLocalRelationProto(Seq(AttributeReference("rid", IntegerType)()), Seq.empty)
+ checkEagerlyAnalyzedSubquery { planId =>
+ proto.Relation
+ .newBuilder()
+ .setAsOfJoin(
+ proto.AsOfJoin
+ .newBuilder()
+ .setLeft(connectTestRelation)
+ .setRight(right)
+ .setLeftAsOf("id".protoAttr)
+ .setRightAsOf("rid".protoAttr)
+ .setJoinExpr(inSubqueryProto(planId))
+ .setJoinType("inner")
+ .setAllowExactMatches(true)
+ .setDirection("backward"))
+ .build()
+ }
+ }
+
+ // Transforms `rel` and returns the SQL executions started meanwhile.
+ private def sqlExecutionsStartedDuring(rel: proto.Relation): (LogicalPlan, Seq[String]) = {
+ val started = new ConcurrentLinkedQueue[String]()
+ val listener = new SparkListener {
+ override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case e: SparkListenerSQLExecutionStart => started.add(e.description)
+ case _ =>
+ }
+ }
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ val plan = transform(rel)
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ (plan, started.asScala.toSeq)
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+ }
+
+ test("SPARK-60096: withColumns with an attached subquery does not analyze eagerly") {
+ // withColumns(x = <scalar subquery on plan id>) cannot be analyzed before WithRelations
+ // substitutes the plan id. A failed eager attempt would post a failed SQL execution.
+ val planId = 9L
+ val scalarSubquery = Expression
+ .newBuilder()
+ .setSubqueryExpression(
+ proto.SubqueryExpression
+ .newBuilder()
+ .setPlanId(planId)
+ .setSubqueryType(proto.SubqueryExpression.SubqueryType.SUBQUERY_TYPE_SCALAR))
+ .build()
+ val withColumns = proto.Relation
+ .newBuilder()
+ .setWithColumns(
+ proto.WithColumns
+ .newBuilder()
+ .setInput(connectTestRelation)
+ .addAliases(Expression.Alias.newBuilder().addName("x").setExpr(scalarSubquery)))
+ .build()
+
+ val (plan, started) = sqlExecutionsStartedDuring(withAttachedRange(withColumns, planId))
+ assert(started.isEmpty)
+ val analyzed = spark.sessionState.executePlan(plan).analyzed
+ assert(analyzed.output.map(_.name) == Seq("id", "name", "x"))
+ }
+
+ test("SPARK-60096: withColumnsRenamed over an attached subquery does not analyze eagerly") {
+ // withColumnsRenamed over filter(id IN <plan id>) cannot be analyzed before WithRelations
+ // substitutes the plan id. A failed eager attempt would post a failed SQL execution.
+ val planId = 10L
+ val renamed = proto.Relation
+ .newBuilder()
+ .setWithColumnsRenamed(
+ proto.WithColumnsRenamed
+ .newBuilder()
+ .setInput(connectTestRelation.where(inSubqueryProto(planId)))
+ .addRenames(
+ proto.WithColumnsRenamed.Rename.newBuilder().setColName("id").setNewColName("id2")))
+ .build()
+
+ val (plan, started) = sqlExecutionsStartedDuring(withAttachedRange(renamed, planId))
+ assert(started.isEmpty)
+ val analyzed = spark.sessionState.executePlan(plan).analyzed
+ assert(analyzed.output.map(_.name) == Seq("id2", "name"))
+ }
+
test("Test cast") {
comparePlans(
connectTestRelation.select("id".protoAttr.cast(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org