[SPARK-55949][SQL][CONNECT] Add DataFrame API and Spark Connect support for CDC queries

project/MimaExcludes.scala

@@ -43,6 +43,9 @@ object MimaExcludes {
  ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.ui.ProxyRedirectHandler#ResponseWrapper.this"),
  // [SPARK-55228][SQL] Implement Dataset.zipWithIndex in Scala API
  ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.Dataset.zipWithIndex"),
+ // [SPARK-55949][SQL] Add DataFrame API for CDC queries
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.DataFrameReader.changes"),
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.streaming.DataStreamReader.changes"),
  // [SPARK-55793][CORE] Add multiple log directories support to SHS
  ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.apply"),
  ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.copy"),

python/pyspark/sql/connect/proto/relations_pb2.pyi

@@ -110,6 +110,7 @@ class Relation(google.protobuf.message.Message):
  UNRESOLVED_TABLE_VALUED_FUNCTION_FIELD_NUMBER: builtins.int
  LATERAL_JOIN_FIELD_NUMBER: builtins.int
  CHUNKED_CACHED_LOCAL_RELATION_FIELD_NUMBER: builtins.int
+ RELATION_CHANGES_FIELD_NUMBER: builtins.int
  FILL_NA_FIELD_NUMBER: builtins.int
  DROP_NA_FIELD_NUMBER: builtins.int
  REPLACE_FIELD_NUMBER: builtins.int
@@ -220,6 +221,8 @@ class Relation(google.protobuf.message.Message):
  @property
  def chunked_cached_local_relation(self) -> global___ChunkedCachedLocalRelation: ...
  @property
+ def relation_changes(self) -> global___RelationChanges: ...
+ @property
  def fill_na(self) -> global___NAFill:
  """NA functions"""
 
@@ -313,6 +316,7 @@ class Relation(google.protobuf.message.Message):
  unresolved_table_valued_function: global___UnresolvedTableValuedFunction | None = ...,
  lateral_join: global___LateralJoin | None = ...,
  chunked_cached_local_relation: global___ChunkedCachedLocalRelation | None = ...,
+ relation_changes: global___RelationChanges | None = ...,
  fill_na: global___NAFill | None = ...,
  drop_na: global___NADrop | None = ...,
  replace: global___NAReplace | None = ...,
@@ -410,6 +414,8 @@ class Relation(google.protobuf.message.Message):
  b"read",
  "rel_type",
  b"rel_type",
+ "relation_changes",
+ b"relation_changes",
  "repartition",
  b"repartition",
  "repartition_by_expression",
@@ -537,6 +543,8 @@ class Relation(google.protobuf.message.Message):
  b"read",
  "rel_type",
  b"rel_type",
+ "relation_changes",
+ b"relation_changes",
  "repartition",
  b"repartition",
  "repartition_by_expression",
@@ -629,6 +637,7 @@ class Relation(google.protobuf.message.Message):
  "unresolved_table_valued_function",
  "lateral_join",
  "chunked_cached_local_relation",
+ "relation_changes",
  "fill_na",
  "drop_na",
  "replace",
@@ -1329,6 +1338,68 @@ class Read(google.protobuf.message.Message):
 
 global___Read = Read
 
+class RelationChanges(google.protobuf.message.Message):
+ """Reads Change Data Capture (CDC) changes for a named table.
+
+ This corresponds to the `DataFrameReader.changes()` or `DataStreamReader.changes()` API.
+ CDC-specific options (startingVersion, endingVersion, startingTimestamp, endingTimestamp,
+ deduplicationMode, computeUpdates, etc.) are passed in the options map.
+ """
+
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ class OptionsEntry(google.protobuf.message.Message):
+ DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+ KEY_FIELD_NUMBER: builtins.int
+ VALUE_FIELD_NUMBER: builtins.int
+ key: builtins.str
+ value: builtins.str
+ def __init__(
+ self,
+ *,
+ key: builtins.str = ...,
+ value: builtins.str = ...,
+ ) -> None: ...
+ def ClearField(
+ self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]
+ ) -> None: ...
+
+ UNPARSED_IDENTIFIER_FIELD_NUMBER: builtins.int
+ OPTIONS_FIELD_NUMBER: builtins.int
+ IS_STREAMING_FIELD_NUMBER: builtins.int
+ unparsed_identifier: builtins.str
+ """(Required) Unparsed identifier for the table."""
+ @property
+ def options(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, builtins.str]:
+ """Options for the CDC query. The map key is case insensitive.
+ Supported keys include: startingVersion, endingVersion, startingTimestamp,
+ endingTimestamp, deduplicationMode, computeUpdates, startingBoundInclusive,
+ endingBoundInclusive.
+ """
+ is_streaming: builtins.bool
+ """(Optional) Indicates if this is a streaming CDC read."""
+ def __init__(
+ self,
+ *,
+ unparsed_identifier: builtins.str = ...,
+ options: collections.abc.Mapping[builtins.str, builtins.str] | None = ...,
+ is_streaming: builtins.bool = ...,
+ ) -> None: ...
+ def ClearField(
+ self,
+ field_name: typing_extensions.Literal[
+ "is_streaming",
+ b"is_streaming",
+ "options",
+ b"options",
+ "unparsed_identifier",
+ b"unparsed_identifier",
+ ],
+ ) -> None: ...
+
+global___RelationChanges = RelationChanges
+
 class Project(google.protobuf.message.Message):
  """Projection of a bag of expressions for a given input relation.
 

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
+ * a qualified or unqualified name that designates a table.
+ * @since 4.2.0
+ */
+ def changes(tableName: String): DataFrame
+
  /**
  * Loads text files and returns a `DataFrame` whose schema starts with a string column named
  * "value", and followed by partitioned columns if there are any. See the documentation on the

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
+ * a qualified or unqualified name that designates a table.
+ * @since 4.2.0
+ */
+ def changes(tableName: String): DataFrame
+
  /**
  * Loads a JSON file stream and returns the results as a `DataFrame`.
  *

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala

@@ -33,7 +33,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
  *
  * Change rows can be pre-populated via [[addChangeRows()]] before querying.
  */
-class InMemoryChangelogCatalog extends InMemoryTableCatalog {
+class InMemoryChangelogCatalog extends InMemoryCatalog {
 
  // tableName -> list of change rows (each row: Array[Any] matching changelog schema)
  private val changeData: mutable.Map[String, mutable.ArrayBuffer[InternalRow]] =

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala

@@ -393,6 +393,21 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
  session.table("myTable")
  }
 
+ test("read changes") {
+ session.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "5")
+ .changes("myTable")
+ }
+
+ test("read changes with options") {
+ session.read
+ .option("startingTimestamp", "2026-01-01")
+ .option("deduplicationMode", "netChanges")
+ .option("computeUpdates", "true")
+ .changes("myTable")
+ }
+
  test("read text") {
  session.read.text(testDataPath.resolve("people.txt").toString)
  }
@@ -3546,6 +3561,13 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
  session.readStream.options(Map("p1" -> "v1", "p2" -> "v2")).table("tempdb.myStreamingTable")
  }
 
+ test("streaming changes API with options") {
+ session.readStream
+ .option("startingVersion", "1")
+ .option("deduplicationMode", "dropCarryovers")
+ .changes("tempdb.myStreamingTable")
+ }
+
  /* Avro functions */
  test("from_avro with options") {
  binary.select(

sql/connect/common/src/main/protobuf/spark/connect/relations.proto

@@ -81,6 +81,7 @@ message Relation {
  UnresolvedTableValuedFunction unresolved_table_valued_function = 43;
  LateralJoin lateral_join = 44;
  ChunkedCachedLocalRelation chunked_cached_local_relation = 45;
+ RelationChanges relation_changes = 46;
 
  // NA functions
  NAFill fill_na = 90;
@@ -264,6 +265,25 @@ message Read {
  }
 }
 
+// Reads Change Data Capture (CDC) changes for a named table.
+//
+// This corresponds to the `DataFrameReader.changes()` or `DataStreamReader.changes()` API.
+// CDC-specific options (startingVersion, endingVersion, startingTimestamp, endingTimestamp,
+// deduplicationMode, computeUpdates, etc.) are passed in the options map.
+message RelationChanges {
+ // (Required) Unparsed identifier for the table.
+ string unparsed_identifier = 1;
+
+ // Options for the CDC query. The map key is case insensitive.
+ // Supported keys include: startingVersion, endingVersion, startingTimestamp,
+ // endingTimestamp, deduplicationMode, computeUpdates, startingBoundInclusive,
+ // endingBoundInclusive.
+ map<string, string> options = 2;
+
+ // (Optional) Indicates if this is a streaming CDC read.
+ bool is_streaming = 3;
+}
+
 // Projection of a bag of expressions for a given input relation.
 //
 // The input relation must be specified.

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala

@@ -196,6 +196,17 @@ 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 =>
+ builder.getRelationChangesBuilder
+ .setUnparsedIdentifier(tableName)
+ .putAllOptions(extraOptions.toMap.asJava)
+ }
+ }
+
  /** @inheritdoc */
  override def text(path: String): DataFrame = super.text(path)
 

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataStreamReader.scala

@@ -118,6 +118,18 @@ 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 =>
+ builder.getRelationChangesBuilder
+ .setUnparsedIdentifier(tableName)
+ .setIsStreaming(true)
+ .putAllOptions(sourceBuilder.getOptionsMap)
+ }
+ }
+
  override protected def assertNoSpecifiedSchema(operation: String): Unit = {
  if (sourceBuilder.hasSchema) {
  throw DataTypeErrors.userSpecifiedSchemaUnsupportedError(operation)

sql/connect/common/src/test/resources/query-tests/explain-results/read_changes.explain

@@ -0,0 +1,2 @@
+SubqueryAlias primary.tempdb.myTable
++- RelationV2[id#0L, _change_type#0, _commit_version#0L, _commit_timestamp#0] primary.tempdb.myTable

sql/connect/common/src/test/resources/query-tests/explain-results/read_changes_with_options.explain

@@ -0,0 +1,2 @@
+SubqueryAlias primary.tempdb.myTable
++- RelationV2[id#0L, _change_type#0, _commit_version#0L, _commit_timestamp#0] primary.tempdb.myTable

sql/connect/common/src/test/resources/query-tests/explain-results/streaming_changes_API_with_options.explain

@@ -0,0 +1,2 @@
+~SubqueryAlias primary.tempdb.myStreamingTable
++- ~StreamingRelationV2 primary.tempdb.myStreamingTable_changelog, ChangelogTable(org.apache.spark.sql.connector.catalog.InMemoryChangelog,ChangelogInfo{range=VersionRange[startingVersion=1, endingVersion=Optional.empty, startingBoundInclusive=true, endingBoundInclusive=true], deduplicationMode=DROP_CARRYOVERS, computeUpdates=false}), [startingVersion=1, deduplicationMode=dropCarryovers], [id#0L, _change_type#0, _commit_version#0L, _commit_timestamp#0], org.apache.spark.sql.connector.catalog.InMemoryChangelogCatalog, tempdb.myStreamingTable, name=<Unassigned>

sql/connect/common/src/test/resources/query-tests/explain-results/streaming_table_API_with_options.explain

@@ -1,2 +1,2 @@
 ~SubqueryAlias primary.tempdb.myStreamingTable
-+- ~StreamingRelationV2 primary.tempdb.myStreamingTable, org.apache.spark.sql.connector.catalog.InMemoryTable, [p1=v1, p2=v2], [id#0L], org.apache.spark.sql.connector.catalog.InMemoryCatalog, tempdb.myStreamingTable, name=<Unassigned>
++- ~StreamingRelationV2 primary.tempdb.myStreamingTable, org.apache.spark.sql.connector.catalog.InMemoryTable, [p1=v1, p2=v2], [id#0L], org.apache.spark.sql.connector.catalog.InMemoryChangelogCatalog, tempdb.myStreamingTable, name=<Unassigned>

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"
+ }
+ }
+}

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"
+ }
+ }
+}

sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.json

@@ -0,0 +1,13 @@
+{
+ "common": {
+ "planId": "0"
+ },
+ "relationChanges": {
+ "unparsedIdentifier": "tempdb.myStreamingTable",
+ "options": {
+ "startingVersion": "1",
+ "deduplicationMode": "dropCarryovers"
+ },
+ "isStreaming": true
+ }
+}

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

@@ -43,7 +43,7 @@ import org.apache.spark.internal.LogKeys.{DATAFRAME_ID, SESSION_ID}
 import org.apache.spark.resource.{ExecutorResourceRequest, ResourceProfile, TaskResourceProfile, TaskResourceRequest}
 import org.apache.spark.sql.{AnalysisException, Column, Encoders, ForeachWriter, Row}
 import org.apache.spark.sql.catalyst.{expressions, AliasIdentifier, FunctionIdentifier, InternalRow, QueryPlanningTracker}
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, GlobalTempView, LocalTempView, MultiAlias, UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar, UnresolvedDeserializer, UnresolvedExtractValue, UnresolvedFunction, UnresolvedOrdinal, UnresolvedPlanId, UnresolvedRegex, UnresolvedRelation, UnresolvedStar, UnresolvedStarWithColumns, UnresolvedStarWithColumnsRenames, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTranspose}
+import org.apache.spark.sql.catalyst.analysis.{ChangelogInfoUtils, FunctionRegistry, GlobalTempView, LocalTempView, MultiAlias, RelationChanges, UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar, UnresolvedDeserializer, UnresolvedExtractValue, UnresolvedFunction, UnresolvedOrdinal, UnresolvedPlanId, UnresolvedRegex, UnresolvedRelation, UnresolvedStar, UnresolvedStarWithColumns, UnresolvedStarWithColumnsRenames, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTranspose}
 import org.apache.spark.sql.catalyst.encoders.{encoderFor, AgnosticEncoder, ExpressionEncoder, RowEncoder}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ProductEncoder, RowEncoder => AgnosticRowEncoder, StringEncoder, UnboundRowEncoder}
 import org.apache.spark.sql.catalyst.expressions._
@@ -148,6 +148,8 @@ class SparkConnectPlanner(
  case proto.Relation.RelTypeCase.SHOW_STRING => transformShowString(rel.getShowString)
  case proto.Relation.RelTypeCase.HTML_STRING => transformHtmlString(rel.getHtmlString)
  case proto.Relation.RelTypeCase.READ => transformReadRel(rel.getRead)
+ case proto.Relation.RelTypeCase.RELATION_CHANGES =>
+ transformRelationChanges(rel.getRelationChanges)
  case proto.Relation.RelTypeCase.PROJECT => transformProject(rel.getProject)
  case proto.Relation.RelTypeCase.FILTER => transformFilter(rel.getFilter)
  case proto.Relation.RelTypeCase.LIMIT => transformLimit(rel.getLimit)
@@ -1715,6 +1717,16 @@ class SparkConnectPlanner(
  }
  }
 
+ private def transformRelationChanges(rel: proto.RelationChanges): LogicalPlan = {
+ val tableName = rel.getUnparsedIdentifier
+ val options = new CaseInsensitiveStringMap(rel.getOptionsMap)
+ 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)
+ RelationChanges(relation, changelogInfo)
+ }
+
  private def transformParse(rel: proto.Parse): LogicalPlan = {
  def dataFrameReader = {
  val localMap = CaseInsensitiveMap[String](rel.getOptionsMap.asScala.toMap)