You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by "gengliangwang (via GitHub)" <gi...@apache.org> on 2026/03/10 20:00:49 UTC

[PR] [SPARK-XXXXX][SQL] Add DataFrame API and Spark Connect support for CDC queries [spark]

gengliangwang opened a new pull request, #54739:
URL: https://github.com/apache/spark/pull/54739

   ## Summary
   
   This is PR 3/3 of the CDC framework. It adds the DataFrame API (`changes()`) and Spark Connect support. Depends on #54738.
   
   ### DataFrame API
   
   New methods on `DataFrameReader` and `DataStreamReader`:
   
   ```scala
   // Batch
   spark.read
     .option("startingVersion", "1")
     .option("endingVersion", "5")
     .option("deduplicationMode", "netChanges")
     .changes("catalog.table")
   
   // Streaming
   spark.readStream
     .option("startingVersion", "1")
     .changes("catalog.table")
   ```
   
   **API definitions** in `sql/api`:
   - `DataFrameReader.changes(tableName: String): DataFrame`
   - `DataStreamReader.changes(tableName: String): DataFrame`
   
   **Classic implementation** (`sql/core`): Parses options via `ChangelogInfoUtils.fromOptions()` using the session timezone, creates `UnresolvedRelation` + `RelationChanges`.
   
   ### Spark Connect
   
   - **Proto** (`relations.proto`): New `ReadChanges` message (field 46) with `unparsed_identifier`, `options` map, and `is_streaming` flag.
   - **Client** (`DataFrameReader.scala`, `DataStreamReader.scala`): Serializes `changes()` calls into `ReadChanges` proto messages. Replaces the previous `throw ConnectClientUnsupportedErrors.tableCDC()`.
   - **Server** (`SparkConnectPlanner.scala`): `transformReadChanges()` deserializes the proto, calls `ChangelogInfoUtils.fromOptions()` with the session timezone, and constructs `UnresolvedRelation` + `RelationChanges`.
   
   ### Tests
   
   - **`ChangelogResolutionSuite`**: Added 5 DataFrame API tests — `changes()` resolution, catalog without CDC, schema rejection on `readStream`, streaming resolution to `StreamingRelationV2`, streaming catalog error.
   - **`ChangelogEndToEndSuite`**: Added 2 DataFrame API tests — data retrieval, projection with filter.
   - **`PlanGenerationTestSuite`**: 3 new golden files for `read changes`, `read changes with options`, and `streaming changes API with options`.
   
   ## Test plan
   
   - [x] `ChangelogResolutionSuite` — 10 tests pass (5 SQL + 5 DataFrame)
   - [x] `ChangelogEndToEndSuite` — 7 tests pass (5 SQL + 2 DataFrame)
   - [x] `PlanGenerationTestSuite` — 719 tests pass (3 new)
   
   🤖 Generated with [Claude Code](https://claude.com/claude-code)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-55949][SQL] Add DataFrame API and Spark Connect support for CDC queries [spark]

Posted by "aokolnychyi (via GitHub)" <gi...@apache.org>.

aokolnychyi commented on code in PR #54739:
URL: https://github.com/apache/spark/pull/54739#discussion_r2958592143


##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala:
##########
@@ -119,6 +119,24 @@ abstract class DataStreamReader {
    */
   def load(path: String): DataFrame
 
+  /**
+   * Returns the row-level changes (Change Data Capture) from the specified table as a streaming
+   * `DataFrame`. Currently this API is only supported for Data Source V2 tables whose catalog
+   * implements `TableCatalog.loadChangelog()`.
+   *
+   * Use `.option()` to specify the starting version/timestamp and processing options:
+   * {{{
+   *   spark.readStream
+   *     .option("startingVersion", "10")
+   *     .changes("my_table")
+   * }}}
+   *
+   * @param tableName
+   *   is either a qualified or unqualified name that designates a table.

Review Comment:
   Minor: can we shorten it to fit on one line?



##########
sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala:
##########
@@ -497,6 +497,25 @@ abstract class DataFrameReader {
    */
   def table(tableName: String): DataFrame
 
+  /**
+   * Returns the row-level changes (Change Data Capture) from the specified table as a
+   * `DataFrame`. Currently this API is only supported for Data Source V2 tables whose catalog
+   * implements `TableCatalog.loadChangelog()`.
+   *
+   * Use `.option()` to specify the version/timestamp range and processing options:
+   * {{{
+   *   spark.read
+   *     .option("startingVersion", "10")
+   *     .option("endingVersion", "20")
+   *     .changes("my_table")
+   * }}}
+   *
+   * @param tableName
+   *   is either a qualified or unqualified name that designates a table.

Review Comment:
   Same.



##########
sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.json:
##########
@@ -0,0 +1,13 @@
+{
+  "common": {
+    "planId": "0"
+  },
+  "relationChanges": {
+    "unparsedIdentifier": "myTable",
+    "options": {
+      "startingtimestamp": "2026-01-01",
+      "deduplicationmode": "netChanges",
+      "computeupdates": "true"
+    }
+  }
+}

Review Comment:
   Same?



##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala:
##########
@@ -196,6 +196,19 @@ class DataFrameReader private[sql] (sparkSession: SparkSession) extends sql.Data
     }
   }
 
+  /** @inheritdoc */
+  def changes(tableName: String): DataFrame = {
+    require(tableName != null, "The table name can't be null")
+    assertNoSpecifiedSchema("changes")
+    sparkSession.newDataFrame { builder =>
+      val changesBuilder = builder.getRelationChangesBuilder
+        .setUnparsedIdentifier(tableName)
+      extraOptions.foreach { case (k, v) =>

Review Comment:
   Should we use `putAllOptions` here to be consistent with `table` method above?



##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala:
##########
@@ -1715,6 +1717,17 @@ class SparkConnectPlanner(
     }
   }
 
+  private def transformRelationChanges(rel: proto.RelationChanges): LogicalPlan = {
+    val tableName = rel.getUnparsedIdentifier
+    val options = new CaseInsensitiveStringMap(rel.getOptionsMap)
+    val changelogInfo = ChangelogInfoUtils.fromOptions(
+      options, session.sessionState.conf.sessionLocalTimeZone)

Review Comment:
   Optional: I would define a temp var for time zone and would move UnresolvedRelation on one line.
   
   ```
   val timeZone = session.sessionState.conf.sessionLocalTimeZone
   val changelogInfo = ChangelogInfoUtils.fromOptions(options, timeZone)
   val ident = parser.parseMultipartIdentifier(tableName)
   val relation = UnresolvedRelation(ident, options, isStreaming = rel.getIsStreaming)
   ```



##########
sql/connect/common/src/test/resources/query-tests/queries/read_changes.json:
##########
@@ -0,0 +1,12 @@
+{
+  "common": {
+    "planId": "0"
+  },
+  "relationChanges": {
+    "unparsedIdentifier": "myTable",
+    "options": {
+      "startingversion": "1",
+      "endingversion": "5"
+    }
+  }
+}

Review Comment:
   Do we ned empty line at the end?



##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala:
##########
@@ -118,6 +118,20 @@ final class DataStreamReader private[sql] (sparkSession: SparkSession)
     }
   }
 
+  /** @inheritdoc */
+  def changes(tableName: String): DataFrame = {
+    require(tableName != null, "The table name can't be null")
+    assertNoSpecifiedSchema("changes")
+    sparkSession.newDataFrame { builder =>
+      val changesBuilder = builder.getRelationChangesBuilder
+        .setUnparsedIdentifier(tableName)
+        .setIsStreaming(true)
+      sourceBuilder.getOptionsMap.forEach { (k, v) =>

Review Comment:
   Same question about `putAllOptions`?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-55949][SQL] Add DataFrame API and Spark Connect support for CDC queries [spark]

Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.

hvanhovell commented on code in PR #54739:
URL: https://github.com/apache/spark/pull/54739#discussion_r2915387399


##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/ConnectClientUnsupportedErrors.scala:
##########
@@ -50,4 +50,5 @@ private[sql] object ConnectClientUnsupportedErrors {
 
   def sparkContext(): SparkUnsupportedOperationException =
     unsupportedFeatureException("SESSION_SPARK_CONTEXT")
+

Review Comment:
   ?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org