This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 7f923c25bbb8 [SPARK-59884][SQL] Mark SparkPlan's columnar execution
cache transient
7f923c25bbb8 is described below
commit 7f923c25bbb8c806c156b461f4645d9a15116779
Author: Chao Sun <[email protected]>
AuthorDate: Tue Oct 6 12:27:02 2026 -0700
[SPARK-59884][SQL] Mark SparkPlan's columnar execution cache transient
### What changes were proposed in this pull request?
Mark `SparkPlan.executeColumnarRDD` transient, matching the row execution
cache, and correct the `executeColumnar()` Scaladoc's references to
`doExecuteColumnar()`.
Add two SPARK-59884 regression tests:
- A direct plan serialization round trip with an initialized columnar cache
containing a nonserializable sentinel, also checking driver-side cache reuse.
- A vectorized Parquet aggregation across a shuffle. The test checks the
query result and verifies that the job used the scan's columnar wrapper RDD. It
then serializes the downstream result-stage RDD/function tuple, asserts that
serialization visits the upstream scan plan, and checks that the wrapper RDD is
absent.
JIRA: [SPARK-59884](https://issues.apache.org/jira/browse/SPARK-59884)
### Why are the changes needed?
Plans captured by downstream operators can retain upstream plans across
shuffle boundaries. An initialized `executeColumnarRDD` cache can therefore
serialize an RDD that is outside the downstream stage's input lineage. Making
this driver-side cache transient removes that reference while preserving
driver-side reuse.
For the Parquet regression, this excludes only the columnar wrapper RDD.
`FileSourceScanExec.inputRDD` and other scan-owned caches remain separate
retention paths. `AQEShuffleReadExec.shuffleRDD`, the broadcast execution
cache, and whole-plan capture are also outside this PR's scope.
The cache was introduced by [SPARK-48195 /
#48037](https://github.com/apache/spark/pull/48037) in
[d2e8c1c](https://github.com/apache/spark/commit/d2e8c1cb60e34a1c7e92374c07d682aa5ca79145),
first included in Spark 4.0.0. Branches 4.0 through 4.3 contain the same
nontransient field and are candidates for backporting.
### Does this PR introduce _any_ user-facing change?
Yes. It avoids serializing an upstream columnar RDD through the base plan
execution cache when that RDD is not needed by a downstream stage. Query
results and public APIs are unchanged. It does not make a nonserializable RDD
usable as a stage's actual input or eliminate other plan-owned RDD references.
### How was this patch tested?
- Full `SparkPlanSuite`: 14 tests passed after compiling the changed files
against CI-built Spark 5.0 dependencies.
- Both SPARK-59884 regressions failed as expected when only the
`executeColumnarRDD` transient annotation was removed.
- Scoped Scalastyle and `git diff --check` passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-6).
Closes #59146 from sunchao/codex/columnar-rdd-transient-upstream.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 895c75f0032a93ac29dff18d6720783ba1f22a5c)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../org/apache/spark/sql/execution/SparkPlan.scala | 5 +-
.../spark/sql/execution/SparkPlanSuite.scala | 81 +++++++++++++++++++++-
2 files changed, 82 insertions(+), 4 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
index 4f8f66eb5969..6912f44a3452 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
@@ -218,15 +218,16 @@ abstract class SparkPlan extends QueryPlan[SparkPlan]
with Logging with Serializ
executeBroadcastBcast.get.asInstanceOf[broadcast.Broadcast[T]]
}
+ @transient
private val executeColumnarRDD = LazyTry {
doExecuteColumnar()
}
/**
- * Returns the result of this query as an RDD[ColumnarBatch] by delegating
to `doColumnarExecute`
+ * Returns the result of this query as an RDD[ColumnarBatch] by delegating
to `doExecuteColumnar`
* after preparations.
*
- * Concrete implementations of SparkPlan should override `doColumnarExecute`
if `supportsColumnar`
+ * Concrete implementations of SparkPlan should override `doExecuteColumnar`
if `supportsColumnar`
* returns true.
*/
final def executeColumnar(): RDD[ColumnarBatch] = executeQuery {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
index c0f5ad61d72e..45a0f4098bc5 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
@@ -17,14 +17,16 @@
package org.apache.spark.sql.execution
+import java.io.{ObjectOutputStream, OutputStream}
import java.lang.management.ManagementFactory
import java.util.concurrent.{CountDownLatch, TimeUnit}
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration.Duration
-import org.apache.spark.{SparkEnv, SparkException,
SparkUnsupportedOperationException}
-import org.apache.spark.rdd.RDD
+import org.apache.spark.{SparkEnv, SparkException,
SparkUnsupportedOperationException, TaskContext}
+import org.apache.spark.rdd.{EmptyRDD, RDD}
+import org.apache.spark.sql.{QueryTest, Row}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{
Attribute, AttributeReference, Expression, ExprId, Literal}
@@ -32,6 +34,7 @@ import
org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCo
import org.apache.spark.sql.catalyst.plans.logical.Deduplicate
import org.apache.spark.sql.catalyst.trees.LeafLike
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{DataType, IntegerType, StringType}
@@ -40,6 +43,68 @@ import org.apache.spark.util.ThreadUtils
class SparkPlanSuite extends SharedSparkSession {
+ test("SPARK-59884: columnar execution cache is not serialized with a plan") {
+ val source = ColumnarCacheSerializationSource(Seq(AttributeReference("id",
IntegerType)()))
+ val rdd = source.executeColumnar()
+ assert(source.executeColumnar() eq rdd)
+ val serializer = SparkEnv.get.closureSerializer.newInstance()
+ // Plans can be captured by task closures, but their execution caches
belong to the driver.
+
serializer.deserialize[ColumnarToRowExec](serializer.serialize(ColumnarToRowExec(source)))
+ }
+
+ test("SPARK-59884: columnar execution cache does not cross a shuffle in the
task binary") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ withTempPath { path =>
+ spark.range(0, 8, 1, 2).selectExpr("id % 2 AS
key").write.parquet(path.getAbsolutePath)
+ val query =
spark.read.parquet(path.getAbsolutePath).groupBy("key").count()
+ QueryTest.checkAnswer(query, Seq(Row(0L, 4L), Row(1L, 4L)), checkToRDD
= false)
+
+ val plan = query.queryExecution.executedPlan
+ val exchanges = plan.collect { case exchange: ShuffleExchangeExec =>
exchange }
+ assert(exchanges.size == 1)
+ val scan = exchanges.head.child.collectFirst {
+ case scan: FileSourceScanExec => scan
+ }.get
+ assert(scan.supportsColumnar)
+ val columnarRDD = scan.executeColumnar()
+ def lineage(rdd: RDD[_]): Seq[RDD[_]] = {
+ rdd +: rdd.dependencies.flatMap(d => lineage(d.rdd))
+ }
+ assert(lineage(exchanges.head.inputRDD).exists(_ eq columnarRDD),
+ "the job must create the columnar RDD, or the test exercises
nothing")
+ val stageRDD = plan.execute()
+
+ var serializedScan = false
+ var serializedColumnarRDD = false
+ val out = new ObjectOutputStream(OutputStream.nullOutputStream()) {
+ enableReplaceObject(true)
+ override protected def replaceObject(obj: AnyRef): AnyRef = {
+ serializedScan ||= obj eq scan
+ serializedColumnarRDD ||= obj eq columnarRDD
+ obj
+ }
+ }
+ try {
+ // DAGScheduler serializes the complete (stage.rdd, stage.func)
tuple for a result task.
+ // The scan's columnar wrapper RDD is outside the downstream stage's
input lineage.
+ // The captured plan must not reintroduce it through its columnar
execution cache.
+ // FileSourceScanExec.inputRDD is a separate cache and is not
covered by this fix.
+ val taskFunc = (_: TaskContext, rows: Iterator[InternalRow]) =>
rows.size
+ out.writeObject((stageRDD, taskFunc))
+ } finally {
+ out.close()
+ }
+ assert(serializedScan, "the downstream stage no longer captures the
upstream plan")
+ assert(!serializedColumnarRDD, "upstream columnar RDD was captured
across the shuffle")
+ }
+ }
+ }
+
test("SPARK-21619 execution of a canonicalized plan should fail") {
val plan = spark.range(10).queryExecution.executedPlan.canonicalized
@@ -268,6 +333,18 @@ case class ColumnarOp(child: SparkPlan) extends
UnaryExecNode {
copy(child = newChild)
}
+case class ColumnarCacheSerializationSource(output: Seq[Attribute]) extends
LeafExecNode {
+ override val supportsColumnar: Boolean = true
+
+ override protected def doExecuteColumnar(): RDD[ColumnarBatch] = {
+ new EmptyRDD[ColumnarBatch](sparkContext) {
+ val driverOnlyState: AnyRef = new Object
+ }
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = throw
SparkUnsupportedOperationException()
+}
+
private case class TestSubqueryExec(child: SparkPlan) extends BaseSubqueryExec
{
override def name: String = "TestSubqueryExec"
override def children: Seq[SparkPlan] = Seq(child)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
