You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2026/03/20 03:58:37 UTC
(spark) branch master updated: [SPARK-55949][SQL][CONNECT] Add DataFrame API and Spark Connect support for CDC queries
This is an automated email from the ASF dual-hosted git repository.
gengliang 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 58841fa416a9 [SPARK-55949][SQL][CONNECT] Add DataFrame API and Spark Connect support for CDC queries
58841fa416a9 is described below
commit 58841fa416a97dfc3b6fb4bda55d157e7fba8143
Author: Gengliang Wang <ge...@apache.org>
AuthorDate: Thu Mar 19 20:58:27 2026 -0700
[SPARK-55949][SQL][CONNECT] Add DataFrame API and Spark Connect support for CDC queries
### What changes were proposed in this pull request?
This is PR 2/2 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()`, `DataStreamReader.changes()`). Currently only supported for DSv2 tables whose catalog implements `TableCatalog.loadChangelog()`.
- Classic implementation in `sql/core`: Parses options via `ChangelogInfoUtils.fromOptions()` using the session timezone, creates `UnresolvedRelation` + `RelationChanges`.
**Spark Connect support:**
- Proto (`relations.proto`): New `RelationChanges` message (field 46) with `unparsed_identifier`, `options` map, and `is_streaming` flag.
- Client (`DataFrameReader.scala`, `DataStreamReader.scala`): Serializes `changes()` calls into `RelationChanges` proto.
- Server (`SparkConnectPlanner.scala`): `transformRelationChanges()` deserializes the proto, calls `ChangelogInfoUtils.fromOptions()` with the session timezone, and constructs `UnresolvedRelation` + `RelationChanges`.
### Why are the changes needed?
Users need a programmatic API to query CDC data, complementing the SQL `CHANGES` clause added in #54738. The DataFrame API provides type-safe, composable access to change data with the standard option-based configuration pattern. Spark Connect support ensures feature parity for remote Spark sessions.
### Does this PR introduce _any_ user-facing change?
Yes. New DataFrame API methods:
```scala
// Before: spark.read.changes("table") throws SparkUnsupportedOperationException
// After:
val changes = spark.read
.option("startingVersion", "1")
.changes("my_catalog.my_table")
changes.filter("_change_type = 'insert'").show()
```
Both `DataFrameReader.changes()` and `DataStreamReader.changes()` are now functional in classic and Spark Connect sessions.
### How was this patch tested?
- **`ChangelogResolutionSuite`** — 5 tests: `changes()` resolution, catalog without CDC support, schema rejection on `readStream`, streaming resolution to `StreamingRelationV2`, streaming catalog error.
- **`ChangelogEndToEndSuite`** — 19 end-to-end tests using `InMemoryChangelogCatalog`, each pairing the DataFrame API with the equivalent SQL syntax:
- Basic retrieval (3 tests): data retrieval, open-ended version range, empty results with schema validation.
- Projection/filter/aggregation (4 tests): CDC metadata column selection, projection with filter, aggregation on change types, schema verification.
- Version range filtering (1 test): inclusive range with boundary rows.
- Bound inclusivity (4 tests): default inclusive bounds (with explicit `INCLUSIVE` keyword), `startingBoundInclusive=false` (SQL: `EXCLUSIVE`), `endingBoundInclusive=false`, both bounds exclusive.
- CDC options (3 tests): default `dropCarryovers` mode, `deduplicationMode=none`, `netChanges` with `computeUpdates`.
- Timestamp range (1 test): `startingTimestamp`/`endingTimestamp` (SQL: `FROM TIMESTAMP ... TO TIMESTAMP ...`).
- Error cases (1 test): user-specified schema rejection.
- Streaming (4 tests): basic retrieval, `startingVersion` filtering, projection with filter, CDC options pass-through — all with both DataFrame API and SQL (`SELECT * FROM STREAM table CHANGES ...`) variants.
- Streaming error cases (1 test): schema rejection.
- **`PlanGenerationTestSuite`** — 3 golden files: `read_changes`, `read_changes_with_options`, `streaming_changes_API_with_options`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.6)
Closes #54739 from gengliangwang/cdc-dataframe.
Authored-by: Gengliang Wang <ge...@apache.org>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
project/MimaExcludes.scala | 3 +
python/pyspark/sql/connect/proto/relations_pb2.py | 346 ++++++------
python/pyspark/sql/connect/proto/relations_pb2.pyi | 71 +++
.../org/apache/spark/sql/DataFrameReader.scala | 19 +
.../spark/sql/streaming/DataStreamReader.scala | 18 +
.../catalog/InMemoryChangelogCatalog.scala | 2 +-
.../apache/spark/sql/PlanGenerationTestSuite.scala | 22 +
.../main/protobuf/spark/connect/relations.proto | 20 +
.../apache/spark/sql/connect/DataFrameReader.scala | 11 +
.../spark/sql/connect/DataStreamReader.scala | 12 +
.../explain-results/read_changes.explain | 2 +
.../read_changes_with_options.explain | 2 +
.../streaming_changes_API_with_options.explain | 2 +
.../streaming_table_API_with_options.explain | 2 +-
.../query-tests/queries/read_changes.json | 12 +
.../query-tests/queries/read_changes.proto.bin | Bin 0 -> 58 bytes
.../queries/read_changes_with_options.json | 13 +
.../queries/read_changes_with_options.proto.bin | Bin 0 -> 106 bytes
.../streaming_changes_API_with_options.json | 13 +
.../streaming_changes_API_with_options.proto.bin | Bin 0 -> 93 bytes
.../sql/connect/planner/SparkConnectPlanner.scala | 14 +-
.../sql/connect/ProtoToParsedPlanTestSuite.scala | 4 +-
.../apache/spark/sql/classic/DataFrameReader.scala | 16 +-
.../spark/sql/classic/DataStreamReader.scala | 16 +-
.../sql/connector/ChangelogEndToEndSuite.scala | 606 ++++++++++++++++++---
.../sql/connector/ChangelogResolutionSuite.scala | 62 +++
26 files changed, 1045 insertions(+), 243 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index c6c012600c3d..89d147419038 100644
--- a/project/MimaExcludes.scala
+++ b/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"),
diff --git a/python/pyspark/sql/connect/proto/relations_pb2.py b/python/pyspark/sql/connect/proto/relations_pb2.py
index dfa511c3ba39..48e587a9fa47 100644
--- a/python/pyspark/sql/connect/proto/relations_pb2.py
+++ b/python/pyspark/sql/connect/proto/relations_pb2.py
@@ -44,7 +44,7 @@ from pyspark.sql.connect.proto import ml_common_pb2 as spark_dot_connect_dot_ml_
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
- b'\n\x1dspark/connect/relations.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1fspark/connect/expressions.proto\x1a\x19spark/connect/types.proto\x1a\x1bspark/connect/catalog.proto\x1a\x1aspark/connect/common.proto\x1a\x1dspark/connect/ml_common.proto"\x8c\x1e\n\x08Relation\x12\x35\n\x06\x63ommon\x18\x01 \x01(\x0b\x32\x1d.spark.connect.RelationCommonR\x06\x63ommon\x12)\n\x04read\x18\x02 \x01(\x0b\x32\x13.spark.connect.ReadH\x00R\x04read\x12\x32\n\x07project\x18\x03 \x [...]
+ b'\n\x1dspark/connect/relations.proto\x12\rspark.connect\x1a\x19google/protobuf/any.proto\x1a\x1fspark/connect/expressions.proto\x1a\x19spark/connect/types.proto\x1a\x1bspark/connect/catalog.proto\x1a\x1aspark/connect/common.proto\x1a\x1dspark/connect/ml_common.proto"\xd9\x1e\n\x08Relation\x12\x35\n\x06\x63ommon\x18\x01 \x01(\x0b\x32\x1d.spark.connect.RelationCommonR\x06\x63ommon\x12)\n\x04read\x18\x02 \x01(\x0b\x32\x13.spark.connect.ReadH\x00R\x04read\x12\x32\n\x07project\x18\x03 \x [...]
)
_globals = globals()
@@ -71,6 +71,8 @@ if not _descriptor._USE_C_DESCRIPTORS:
_globals["_READ_NAMEDTABLE_OPTIONSENTRY"]._serialized_options = b"8\001"
_globals["_READ_DATASOURCE_OPTIONSENTRY"]._loaded_options = None
_globals["_READ_DATASOURCE_OPTIONSENTRY"]._serialized_options = b"8\001"
+ _globals["_RELATIONCHANGES_OPTIONSENTRY"]._loaded_options = None
+ _globals["_RELATIONCHANGES_OPTIONSENTRY"]._serialized_options = b"8\001"
_globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._loaded_options = None
_globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._serialized_options = b"8\001"
_globals["_WITHCOLUMNSRENAMED"].fields_by_name["rename_columns_map"]._loaded_options = None
@@ -80,173 +82,177 @@ if not _descriptor._USE_C_DESCRIPTORS:
_globals["_PARSE_OPTIONSENTRY"]._loaded_options = None
_globals["_PARSE_OPTIONSENTRY"]._serialized_options = b"8\001"
_globals["_RELATION"]._serialized_start = 224
- _globals["_RELATION"]._serialized_end = 4076
- _globals["_MLRELATION"]._serialized_start = 4079
- _globals["_MLRELATION"]._serialized_end = 4563
- _globals["_MLRELATION_TRANSFORM"]._serialized_start = 4291
- _globals["_MLRELATION_TRANSFORM"]._serialized_end = 4526
- _globals["_FETCH"]._serialized_start = 4566
- _globals["_FETCH"]._serialized_end = 4897
- _globals["_FETCH_METHOD"]._serialized_start = 4682
- _globals["_FETCH_METHOD"]._serialized_end = 4897
- _globals["_FETCH_METHOD_ARGS"]._serialized_start = 4770
- _globals["_FETCH_METHOD_ARGS"]._serialized_end = 4897
- _globals["_UNKNOWN"]._serialized_start = 4899
- _globals["_UNKNOWN"]._serialized_end = 4908
- _globals["_RELATIONCOMMON"]._serialized_start = 4911
- _globals["_RELATIONCOMMON"]._serialized_end = 5053
- _globals["_SQL"]._serialized_start = 5056
- _globals["_SQL"]._serialized_end = 5534
- _globals["_SQL_ARGSENTRY"]._serialized_start = 5350
- _globals["_SQL_ARGSENTRY"]._serialized_end = 5440
- _globals["_SQL_NAMEDARGUMENTSENTRY"]._serialized_start = 5442
- _globals["_SQL_NAMEDARGUMENTSENTRY"]._serialized_end = 5534
- _globals["_WITHRELATIONS"]._serialized_start = 5536
- _globals["_WITHRELATIONS"]._serialized_end = 5653
- _globals["_READ"]._serialized_start = 5656
- _globals["_READ"]._serialized_end = 6373
- _globals["_READ_NAMEDTABLE"]._serialized_start = 5834
- _globals["_READ_NAMEDTABLE"]._serialized_end = 6026
- _globals["_READ_NAMEDTABLE_OPTIONSENTRY"]._serialized_start = 5968
- _globals["_READ_NAMEDTABLE_OPTIONSENTRY"]._serialized_end = 6026
- _globals["_READ_DATASOURCE"]._serialized_start = 6029
- _globals["_READ_DATASOURCE"]._serialized_end = 6360
- _globals["_READ_DATASOURCE_OPTIONSENTRY"]._serialized_start = 5968
- _globals["_READ_DATASOURCE_OPTIONSENTRY"]._serialized_end = 6026
- _globals["_PROJECT"]._serialized_start = 6375
- _globals["_PROJECT"]._serialized_end = 6492
- _globals["_FILTER"]._serialized_start = 6494
- _globals["_FILTER"]._serialized_end = 6606
- _globals["_JOIN"]._serialized_start = 6609
- _globals["_JOIN"]._serialized_end = 7270
- _globals["_JOIN_JOINDATATYPE"]._serialized_start = 6948
- _globals["_JOIN_JOINDATATYPE"]._serialized_end = 7040
- _globals["_JOIN_JOINTYPE"]._serialized_start = 7043
- _globals["_JOIN_JOINTYPE"]._serialized_end = 7251
- _globals["_SETOPERATION"]._serialized_start = 7273
- _globals["_SETOPERATION"]._serialized_end = 7752
- _globals["_SETOPERATION_SETOPTYPE"]._serialized_start = 7589
- _globals["_SETOPERATION_SETOPTYPE"]._serialized_end = 7703
- _globals["_LIMIT"]._serialized_start = 7754
- _globals["_LIMIT"]._serialized_end = 7830
- _globals["_OFFSET"]._serialized_start = 7832
- _globals["_OFFSET"]._serialized_end = 7911
- _globals["_TAIL"]._serialized_start = 7913
- _globals["_TAIL"]._serialized_end = 7988
- _globals["_AGGREGATE"]._serialized_start = 7991
- _globals["_AGGREGATE"]._serialized_end = 8757
- _globals["_AGGREGATE_PIVOT"]._serialized_start = 8406
- _globals["_AGGREGATE_PIVOT"]._serialized_end = 8517
- _globals["_AGGREGATE_GROUPINGSETS"]._serialized_start = 8519
- _globals["_AGGREGATE_GROUPINGSETS"]._serialized_end = 8595
- _globals["_AGGREGATE_GROUPTYPE"]._serialized_start = 8598
- _globals["_AGGREGATE_GROUPTYPE"]._serialized_end = 8757
- _globals["_SORT"]._serialized_start = 8760
- _globals["_SORT"]._serialized_end = 8920
- _globals["_DROP"]._serialized_start = 8923
- _globals["_DROP"]._serialized_end = 9064
- _globals["_DEDUPLICATE"]._serialized_start = 9067
- _globals["_DEDUPLICATE"]._serialized_end = 9307
- _globals["_LOCALRELATION"]._serialized_start = 9309
- _globals["_LOCALRELATION"]._serialized_end = 9398
- _globals["_CACHEDLOCALRELATION"]._serialized_start = 9400
- _globals["_CACHEDLOCALRELATION"]._serialized_end = 9472
- _globals["_CHUNKEDCACHEDLOCALRELATION"]._serialized_start = 9474
- _globals["_CHUNKEDCACHEDLOCALRELATION"]._serialized_end = 9586
- _globals["_CACHEDREMOTERELATION"]._serialized_start = 9588
- _globals["_CACHEDREMOTERELATION"]._serialized_end = 9643
- _globals["_SAMPLE"]._serialized_start = 9646
- _globals["_SAMPLE"]._serialized_end = 9919
- _globals["_RANGE"]._serialized_start = 9922
- _globals["_RANGE"]._serialized_end = 10067
- _globals["_SUBQUERYALIAS"]._serialized_start = 10069
- _globals["_SUBQUERYALIAS"]._serialized_end = 10183
- _globals["_REPARTITION"]._serialized_start = 10186
- _globals["_REPARTITION"]._serialized_end = 10328
- _globals["_SHOWSTRING"]._serialized_start = 10331
- _globals["_SHOWSTRING"]._serialized_end = 10473
- _globals["_HTMLSTRING"]._serialized_start = 10475
- _globals["_HTMLSTRING"]._serialized_end = 10589
- _globals["_STATSUMMARY"]._serialized_start = 10591
- _globals["_STATSUMMARY"]._serialized_end = 10683
- _globals["_STATDESCRIBE"]._serialized_start = 10685
- _globals["_STATDESCRIBE"]._serialized_end = 10766
- _globals["_STATCROSSTAB"]._serialized_start = 10768
- _globals["_STATCROSSTAB"]._serialized_end = 10869
- _globals["_STATCOV"]._serialized_start = 10871
- _globals["_STATCOV"]._serialized_end = 10967
- _globals["_STATCORR"]._serialized_start = 10970
- _globals["_STATCORR"]._serialized_end = 11107
- _globals["_STATAPPROXQUANTILE"]._serialized_start = 11110
- _globals["_STATAPPROXQUANTILE"]._serialized_end = 11274
- _globals["_STATFREQITEMS"]._serialized_start = 11276
- _globals["_STATFREQITEMS"]._serialized_end = 11401
- _globals["_STATSAMPLEBY"]._serialized_start = 11404
- _globals["_STATSAMPLEBY"]._serialized_end = 11713
- _globals["_STATSAMPLEBY_FRACTION"]._serialized_start = 11605
- _globals["_STATSAMPLEBY_FRACTION"]._serialized_end = 11704
- _globals["_NAFILL"]._serialized_start = 11716
- _globals["_NAFILL"]._serialized_end = 11850
- _globals["_NADROP"]._serialized_start = 11853
- _globals["_NADROP"]._serialized_end = 11987
- _globals["_NAREPLACE"]._serialized_start = 11990
- _globals["_NAREPLACE"]._serialized_end = 12286
- _globals["_NAREPLACE_REPLACEMENT"]._serialized_start = 12145
- _globals["_NAREPLACE_REPLACEMENT"]._serialized_end = 12286
- _globals["_TODF"]._serialized_start = 12288
- _globals["_TODF"]._serialized_end = 12376
- _globals["_WITHCOLUMNSRENAMED"]._serialized_start = 12379
- _globals["_WITHCOLUMNSRENAMED"]._serialized_end = 12761
- _globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._serialized_start = 12623
- _globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._serialized_end = 12690
- _globals["_WITHCOLUMNSRENAMED_RENAME"]._serialized_start = 12692
- _globals["_WITHCOLUMNSRENAMED_RENAME"]._serialized_end = 12761
- _globals["_WITHCOLUMNS"]._serialized_start = 12763
- _globals["_WITHCOLUMNS"]._serialized_end = 12882
- _globals["_WITHWATERMARK"]._serialized_start = 12885
- _globals["_WITHWATERMARK"]._serialized_end = 13019
- _globals["_HINT"]._serialized_start = 13022
- _globals["_HINT"]._serialized_end = 13154
- _globals["_UNPIVOT"]._serialized_start = 13157
- _globals["_UNPIVOT"]._serialized_end = 13484
- _globals["_UNPIVOT_VALUES"]._serialized_start = 13414
- _globals["_UNPIVOT_VALUES"]._serialized_end = 13473
- _globals["_TRANSPOSE"]._serialized_start = 13486
- _globals["_TRANSPOSE"]._serialized_end = 13608
- _globals["_UNRESOLVEDTABLEVALUEDFUNCTION"]._serialized_start = 13610
- _globals["_UNRESOLVEDTABLEVALUEDFUNCTION"]._serialized_end = 13735
- _globals["_TOSCHEMA"]._serialized_start = 13737
- _globals["_TOSCHEMA"]._serialized_end = 13843
- _globals["_REPARTITIONBYEXPRESSION"]._serialized_start = 13846
- _globals["_REPARTITIONBYEXPRESSION"]._serialized_end = 14049
- _globals["_MAPPARTITIONS"]._serialized_start = 14052
- _globals["_MAPPARTITIONS"]._serialized_end = 14284
- _globals["_GROUPMAP"]._serialized_start = 14287
- _globals["_GROUPMAP"]._serialized_end = 15137
- _globals["_TRANSFORMWITHSTATEINFO"]._serialized_start = 15140
- _globals["_TRANSFORMWITHSTATEINFO"]._serialized_end = 15363
- _globals["_COGROUPMAP"]._serialized_start = 15366
- _globals["_COGROUPMAP"]._serialized_end = 15892
- _globals["_APPLYINPANDASWITHSTATE"]._serialized_start = 15895
- _globals["_APPLYINPANDASWITHSTATE"]._serialized_end = 16252
- _globals["_COMMONINLINEUSERDEFINEDTABLEFUNCTION"]._serialized_start = 16255
- _globals["_COMMONINLINEUSERDEFINEDTABLEFUNCTION"]._serialized_end = 16499
- _globals["_PYTHONUDTF"]._serialized_start = 16502
- _globals["_PYTHONUDTF"]._serialized_end = 16679
- _globals["_COMMONINLINEUSERDEFINEDDATASOURCE"]._serialized_start = 16682
- _globals["_COMMONINLINEUSERDEFINEDDATASOURCE"]._serialized_end = 16833
- _globals["_PYTHONDATASOURCE"]._serialized_start = 16835
- _globals["_PYTHONDATASOURCE"]._serialized_end = 16910
- _globals["_COLLECTMETRICS"]._serialized_start = 16913
- _globals["_COLLECTMETRICS"]._serialized_end = 17049
- _globals["_PARSE"]._serialized_start = 17052
- _globals["_PARSE"]._serialized_end = 17440
- _globals["_PARSE_OPTIONSENTRY"]._serialized_start = 5968
- _globals["_PARSE_OPTIONSENTRY"]._serialized_end = 6026
- _globals["_PARSE_PARSEFORMAT"]._serialized_start = 17341
- _globals["_PARSE_PARSEFORMAT"]._serialized_end = 17429
- _globals["_ASOFJOIN"]._serialized_start = 17443
- _globals["_ASOFJOIN"]._serialized_end = 17918
- _globals["_LATERALJOIN"]._serialized_start = 17921
- _globals["_LATERALJOIN"]._serialized_end = 18151
+ _globals["_RELATION"]._serialized_end = 4153
+ _globals["_MLRELATION"]._serialized_start = 4156
+ _globals["_MLRELATION"]._serialized_end = 4640
+ _globals["_MLRELATION_TRANSFORM"]._serialized_start = 4368
+ _globals["_MLRELATION_TRANSFORM"]._serialized_end = 4603
+ _globals["_FETCH"]._serialized_start = 4643
+ _globals["_FETCH"]._serialized_end = 4974
+ _globals["_FETCH_METHOD"]._serialized_start = 4759
+ _globals["_FETCH_METHOD"]._serialized_end = 4974
+ _globals["_FETCH_METHOD_ARGS"]._serialized_start = 4847
+ _globals["_FETCH_METHOD_ARGS"]._serialized_end = 4974
+ _globals["_UNKNOWN"]._serialized_start = 4976
+ _globals["_UNKNOWN"]._serialized_end = 4985
+ _globals["_RELATIONCOMMON"]._serialized_start = 4988
+ _globals["_RELATIONCOMMON"]._serialized_end = 5130
+ _globals["_SQL"]._serialized_start = 5133
+ _globals["_SQL"]._serialized_end = 5611
+ _globals["_SQL_ARGSENTRY"]._serialized_start = 5427
+ _globals["_SQL_ARGSENTRY"]._serialized_end = 5517
+ _globals["_SQL_NAMEDARGUMENTSENTRY"]._serialized_start = 5519
+ _globals["_SQL_NAMEDARGUMENTSENTRY"]._serialized_end = 5611
+ _globals["_WITHRELATIONS"]._serialized_start = 5613
+ _globals["_WITHRELATIONS"]._serialized_end = 5730
+ _globals["_READ"]._serialized_start = 5733
+ _globals["_READ"]._serialized_end = 6450
+ _globals["_READ_NAMEDTABLE"]._serialized_start = 5911
+ _globals["_READ_NAMEDTABLE"]._serialized_end = 6103
+ _globals["_READ_NAMEDTABLE_OPTIONSENTRY"]._serialized_start = 6045
+ _globals["_READ_NAMEDTABLE_OPTIONSENTRY"]._serialized_end = 6103
+ _globals["_READ_DATASOURCE"]._serialized_start = 6106
+ _globals["_READ_DATASOURCE"]._serialized_end = 6437
+ _globals["_READ_DATASOURCE_OPTIONSENTRY"]._serialized_start = 6045
+ _globals["_READ_DATASOURCE_OPTIONSENTRY"]._serialized_end = 6103
+ _globals["_RELATIONCHANGES"]._serialized_start = 6453
+ _globals["_RELATIONCHANGES"]._serialized_end = 6685
+ _globals["_RELATIONCHANGES_OPTIONSENTRY"]._serialized_start = 6045
+ _globals["_RELATIONCHANGES_OPTIONSENTRY"]._serialized_end = 6103
+ _globals["_PROJECT"]._serialized_start = 6687
+ _globals["_PROJECT"]._serialized_end = 6804
+ _globals["_FILTER"]._serialized_start = 6806
+ _globals["_FILTER"]._serialized_end = 6918
+ _globals["_JOIN"]._serialized_start = 6921
+ _globals["_JOIN"]._serialized_end = 7582
+ _globals["_JOIN_JOINDATATYPE"]._serialized_start = 7260
+ _globals["_JOIN_JOINDATATYPE"]._serialized_end = 7352
+ _globals["_JOIN_JOINTYPE"]._serialized_start = 7355
+ _globals["_JOIN_JOINTYPE"]._serialized_end = 7563
+ _globals["_SETOPERATION"]._serialized_start = 7585
+ _globals["_SETOPERATION"]._serialized_end = 8064
+ _globals["_SETOPERATION_SETOPTYPE"]._serialized_start = 7901
+ _globals["_SETOPERATION_SETOPTYPE"]._serialized_end = 8015
+ _globals["_LIMIT"]._serialized_start = 8066
+ _globals["_LIMIT"]._serialized_end = 8142
+ _globals["_OFFSET"]._serialized_start = 8144
+ _globals["_OFFSET"]._serialized_end = 8223
+ _globals["_TAIL"]._serialized_start = 8225
+ _globals["_TAIL"]._serialized_end = 8300
+ _globals["_AGGREGATE"]._serialized_start = 8303
+ _globals["_AGGREGATE"]._serialized_end = 9069
+ _globals["_AGGREGATE_PIVOT"]._serialized_start = 8718
+ _globals["_AGGREGATE_PIVOT"]._serialized_end = 8829
+ _globals["_AGGREGATE_GROUPINGSETS"]._serialized_start = 8831
+ _globals["_AGGREGATE_GROUPINGSETS"]._serialized_end = 8907
+ _globals["_AGGREGATE_GROUPTYPE"]._serialized_start = 8910
+ _globals["_AGGREGATE_GROUPTYPE"]._serialized_end = 9069
+ _globals["_SORT"]._serialized_start = 9072
+ _globals["_SORT"]._serialized_end = 9232
+ _globals["_DROP"]._serialized_start = 9235
+ _globals["_DROP"]._serialized_end = 9376
+ _globals["_DEDUPLICATE"]._serialized_start = 9379
+ _globals["_DEDUPLICATE"]._serialized_end = 9619
+ _globals["_LOCALRELATION"]._serialized_start = 9621
+ _globals["_LOCALRELATION"]._serialized_end = 9710
+ _globals["_CACHEDLOCALRELATION"]._serialized_start = 9712
+ _globals["_CACHEDLOCALRELATION"]._serialized_end = 9784
+ _globals["_CHUNKEDCACHEDLOCALRELATION"]._serialized_start = 9786
+ _globals["_CHUNKEDCACHEDLOCALRELATION"]._serialized_end = 9898
+ _globals["_CACHEDREMOTERELATION"]._serialized_start = 9900
+ _globals["_CACHEDREMOTERELATION"]._serialized_end = 9955
+ _globals["_SAMPLE"]._serialized_start = 9958
+ _globals["_SAMPLE"]._serialized_end = 10231
+ _globals["_RANGE"]._serialized_start = 10234
+ _globals["_RANGE"]._serialized_end = 10379
+ _globals["_SUBQUERYALIAS"]._serialized_start = 10381
+ _globals["_SUBQUERYALIAS"]._serialized_end = 10495
+ _globals["_REPARTITION"]._serialized_start = 10498
+ _globals["_REPARTITION"]._serialized_end = 10640
+ _globals["_SHOWSTRING"]._serialized_start = 10643
+ _globals["_SHOWSTRING"]._serialized_end = 10785
+ _globals["_HTMLSTRING"]._serialized_start = 10787
+ _globals["_HTMLSTRING"]._serialized_end = 10901
+ _globals["_STATSUMMARY"]._serialized_start = 10903
+ _globals["_STATSUMMARY"]._serialized_end = 10995
+ _globals["_STATDESCRIBE"]._serialized_start = 10997
+ _globals["_STATDESCRIBE"]._serialized_end = 11078
+ _globals["_STATCROSSTAB"]._serialized_start = 11080
+ _globals["_STATCROSSTAB"]._serialized_end = 11181
+ _globals["_STATCOV"]._serialized_start = 11183
+ _globals["_STATCOV"]._serialized_end = 11279
+ _globals["_STATCORR"]._serialized_start = 11282
+ _globals["_STATCORR"]._serialized_end = 11419
+ _globals["_STATAPPROXQUANTILE"]._serialized_start = 11422
+ _globals["_STATAPPROXQUANTILE"]._serialized_end = 11586
+ _globals["_STATFREQITEMS"]._serialized_start = 11588
+ _globals["_STATFREQITEMS"]._serialized_end = 11713
+ _globals["_STATSAMPLEBY"]._serialized_start = 11716
+ _globals["_STATSAMPLEBY"]._serialized_end = 12025
+ _globals["_STATSAMPLEBY_FRACTION"]._serialized_start = 11917
+ _globals["_STATSAMPLEBY_FRACTION"]._serialized_end = 12016
+ _globals["_NAFILL"]._serialized_start = 12028
+ _globals["_NAFILL"]._serialized_end = 12162
+ _globals["_NADROP"]._serialized_start = 12165
+ _globals["_NADROP"]._serialized_end = 12299
+ _globals["_NAREPLACE"]._serialized_start = 12302
+ _globals["_NAREPLACE"]._serialized_end = 12598
+ _globals["_NAREPLACE_REPLACEMENT"]._serialized_start = 12457
+ _globals["_NAREPLACE_REPLACEMENT"]._serialized_end = 12598
+ _globals["_TODF"]._serialized_start = 12600
+ _globals["_TODF"]._serialized_end = 12688
+ _globals["_WITHCOLUMNSRENAMED"]._serialized_start = 12691
+ _globals["_WITHCOLUMNSRENAMED"]._serialized_end = 13073
+ _globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._serialized_start = 12935
+ _globals["_WITHCOLUMNSRENAMED_RENAMECOLUMNSMAPENTRY"]._serialized_end = 13002
+ _globals["_WITHCOLUMNSRENAMED_RENAME"]._serialized_start = 13004
+ _globals["_WITHCOLUMNSRENAMED_RENAME"]._serialized_end = 13073
+ _globals["_WITHCOLUMNS"]._serialized_start = 13075
+ _globals["_WITHCOLUMNS"]._serialized_end = 13194
+ _globals["_WITHWATERMARK"]._serialized_start = 13197
+ _globals["_WITHWATERMARK"]._serialized_end = 13331
+ _globals["_HINT"]._serialized_start = 13334
+ _globals["_HINT"]._serialized_end = 13466
+ _globals["_UNPIVOT"]._serialized_start = 13469
+ _globals["_UNPIVOT"]._serialized_end = 13796
+ _globals["_UNPIVOT_VALUES"]._serialized_start = 13726
+ _globals["_UNPIVOT_VALUES"]._serialized_end = 13785
+ _globals["_TRANSPOSE"]._serialized_start = 13798
+ _globals["_TRANSPOSE"]._serialized_end = 13920
+ _globals["_UNRESOLVEDTABLEVALUEDFUNCTION"]._serialized_start = 13922
+ _globals["_UNRESOLVEDTABLEVALUEDFUNCTION"]._serialized_end = 14047
+ _globals["_TOSCHEMA"]._serialized_start = 14049
+ _globals["_TOSCHEMA"]._serialized_end = 14155
+ _globals["_REPARTITIONBYEXPRESSION"]._serialized_start = 14158
+ _globals["_REPARTITIONBYEXPRESSION"]._serialized_end = 14361
+ _globals["_MAPPARTITIONS"]._serialized_start = 14364
+ _globals["_MAPPARTITIONS"]._serialized_end = 14596
+ _globals["_GROUPMAP"]._serialized_start = 14599
+ _globals["_GROUPMAP"]._serialized_end = 15449
+ _globals["_TRANSFORMWITHSTATEINFO"]._serialized_start = 15452
+ _globals["_TRANSFORMWITHSTATEINFO"]._serialized_end = 15675
+ _globals["_COGROUPMAP"]._serialized_start = 15678
+ _globals["_COGROUPMAP"]._serialized_end = 16204
+ _globals["_APPLYINPANDASWITHSTATE"]._serialized_start = 16207
+ _globals["_APPLYINPANDASWITHSTATE"]._serialized_end = 16564
+ _globals["_COMMONINLINEUSERDEFINEDTABLEFUNCTION"]._serialized_start = 16567
+ _globals["_COMMONINLINEUSERDEFINEDTABLEFUNCTION"]._serialized_end = 16811
+ _globals["_PYTHONUDTF"]._serialized_start = 16814
+ _globals["_PYTHONUDTF"]._serialized_end = 16991
+ _globals["_COMMONINLINEUSERDEFINEDDATASOURCE"]._serialized_start = 16994
+ _globals["_COMMONINLINEUSERDEFINEDDATASOURCE"]._serialized_end = 17145
+ _globals["_PYTHONDATASOURCE"]._serialized_start = 17147
+ _globals["_PYTHONDATASOURCE"]._serialized_end = 17222
+ _globals["_COLLECTMETRICS"]._serialized_start = 17225
+ _globals["_COLLECTMETRICS"]._serialized_end = 17361
+ _globals["_PARSE"]._serialized_start = 17364
+ _globals["_PARSE"]._serialized_end = 17752
+ _globals["_PARSE_OPTIONSENTRY"]._serialized_start = 6045
+ _globals["_PARSE_OPTIONSENTRY"]._serialized_end = 6103
+ _globals["_PARSE_PARSEFORMAT"]._serialized_start = 17653
+ _globals["_PARSE_PARSEFORMAT"]._serialized_end = 17741
+ _globals["_ASOFJOIN"]._serialized_start = 17755
+ _globals["_ASOFJOIN"]._serialized_end = 18230
+ _globals["_LATERALJOIN"]._serialized_start = 18233
+ _globals["_LATERALJOIN"]._serialized_end = 18463
# @@protoc_insertion_point(module_scope)
diff --git a/python/pyspark/sql/connect/proto/relations_pb2.pyi b/python/pyspark/sql/connect/proto/relations_pb2.pyi
index d77517af29ea..8f77615d1d1d 100644
--- a/python/pyspark/sql/connect/proto/relations_pb2.pyi
+++ b/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"""
@property
@@ -306,6 +309,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 = ...,
@@ -403,6 +407,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",
@@ -530,6 +536,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",
@@ -624,6 +632,7 @@ class Relation(google.protobuf.message.Message):
"unresolved_table_valued_function",
"lateral_join",
"chunked_cached_local_relation",
+ "relation_changes",
"fill_na",
"drop_na",
"replace",
@@ -1303,6 +1312,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.
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala b/sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
index 95f1eca66578..501ed3333657 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
+++ b/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
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 242b2a0cee45..dc7d2b94576e 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
@@ -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`.
*
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala
index 2ba2720bbcaa..c47ed2668e3b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryChangelogCatalog.scala
+++ b/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]] =
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
index 1a1b7cd1e24e..0c1123f1a76a 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
+++ b/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(
diff --git a/sql/connect/common/src/main/protobuf/spark/connect/relations.proto b/sql/connect/common/src/main/protobuf/spark/connect/relations.proto
index bf3eb3214e5d..4942142ca50b 100644
--- a/sql/connect/common/src/main/protobuf/spark/connect/relations.proto
+++ b/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.
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala
index 67a4d983f56b..cb6a4c456c22 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/DataFrameReader.scala
+++ b/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)
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 529abc0ec0d1..2f8040273995 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
@@ -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)
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/read_changes.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/read_changes.explain
new file mode 100644
index 000000000000..413eaf8f7a68
--- /dev/null
+++ b/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
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/read_changes_with_options.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/read_changes_with_options.explain
new file mode 100644
index 000000000000..413eaf8f7a68
--- /dev/null
+++ b/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
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/streaming_changes_API_with_options.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/streaming_changes_API_with_options.explain
new file mode 100644
index 000000000000..6f12567607ac
--- /dev/null
+++ b/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.cat [...]
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/streaming_table_API_with_options.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/streaming_table_API_with_options.explain
index 636b53788d39..9ea4ad218a5f 100644
--- a/sql/connect/common/src/test/resources/query-tests/explain-results/streaming_table_API_with_options.explain
+++ b/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>
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/read_changes.json b/sql/connect/common/src/test/resources/query-tests/queries/read_changes.json
new file mode 100644
index 000000000000..75dabbd87077
--- /dev/null
+++ b/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"
+ }
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/read_changes.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/read_changes.proto.bin
new file mode 100644
index 000000000000..fd8da570e7ce
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/read_changes.proto.bin differ
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.json b/sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.json
new file mode 100644
index 000000000000..ddc20aada18b
--- /dev/null
+++ b/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"
+ }
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.proto.bin
new file mode 100644
index 000000000000..00ab977b4659
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/read_changes_with_options.proto.bin differ
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.json b/sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.json
new file mode 100644
index 000000000000..edff217db760
--- /dev/null
+++ b/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
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.proto.bin
new file mode 100644
index 000000000000..7a6442b4975a
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/streaming_changes_API_with_options.proto.bin differ
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index 5da4b022942b..23403381ed8b 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++ b/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, U [...]
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)
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/ProtoToParsedPlanTestSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/ProtoToParsedPlanTestSuite.scala
index 02c5683e0276..506a75ca3b4c 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/ProtoToParsedPlanTestSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/ProtoToParsedPlanTestSuite.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.connect.config.Connect
import org.apache.spark.sql.connect.planner.SparkConnectPlanner
-import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, Identifier, InMemoryCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, Identifier, InMemoryChangelogCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -134,7 +134,7 @@ class ProtoToParsedPlanTestSuite
private val emptyProps: util.Map[String, String] = util.Collections.emptyMap()
private val analyzer = {
- val inMemoryCatalog = new InMemoryCatalog
+ val inMemoryCatalog = new InMemoryChangelogCatalog
inMemoryCatalog.initialize("primary", CaseInsensitiveStringMap.empty())
inMemoryCatalog.createNamespace(Array("tempdb"), emptyProps)
inMemoryCatalog.createTable(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
index bc01517e1c6f..d0d6bf1e8ec0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
@@ -28,7 +28,8 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql
import org.apache.spark.sql.Encoders
import org.apache.spark.sql.catalyst.DataSourceOptions
-import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.analysis.{RelationChanges, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.ChangelogInfoUtils
import org.apache.spark.sql.catalyst.csv.{CSVHeaderChecker, CSVOptions, UnivocityParser}
import org.apache.spark.sql.catalyst.expressions.ExprUtils
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JSONOptions}
@@ -320,6 +321,19 @@ class DataFrameReader private[sql](sparkSession: SparkSession)
new CaseInsensitiveStringMap(extraOptions.toMap.asJava)))
}
+ /** @inheritdoc */
+ def changes(tableName: String): DataFrame = {
+ require(tableName != null, "The table name can't be null")
+ assertNoSpecifiedSchema("changes")
+ val multipartIdentifier =
+ sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName)
+ val options = new CaseInsensitiveStringMap(extraOptions.toMap.asJava)
+ val changelogInfo = ChangelogInfoUtils.fromOptions(
+ options, sparkSession.sessionState.conf.sessionLocalTimeZone)
+ val relation = UnresolvedRelation(multipartIdentifier, options)
+ Dataset.ofRows(sparkSession, RelationChanges(relation, changelogInfo))
+ }
+
/** @inheritdoc */
override def text(path: String): DataFrame = super.text(path)
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 0469d77e2b15..f2a2e99fdc14 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
@@ -20,7 +20,7 @@ package org.apache.spark.sql.classic
import scala.jdk.CollectionConverters._
import org.apache.spark.annotation.{Evolving, Experimental}
-import org.apache.spark.sql.catalyst.analysis.{NamedStreamingRelation, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{ChangelogInfoUtils, NamedStreamingRelation, RelationChanges, UnresolvedRelation}
import org.apache.spark.sql.catalyst.plans.logical.UnresolvedDataSource
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
import org.apache.spark.sql.classic.ClassicConversions._
@@ -117,6 +117,20 @@ final class DataStreamReader private[sql](sparkSession: SparkSession)
Dataset.ofRows(sparkSession, plan)
}
+ /** @inheritdoc */
+ def changes(tableName: String): DataFrame = {
+ require(tableName != null, "The table name can't be null")
+ assertNoSpecifiedSchema("changes")
+ val identifier = sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName)
+ val options = new CaseInsensitiveStringMap(extraOptions.toMap.asJava)
+ val changelogInfo = ChangelogInfoUtils.fromOptions(
+ options, sparkSession.sessionState.conf.sessionLocalTimeZone)
+ val unresolved = UnresolvedRelation(identifier, options, isStreaming = true)
+ val changes = RelationChanges(unresolved, changelogInfo)
+ val plan = NamedStreamingRelation.withUserProvidedName(changes, userProvidedSourceName)
+ Dataset.ofRows(sparkSession, plan)
+ }
+
override protected def assertNoSpecifiedSchema(operation: String): Unit = {
if (userSpecifiedSchema.nonEmpty) {
throw QueryCompilationErrors.userSpecifiedSchemaUnsupportedError(operation)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala
index 31122e4ec167..58a2c7591ded 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogEndToEndSuite.scala
@@ -20,9 +20,12 @@ package org.apache.spark.sql.connector
import java.sql.Timestamp
import java.util
-import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.NamedStreamingRelation
+import org.apache.spark.sql.catalyst.streaming.UserProvided
import org.apache.spark.sql.connector.catalog._
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{LongType, StringType}
import org.apache.spark.unsafe.types.UTF8String
@@ -86,115 +89,586 @@ class ChangelogEndToEndSuite extends QueryTest with SharedSparkSession {
commitTimestamp)
}
- test("SQL CHANGES clause returns change data") {
+ // ---------- Batch: basic data retrieval ----------
+
+ test("changes() returns change data") {
catalog.addChangeRows(ident, Seq(
makeChangeRow(1L, "a", "insert", 1L, 1000000L),
- makeChangeRow(2L, "b", "insert", 1L, 1000000L),
- makeChangeRow(1L, "a", "delete", 2L, 2000000L)))
+ makeChangeRow(2L, "b", "delete", 2L, 2000000L)))
- val df = sql(
- s"SELECT * FROM $fullTableName " +
- "CHANGES FROM VERSION 1 TO VERSION 2")
- checkAnswer(df, Seq(
+ val expected = Seq(
Row(1L, "a", "insert", 1L, new Timestamp(1000L)),
- Row(2L, "b", "insert", 1L, new Timestamp(1000L)),
- Row(1L, "a", "delete", 2L, new Timestamp(2000L))))
+ Row(2L, "b", "delete", 2L, new Timestamp(2000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "2")
+ .changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1 TO VERSION 2"),
+ expected)
}
- test("SQL CHANGES clause with column projection") {
+ test("changes() with open-ended version range") {
catalog.addChangeRows(ident, Seq(
makeChangeRow(1L, "a", "insert", 1L, 1000000L),
- makeChangeRow(2L, "b", "insert", 1L, 1000000L)))
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L)))
+
+ val expected = Seq(
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)),
+ Row(3L, "c", "insert", 3L, new Timestamp(3000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read.option("startingVersion", "2").changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 2"),
+ expected)
+ }
- val df = sql(
- s"SELECT id, _change_type FROM $fullTableName " +
- "CHANGES FROM VERSION 1")
- checkAnswer(df, Seq(
- Row(1L, "insert"),
- Row(2L, "insert")))
+ test("changes() returns empty result when no changes exist") {
+ // DataFrame API
+ val dfApi = spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "5")
+ .changes(fullTableName)
+ assert(dfApi.collect().isEmpty)
+ assert(dfApi.schema.fieldNames === Array(
+ "id", "data", "_change_type",
+ "_commit_version", "_commit_timestamp"))
+
+ // SQL
+ val dfSql = sql(
+ s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1 TO VERSION 5")
+ assert(dfSql.collect().isEmpty)
+ assert(dfSql.schema.fieldNames.contains("_change_type"))
}
- test("SQL CHANGES clause with filter") {
+ // ---------- Batch: projection, filter, aggregation ----------
+
+ test("changes() select CDC metadata columns") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "delete", 2L, 2000000L)))
+
+ val expected = Seq(
+ Row(1L, "insert", 1L),
+ Row(2L, "delete", 2L))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read.option("startingVersion", "1").changes(fullTableName)
+ .select("id", "_change_type", "_commit_version"),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT id, _change_type, _commit_version FROM $fullTableName " +
+ "CHANGES FROM VERSION 1"),
+ expected)
+ }
+
+ test("changes() with projection and filter") {
catalog.addChangeRows(ident, Seq(
makeChangeRow(1L, "a", "insert", 1L, 1000000L),
makeChangeRow(2L, "b", "insert", 1L, 1000000L),
- makeChangeRow(3L, "c", "insert", 2L, 2000000L)))
+ makeChangeRow(1L, "a2", "insert", 2L, 2000000L)))
+
+ val expected = Seq(Row(1L, "a2"))
- val df = sql(
- s"SELECT id, data FROM $fullTableName " +
- "CHANGES FROM VERSION 1 TO VERSION 2 " +
- "WHERE _commit_version = 1")
- checkAnswer(df, Seq(
- Row(1L, "a"),
- Row(2L, "b")))
+ // DataFrame API
+ checkAnswer(
+ spark.read.option("startingVersion", "1").changes(fullTableName)
+ .filter("_commit_version = 2").select("id", "data"),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT id, data FROM $fullTableName CHANGES FROM VERSION 1 " +
+ "WHERE _commit_version = 2"),
+ expected)
+ }
+
+ test("changes() with aggregation on change types") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 1L, 1000000L),
+ makeChangeRow(1L, "a", "delete", 2L, 2000000L)))
+
+ val expected = Seq(
+ Row("insert", 2L),
+ Row("delete", 1L))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read.option("startingVersion", "1").changes(fullTableName)
+ .groupBy("_change_type").count(),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT _change_type, count(*) FROM $fullTableName " +
+ "CHANGES FROM VERSION 1 GROUP BY _change_type"),
+ expected)
}
test("schema includes CDC metadata columns") {
- val df = sql(
- s"SELECT * FROM $fullTableName " +
- "CHANGES FROM VERSION 1 TO VERSION 5")
- val colNames = df.schema.fieldNames
- assert(colNames === Array(
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ val dfApi = spark.read.option("startingVersion", "1").changes(fullTableName)
+ assert(dfApi.schema.fieldNames === Array(
"id", "data", "_change_type",
"_commit_version", "_commit_timestamp"))
+
+ // SQL
+ val dfSql = sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1")
+ assert(dfSql.schema.fieldNames === Array(
+ "id", "data", "_change_type",
+ "_commit_version", "_commit_timestamp"))
+ }
+
+ // ---------- Batch: version range filtering ----------
+
+ test("changes() version range filters correctly") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L),
+ makeChangeRow(4L, "d", "insert", 4L, 4000000L)))
+
+ val expected = Seq(
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)),
+ Row(3L, "c", "insert", 3L, new Timestamp(3000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "2")
+ .option("endingVersion", "3")
+ .changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 2 TO VERSION 3"),
+ expected)
}
- test("empty change data returns no rows") {
- val df = sql(
- s"SELECT * FROM $fullTableName " +
- "CHANGES FROM VERSION 1 TO VERSION 5")
- assert(df.collect().isEmpty)
- val colNames = df.schema.fieldNames
- assert(colNames.contains("_change_type"))
+ // ---------- Batch: bound inclusivity ----------
+
+ test("changes() default bounds are inclusive") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L)))
+
+ val expected = Seq(
+ Row(1L, "a", "insert", 1L, new Timestamp(1000L)),
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)),
+ Row(3L, "c", "insert", 3L, new Timestamp(3000L)))
+
+ // DataFrame API - default (both inclusive)
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "3")
+ .changes(fullTableName),
+ expected)
+
+ // SQL - default (both inclusive)
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1 TO VERSION 3"),
+ expected)
+
+ // SQL - explicit INCLUSIVE keywords
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName " +
+ "CHANGES FROM VERSION 1 INCLUSIVE TO VERSION 3 INCLUSIVE"),
+ expected)
}
- test("streaming CDC reads change data via STREAM CHANGES") {
+ test("changes() with startingBoundInclusive=false") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L)))
+
+ // Exclusive start: version 1 excluded, versions 2 and 3 included
+ val expected = Seq(
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)),
+ Row(3L, "c", "insert", 3L, new Timestamp(3000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "3")
+ .option("startingBoundInclusive", "false")
+ .changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName " +
+ "CHANGES FROM VERSION 1 EXCLUSIVE TO VERSION 3"),
+ expected)
+ }
+
+ test("changes() with endingBoundInclusive=false") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L)))
+
+ // Exclusive end: versions 1 and 2 included, version 3 excluded
+ val expected = Seq(
+ Row(1L, "a", "insert", 1L, new Timestamp(1000L)),
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "3")
+ .option("endingBoundInclusive", "false")
+ .changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName " +
+ "CHANGES FROM VERSION 1 TO VERSION 3 EXCLUSIVE"),
+ expected)
+ }
+
+ test("changes() with both bounds exclusive") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 2L, 2000000L),
+ makeChangeRow(3L, "c", "insert", 3L, 3000000L)))
+
+ // Both exclusive: only version 2 included
+ val expected = Seq(
+ Row(2L, "b", "insert", 2L, new Timestamp(2000L)))
+
+ // DataFrame API
+ checkAnswer(
+ spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "3")
+ .option("startingBoundInclusive", "false")
+ .option("endingBoundInclusive", "false")
+ .changes(fullTableName),
+ expected)
+
+ // SQL
+ checkAnswer(
+ sql(s"SELECT * FROM $fullTableName " +
+ "CHANGES FROM VERSION 1 EXCLUSIVE TO VERSION 3 EXCLUSIVE"),
+ expected)
+ }
+
+ // ---------- Batch: CDC options ----------
+
+ test("changes() default deduplication mode is dropCarryovers") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ spark.read.option("startingVersion", "1").changes(fullTableName).collect()
+ val info1 = catalog.lastChangelogInfo.get
+ assert(info1.deduplicationMode() === ChangelogInfo.DeduplicationMode.DROP_CARRYOVERS)
+ assert(info1.computeUpdates() === false)
+
+ // SQL (no WITH clause = defaults)
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1").collect()
+ val info2 = catalog.lastChangelogInfo.get
+ assert(info2.deduplicationMode() === ChangelogInfo.DeduplicationMode.DROP_CARRYOVERS)
+ assert(info2.computeUpdates() === false)
+ }
+
+ test("changes() with deduplicationMode none") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ spark.read
+ .option("startingVersion", "1")
+ .option("deduplicationMode", "none")
+ .changes(fullTableName)
+ .collect()
+ assert(catalog.lastChangelogInfo.get.deduplicationMode() ===
+ ChangelogInfo.DeduplicationMode.NONE)
+
+ // SQL
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1 " +
+ "WITH (deduplicationMode = 'none')").collect()
+ assert(catalog.lastChangelogInfo.get.deduplicationMode() ===
+ ChangelogInfo.DeduplicationMode.NONE)
+ }
+
+ test("changes() passes deduplicationMode and computeUpdates to catalog") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ spark.read
+ .option("startingVersion", "1")
+ .option("deduplicationMode", "netChanges")
+ .option("computeUpdates", "true")
+ .changes(fullTableName)
+ .collect()
+ val info1 = catalog.lastChangelogInfo.get
+ assert(info1.deduplicationMode() === ChangelogInfo.DeduplicationMode.NET_CHANGES)
+ assert(info1.computeUpdates() === true)
+
+ // SQL
+ sql(s"SELECT * FROM $fullTableName CHANGES FROM VERSION 1 " +
+ "WITH (deduplicationMode = 'netChanges', computeUpdates = 'true')").collect()
+ val info2 = catalog.lastChangelogInfo.get
+ assert(info2.deduplicationMode() === ChangelogInfo.DeduplicationMode.NET_CHANGES)
+ assert(info2.computeUpdates() === true)
+ }
+
+ // ---------- Batch: timestamp range ----------
+
+ test("changes() with timestamp range") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ spark.read
+ .option("startingTimestamp", "2024-01-01 00:00:00")
+ .option("endingTimestamp", "2024-12-31 23:59:59")
+ .changes(fullTableName)
+ .collect()
+ assert(catalog.lastChangelogInfo.get.range()
+ .isInstanceOf[ChangelogRange.TimestampRange])
+
+ // SQL
+ sql(s"SELECT * FROM $fullTableName " +
+ "CHANGES FROM TIMESTAMP '2024-01-01 00:00:00' " +
+ "TO TIMESTAMP '2024-12-31 23:59:59'").collect()
+ assert(catalog.lastChangelogInfo.get.range()
+ .isInstanceOf[ChangelogRange.TimestampRange])
+ }
+
+ // ---------- Batch: error cases ----------
+
+ test("changes() rejects user-specified schema") {
+ val e = intercept[AnalysisException] {
+ spark.read
+ .schema("id LONG, data STRING")
+ .option("startingVersion", "1")
+ .changes(fullTableName)
+ }
+ assert(e.getMessage.contains("changes"))
+ }
+
+ // ---------- Streaming: basic data retrieval ----------
+
+ test("streaming changes() returns change data") {
catalog.addChangeRows(ident, Seq(
makeChangeRow(1L, "a", "insert", 1L, 1000000L),
makeChangeRow(2L, "b", "insert", 1L, 1000000L),
makeChangeRow(1L, "a", "delete", 2L, 2000000L)))
- val streamDf = sql(
- s"SELECT * FROM STREAM $fullTableName CHANGES FROM VERSION 1")
+ val expected = Seq(
+ Row(1L, "a", "insert", 1L, new Timestamp(1000L)),
+ Row(2L, "b", "insert", 1L, new Timestamp(1000L)),
+ Row(1L, "a", "delete", 2L, new Timestamp(2000L)))
- val queryName = "cdc_streaming_test"
- val query = streamDf.writeStream
- .format("memory")
- .queryName(queryName)
- .start()
+ // DataFrame API
+ val dfApiStream = spark.readStream
+ .option("startingVersion", "1")
+ .changes(fullTableName)
+ val q1 = dfApiStream.writeStream
+ .format("memory").queryName("cdc_stream_df").start()
try {
- query.processAllAvailable()
- val result = spark.sql(s"SELECT * FROM $queryName")
- checkAnswer(result, Seq(
- Row(1L, "a", "insert", 1L, new Timestamp(1000L)),
- Row(2L, "b", "insert", 1L, new Timestamp(1000L)),
- Row(1L, "a", "delete", 2L, new Timestamp(2000L))))
+ q1.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_df"), expected)
} finally {
- query.stop()
+ q1.stop()
+ }
+
+ // SQL
+ val sqlStream = sql(
+ s"SELECT * FROM STREAM $fullTableName CHANGES FROM VERSION 1")
+ val q2 = sqlStream.writeStream
+ .format("memory").queryName("cdc_stream_sql").start()
+ try {
+ q2.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_sql"), expected)
+ } finally {
+ q2.stop()
}
}
- test("streaming CDC reads change data via STREAM CHANGES FROM VERSION 2") {
+ test("streaming changes() with startingVersion filters data") {
catalog.addChangeRows(ident, Seq(
makeChangeRow(1L, "a", "insert", 1L, 1000000L),
makeChangeRow(2L, "b", "insert", 1L, 1000000L),
makeChangeRow(1L, "a", "delete", 2L, 2000000L)))
- val streamDf = sql(
+ val expected = Seq(
+ Row(1L, "a", "delete", 2L, new Timestamp(2000L)))
+
+ // DataFrame API
+ val dfApiStream = spark.readStream
+ .option("startingVersion", "2")
+ .changes(fullTableName)
+ val q1 = dfApiStream.writeStream
+ .format("memory").queryName("cdc_stream_v2_df").start()
+ try {
+ q1.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_v2_df"), expected)
+ } finally {
+ q1.stop()
+ }
+
+ // SQL
+ val sqlStream = sql(
s"SELECT * FROM STREAM $fullTableName CHANGES FROM VERSION 2")
+ val q2 = sqlStream.writeStream
+ .format("memory").queryName("cdc_stream_v2_sql").start()
+ try {
+ q2.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_v2_sql"), expected)
+ } finally {
+ q2.stop()
+ }
+ }
+
+ test("streaming changes() with projection and filter") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L),
+ makeChangeRow(2L, "b", "insert", 1L, 1000000L),
+ makeChangeRow(3L, "c", "insert", 2L, 2000000L)))
+
+ val expected = Seq(Row(1L, "a"), Row(2L, "b"))
+
+ // DataFrame API
+ val dfApiStream = spark.readStream
+ .option("startingVersion", "1")
+ .changes(fullTableName)
+ .filter("_commit_version = 1")
+ .select("id", "data")
+ val q1 = dfApiStream.writeStream
+ .format("memory").queryName("cdc_stream_proj_df").start()
+ try {
+ q1.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_proj_df"), expected)
+ } finally {
+ q1.stop()
+ }
+
+ // SQL
+ val sqlStream = sql(
+ s"SELECT id, data FROM STREAM $fullTableName " +
+ "CHANGES FROM VERSION 1 WHERE _commit_version = 1")
+ val q2 = sqlStream.writeStream
+ .format("memory").queryName("cdc_stream_proj_sql").start()
+ try {
+ q2.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_proj_sql"), expected)
+ } finally {
+ q2.stop()
+ }
+ }
+
+ // ---------- Streaming: CDC options ----------
- val queryName = "cdc_streaming_test_v2"
- val query = streamDf.writeStream
- .format("memory")
- .queryName(queryName)
- .start()
+ test("streaming changes() passes deduplicationMode and computeUpdates to catalog") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ // DataFrame API
+ val dfApiStream = spark.readStream
+ .option("startingVersion", "1")
+ .option("deduplicationMode", "netChanges")
+ .option("computeUpdates", "true")
+ .changes(fullTableName)
+ val q1 = dfApiStream.writeStream
+ .format("memory").queryName("cdc_stream_opts_df").start()
+ try {
+ q1.processAllAvailable()
+ val info1 = catalog.lastChangelogInfo.get
+ assert(info1.deduplicationMode() === ChangelogInfo.DeduplicationMode.NET_CHANGES)
+ assert(info1.computeUpdates() === true)
+ } finally {
+ q1.stop()
+ }
+
+ // SQL
+ val sqlStream = sql(
+ s"SELECT * FROM STREAM $fullTableName CHANGES FROM VERSION 1 " +
+ "WITH (deduplicationMode = 'netChanges', computeUpdates = 'true')")
+ val q2 = sqlStream.writeStream
+ .format("memory").queryName("cdc_stream_opts_sql").start()
try {
- query.processAllAvailable()
- val result = spark.sql(s"SELECT * FROM $queryName")
- checkAnswer(result, Seq(
- Row(1L, "a", "delete", 2L, new Timestamp(2000L))))
+ q2.processAllAvailable()
+ val info2 = catalog.lastChangelogInfo.get
+ assert(info2.deduplicationMode() === ChangelogInfo.DeduplicationMode.NET_CHANGES)
+ assert(info2.computeUpdates() === true)
} finally {
- query.stop()
+ q2.stop()
+ }
+ }
+
+ // ---------- Streaming: .name() API ----------
+
+ test("streaming changes() supports .name() API with source evolution enabled") {
+ catalog.addChangeRows(ident, Seq(
+ makeChangeRow(1L, "a", "insert", 1L, 1000000L)))
+
+ val expected = Seq(
+ Row(1L, "a", "insert", 1L, new Timestamp(1000L)))
+
+ withSQLConf(SQLConf.ENABLE_STREAMING_SOURCE_EVOLUTION.key -> "true") {
+ val stream = spark.readStream
+ .name("my_cdc_source")
+ .option("startingVersion", "1")
+ .changes(fullTableName)
+
+ // Verify the logical plan contains NamedStreamingRelation with the user-provided name
+ val plan = stream.queryExecution.logical
+ val namedRelations = plan.collect {
+ case n: NamedStreamingRelation => n
+ }
+ assert(namedRelations.size === 1)
+ assert(namedRelations.head.sourceIdentifyingName === UserProvided("my_cdc_source"))
+
+ val q = stream.writeStream
+ .format("memory").queryName("cdc_stream_named").start()
+ try {
+ q.processAllAvailable()
+ checkAnswer(spark.sql("SELECT * FROM cdc_stream_named"), expected)
+ } finally {
+ q.stop()
+ }
+ }
+ }
+
+ // ---------- Streaming: error cases ----------
+
+ test("streaming changes() rejects user-specified schema") {
+ val e = intercept[AnalysisException] {
+ spark.readStream
+ .schema("id LONG, data STRING")
+ .option("startingVersion", "1")
+ .changes(fullTableName)
}
+ assert(e.getMessage.contains("changes"))
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
index a30db17356bc..143ca3fe1a09 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/ChangelogResolutionSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.connector
import java.util
import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
import org.apache.spark.sql.connector.catalog._
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.connector.catalog.ChangelogRange
@@ -109,6 +110,30 @@ class ChangelogResolutionSuite extends QueryTest with SharedSparkSession {
e.getMessage.contains("nonexistent"))
}
+ test("DataFrame API - changes() resolves correctly") {
+ val df = spark.read
+ .option("startingVersion", "1")
+ .option("endingVersion", "5")
+ .changes(s"$cdcCatalogName.test_table")
+ val analyzed = df.queryExecution.analyzed
+ val dsv2Relations = analyzed.collect {
+ case r: DataSourceV2Relation => r
+ }
+ assert(dsv2Relations.length == 1)
+ assert(dsv2Relations.head.table.isInstanceOf[ChangelogTable])
+ }
+
+ test("DataFrame API - changes() on catalog without CDC throws") {
+ checkError(
+ intercept[AnalysisException] {
+ spark.read
+ .option("startingVersion", "1")
+ .changes(s"$noCdcCatalogName.test_table")
+ },
+ condition = "UNSUPPORTED_FEATURE.CHANGE_DATA_CAPTURE",
+ parameters = Map("catalogName" -> noCdcCatalogName))
+ }
+
test("CHANGES clause - schema includes CDC metadata columns") {
val df = sql(
s"SELECT * FROM $cdcCatalogName.test_table CHANGES FROM VERSION 1 TO VERSION 5")
@@ -120,6 +145,43 @@ class ChangelogResolutionSuite extends QueryTest with SharedSparkSession {
assert(colNames.contains("_commit_timestamp"))
}
+ test("DataStreamReader - changes() rejects user-specified schema") {
+ val e = intercept[AnalysisException] {
+ import org.apache.spark.sql.types.StructType
+ spark.readStream
+ .schema(new StructType().add("id", LongType))
+ .changes(s"$cdcCatalogName.test_table")
+ }
+ assert(e.getMessage.contains("changes"))
+ }
+
+ test("DataStreamReader - changes() resolves to StreamingRelationV2 with ChangelogTable") {
+ val df = spark.readStream
+ .option("startingVersion", "1")
+ .changes(s"$cdcCatalogName.test_table")
+ val analyzed = df.queryExecution.analyzed
+ val streamRelations = analyzed.collect {
+ case r: StreamingRelationV2 => r
+ }
+ assert(streamRelations.length == 1)
+ assert(streamRelations.head.table.isInstanceOf[ChangelogTable])
+ val colNames = df.schema.fieldNames
+ assert(colNames.contains("_change_type"))
+ assert(colNames.contains("_commit_version"))
+ assert(colNames.contains("_commit_timestamp"))
+ }
+
+ test("DataStreamReader - changes() on catalog without CDC throws") {
+ checkError(
+ intercept[AnalysisException] {
+ spark.readStream
+ .option("startingVersion", "1")
+ .changes(s"$noCdcCatalogName.test_table")
+ },
+ condition = "UNSUPPORTED_FEATURE.CHANGE_DATA_CAPTURE",
+ parameters = Map("catalogName" -> noCdcCatalogName))
+ }
+
test("CHANGES clause on CTE relation throws") {
checkError(
intercept[AnalysisException] {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org