This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a6729073dc02 [SPARK-57885][SQL] Promote `EmptyRDDWithPartitions` to a
top-level class
a6729073dc02 is described below
commit a6729073dc02a3b79ad593d582f702ad57064346
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Jul 2 14:06:49 2026 -0700
[SPARK-57885][SQL] Promote `EmptyRDDWithPartitions` to a top-level class
### What changes were proposed in this pull request?
This PR promotes `EmptyRDDWithPartitions` from the `CoalesceExec` companion
object to a top-level class in `org.apache.spark.sql.execution`, and uses it in
the other places that hand-rolled an empty single-partition RDD via
`sparkContext.parallelize(...)`.
Note that `EmptyRDDWithPartitions extends RDD[InternalRow]`. So, this is
under `sql` module unlike `EmptyRDD` in `core` module.
### Why are the changes needed?
To deduplicate the "empty RDD with one partition" idiom with a single
purpose-named class consistently.
```scala
-
sparkSession.sparkContext.parallelize(Array.empty[InternalRow].toImmutableArraySeq,
1)
+ new EmptyRDDWithPartitions(sparkSession.sparkContext, 1)
```
https://github.com/apache/spark/blob/17e7e1479eadcdc3a97e06e3b69127576df17412/core/src/main/scala/org/apache/spark/rdd/EmptyRDD.scala#L28
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #56967 from dongjoon-hyun/SPARK-57885.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../sql/execution/EmptyRDDWithPartitions.scala | 42 ++++++++++++++++++++++
.../sql/execution/basicPhysicalOperators.scala | 21 ++---------
.../execution/datasources/FileFormatWriter.scala | 4 +--
.../sql/execution/datasources/WriteFiles.scala | 5 ++-
.../execution/datasources/v2/BatchScanExec.scala | 3 +-
.../datasources/v2/WriteToDataSourceV2Exec.scala | 4 +--
6 files changed, 52 insertions(+), 27 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/EmptyRDDWithPartitions.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/EmptyRDDWithPartitions.scala
new file mode 100644
index 000000000000..58cdb73c98fd
--- /dev/null
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/EmptyRDDWithPartitions.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import org.apache.spark.{Partition, SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+
+/**
+ * A simple RDD with no data, but with the given number of partitions. Used by
operators that
+ * must produce at least one (empty) partition even when their input has none,
e.g. to honor a
+ * reported `SinglePartition` output partitioning, or to set up a single write
task that writes
+ * the metadata for an empty write.
+ */
+class EmptyRDDWithPartitions(
+ @transient private val sc: SparkContext,
+ numPartitions: Int) extends RDD[InternalRow](sc, Nil) {
+
+ override def getPartitions: Array[Partition] =
+ Array.tabulate(numPartitions)(i => EmptyPartition(i))
+
+ override def compute(split: Partition, context: TaskContext):
Iterator[InternalRow] = {
+ Iterator.empty
+ }
+}
+
+private[execution] case class EmptyPartition(index: Int) extends Partition
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
index 8c96f1ff9579..40c27b9e77f5 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala
@@ -24,7 +24,7 @@ import scala.collection.mutable
import scala.concurrent.ExecutionContext
import scala.concurrent.duration.Duration
-import org.apache.spark.{InterruptibleIterator, Partition, SparkContext,
SparkException, TaskContext}
+import org.apache.spark.{InterruptibleIterator, SparkException, TaskContext}
import org.apache.spark.internal.LogKeys
import org.apache.spark.rdd.{EmptyRDD, PartitionwiseSampledRDD, RDD,
SQLPartitioningAwareUnionRDD, UnionPartition, UnionRDD}
import org.apache.spark.sql.catalyst.InternalRow
@@ -1243,7 +1243,7 @@ case class CoalesceExec(numPartitions: Int, child:
SparkPlan) extends UnaryExecN
if (numPartitions == 1 && rdd.getNumPartitions < 1) {
// Make sure we don't output an RDD with 0 partitions, when claiming
that we have a
// `SinglePartition`.
- new CoalesceExec.EmptyRDDWithPartitions(sparkContext, numPartitions)
+ new EmptyRDDWithPartitions(sparkContext, numPartitions)
} else {
rdd.coalesce(numPartitions, shuffle = false)
}
@@ -1253,23 +1253,6 @@ case class CoalesceExec(numPartitions: Int, child:
SparkPlan) extends UnaryExecN
copy(child = newChild)
}
-object CoalesceExec {
- /** A simple RDD with no data, but with the given number of partitions. */
- class EmptyRDDWithPartitions(
- @transient private val sc: SparkContext,
- numPartitions: Int) extends RDD[InternalRow](sc, Nil) {
-
- override def getPartitions: Array[Partition] =
- Array.tabulate(numPartitions)(i => EmptyPartition(i))
-
- override def compute(split: Partition, context: TaskContext):
Iterator[InternalRow] = {
- Iterator.empty
- }
- }
-
- case class EmptyPartition(index: Int) extends Partition
-}
-
/**
* Parent class for different types of subquery plans
*/
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index c75c8f046214..de642a3e850a 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -37,7 +37,7 @@ import
org.apache.spark.sql.catalyst.expressions.BindReferences.bindReferences
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils}
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.write.WriterCommitMessage
-import org.apache.spark.sql.execution.{ProjectExec, SortExec, SparkPlan,
SQLExecution, UnsafeExternalRowSorter}
+import org.apache.spark.sql.execution.{EmptyRDDWithPartitions, ProjectExec,
SortExec, SparkPlan, SQLExecution, UnsafeExternalRowSorter}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
import org.apache.spark.util.{SerializableConfiguration, Utils}
import org.apache.spark.util.ArrayImplicits._
@@ -248,7 +248,7 @@ object FileFormatWriter extends Logging {
// SPARK-23271 If we are attempting to write a zero partition rdd,
create a dummy single
// partition rdd to make sure we at least set up one write task to write
the metadata.
val rddWithNonEmptyPartitions = if (rdd.partitions.length == 0) {
-
sparkSession.sparkContext.parallelize(Array.empty[InternalRow].toImmutableArraySeq,
1)
+ new EmptyRDDWithPartitions(sparkSession.sparkContext, 1)
} else {
rdd
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
index c6c34b7fcea3..e2b75c7e8119 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
@@ -28,9 +28,8 @@ import
org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
import org.apache.spark.sql.connector.write.WriterCommitMessage
-import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.execution.{EmptyRDDWithPartitions, SparkPlan,
UnaryExecNode}
import
org.apache.spark.sql.execution.datasources.FileFormatWriter.ConcurrentOutputWriterSpec
-import org.apache.spark.util.ArrayImplicits._
/**
* The write files spec holds all information of [[V1WriteCommand]] if its
provider is
@@ -82,7 +81,7 @@ case class WriteFilesExec(
// SPARK-23271 If we are attempting to write a zero partition rdd, create
a dummy single
// partition rdd to make sure we at least set up one write task to write
the metadata.
val rddWithNonEmptyPartitions = if (rdd.partitions.length == 0) {
-
session.sparkContext.parallelize(Array.empty[InternalRow].toImmutableArraySeq,
1)
+ new EmptyRDDWithPartitions(session.sparkContext, 1)
} else {
rdd
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
index 638d8bc415f9..e7a7dc814563 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala
@@ -29,6 +29,7 @@ import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE
import org.apache.spark.sql.connector.write.RowLevelOperationTable
+import org.apache.spark.sql.execution.EmptyRDDWithPartitions
import org.apache.spark.sql.execution.metric.{SQLLastAttemptMetrics,
SQLMetric, SQLMetrics}
import org.apache.spark.util.ArrayImplicits._
@@ -90,7 +91,7 @@ case class BatchScanExec(
override lazy val inputRDD: RDD[InternalRow] = {
val rdd = if (filteredPartitions.isEmpty && outputPartitioning ==
SinglePartition) {
// return an empty RDD with 1 partition if dynamic filtering removed the
only split
- sparkContext.parallelize(Array.empty[InternalRow].toImmutableArraySeq, 1)
+ new EmptyRDDWithPartitions(sparkContext, 1)
} else {
new DataSourceRDD(
sparkContext, filteredPartitions, readerFactory, supportsColumnar,
customMetrics)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
index a1a4ca6196eb..d280076622b0 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -36,7 +36,7 @@ import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter,
DataWriterFactory, DeleteSummaryImpl, DeltaWrite, DeltaWriter,
InsertSummaryImpl, MergeSummaryImpl, PhysicalWriteInfoImpl, RowLevelOperation,
RowLevelOperationTable, UpdateSummaryImpl, Write, WriterCommitMessage,
WriteSummary}
import org.apache.spark.sql.connector.write.RowLevelOperation.Command._
import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
-import org.apache.spark.sql.execution.{QueryExecution, SparkPlan,
SQLExecution, UnaryExecNode}
+import org.apache.spark.sql.execution.{EmptyRDDWithPartitions, QueryExecution,
SparkPlan, SQLExecution, UnaryExecNode}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.metric.{CustomMetrics,
SQLLastAttemptMetric, SQLLastAttemptMetrics, SQLMetric, SQLMetrics}
import org.apache.spark.sql.types.StructType
@@ -606,7 +606,7 @@ trait V2TableWriteExec
// SPARK-23271 If we are attempting to write a zero partition rdd,
create a dummy single
// partition rdd to make sure we at least set up one write task to write
the metadata.
if (tempRdd.partitions.length == 0) {
- sparkContext.parallelize(Array.empty[InternalRow].toImmutableArraySeq,
1)
+ new EmptyRDDWithPartitions(sparkContext, 1)
} else {
tempRdd
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
