You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by as...@apache.org on 2026/05/06 05:10:17 UTC
(spark) branch master updated: [SPARK-56700][SS] Make DataStreamReader.name public
This is an automated email from the ASF dual-hosted git repository.
anishshri-db 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 0af3d42c2c35 [SPARK-56700][SS] Make DataStreamReader.name public
0af3d42c2c35 is described below
commit 0af3d42c2c35cc6e7124eb1382aa240ecdf0f711
Author: ericm-db <er...@databricks.com>
AuthorDate: Tue May 5 22:10:05 2026 -0700
[SPARK-56700][SS] Make DataStreamReader.name public
### What changes were proposed in this pull request?
Remove the `private[sql]` access modifier from `DataStreamReader.name` and add the method as a public abstract API to the `DataStreamReader` base class.
- Added abstract `def name(sourceName: String): this.type` to the API base class (`sql/api/.../DataStreamReader.scala`)
- Changed both classic and connect implementations from `private[sql] def name` to `override def name`
- Moved Scaladoc to the base class; implementations use `inheritdoc`
### Why are the changes needed?
The `name` method was introduced in SPARK-56453 as `private[sql]` while the API was being finalized. Now that the feature is ready, making it public allows users to assign names to streaming sources for stable checkpoint metadata and source evolution.
### Does this PR introduce _any_ user-facing change?
Yes. `DataStreamReader.name(sourceName)` is now a public `Experimental` API available to all users. Previously it was package-private to `org.apache.spark.sql`.
### How was this patch tested?
Existing tests cover the `name` functionality. This change only modifies the access level.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.6)
Closes #55651 from ericm-db/datastreamreader-name-public.
Authored-by: ericm-db <er...@databricks.com>
Signed-off-by: Anish Shrigondekar <an...@databricks.com>
---
project/MimaExcludes.scala | 4 +++-
.../org/apache/spark/sql/streaming/DataStreamReader.scala | 13 ++++++++++++-
.../org/apache/spark/sql/connect/DataStreamReader.scala | 11 ++---------
.../org/apache/spark/sql/classic/DataStreamReader.scala | 10 ++--------
4 files changed, 19 insertions(+), 19 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index b5434efee090..bb0bc97321c3 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -55,7 +55,9 @@ object MimaExcludes {
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.copy"),
ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo$"),
// [SPARK-56330][CORE] Add TaskInterruptListener to TaskContext for interrupt notifications
- ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.addTaskInterruptListener")
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.addTaskInterruptListener"),
+ // [SPARK-56700][SS] Make DataStreamReader.name public
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.streaming.DataStreamReader.name")
)
// Exclude rules for 4.1.x from 4.0.0
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala b/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
index b90a7c910f9c..102d71984a33 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.streaming
import scala.jdk.CollectionConverters._
import scala.util.matching.Regex
-import org.apache.spark.annotation.Evolving
+import org.apache.spark.annotation.{Evolving, Experimental}
import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoders}
import org.apache.spark.sql.types.StructType
@@ -104,6 +104,17 @@ abstract class DataStreamReader {
this
}
+ /**
+ * Specifies a name for the streaming source. This name is used to identify the source in
+ * checkpoint metadata and enables stable checkpoint locations for source evolution.
+ *
+ * @param sourceName
+ * the name to assign to this streaming source
+ * @since 4.2.0
+ */
+ @Experimental
+ def name(sourceName: String): this.type
+
/**
* Loads input data stream in as a `DataFrame`, for data streams that don't require a path (e.g.
* external key-value stores).
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala
index 2f8040273995..50c0de85e0cb 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala
@@ -75,16 +75,9 @@ final class DataStreamReader private[sql] (sparkSession: SparkSession)
this
}
- /**
- * Specifies a name for the streaming source. This name is used to identify the source in
- * checkpoint metadata and enables stable checkpoint locations for source evolution.
- *
- * @param sourceName
- * the name to assign to this streaming source
- * @since 4.2.0
- */
+ /** @inheritdoc */
@Experimental
- private[sql] def name(sourceName: String): this.type = {
+ override def name(sourceName: String): this.type = {
validateSourceName(sourceName)
sourceBuilder.setSourceName(sourceName)
this
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
index f2a2e99fdc14..a3ab235372d8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
@@ -67,15 +67,9 @@ final class DataStreamReader private[sql](sparkSession: SparkSession)
this
}
- /**
- * Specifies a name for the streaming source. This name is used to identify the source
- * in checkpoint metadata and enables stable checkpoint locations for source evolution.
- *
- * @param sourceName the name to assign to this streaming source
- * @since 4.2.0
- */
+ /** @inheritdoc */
@Experimental
- private[sql] def name(sourceName: String): this.type = {
+ override def name(sourceName: String): this.type = {
validateSourceName(sourceName)
this.userProvidedSourceName = Option(sourceName)
this
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org