[Delta] Add Changelog support - #6794
Conversation
| Configuration hadoopConf, | ||
| SQLConf sqlConf, | ||
| boolean isCDCRead) { | ||
| CdcReadMode cdcReadMode) { |
There was a problem hiding this comment.
Handling different cdcReadModes
Problem. Before this PR
createDeltaParquetReaderFactorytookboolean isCDCRead. Two callers passedtruefor completely different semantics:SparkMicroBatchStreamfor streaming CDC (PartitionUtils owns schema augmentation +CDCReadFunctionwrap), andDeltaChangelogBatchfor Auto-CDF (outerCDCPartitionReaderFactoryowns CDC tail injection). The method had to disambiguate by inspectingreadDataSchemafor a_metadatafield — coupling a CDC decision to a row-tracking schema check.Options.
- Keep
isCDCReadboolean, continue the_metadata-presence heuristic — fragile; breaks if row tracking ever appears without Auto-CDF.- Split into two factory methods — duplicates the shared DV / row-tracking / format wiring.
- Single factory with
CdcReadMode { NONE, STREAMING, BATCH_CHANGELOG }.Chosen. Option 3. Dispatch contract is explicit at the call site, shared wiring stays in one place, and a future
BATCH_CHANGELOG_NETCHANGESmode (post apache/spark#55637) drops in without changing call signatures.
Like in the comment below I'm a bit confused on how batch/streaming should live with each other. Advice is appreciated.
| // CDCPartitionReaderFactory injects the CDC tail columns as constants instead. | ||
| Optional<CDCSchemaContext> cdcSchemaContext = | ||
| isCDCRead | ||
| cdcReadMode.injectsCdcAtReaderLevel() |
There was a problem hiding this comment.
Streaming CDC and Auto-CDF coexistence
Problem. The streaming CDC entrypoint shipped in #6359 mutates
readDataSchemawith CDC tail columns (_change_type,_commit_version,_commit_timestamp) and wraps the reader withCDCReadFunctionto inject those columns per row. Auto-CDF (DeltaChangelogBatch) wraps its delegate withCDCPartitionReaderFactory, which injects the same columns as per-partition constants (one value per CDC file change). If both fire, the schema sees the CDC tail twice and projection blows up withGenericInternalRow cannot be cast to Long.Options.
- Tear down the OSS streaming-CDC path and route streaming through Auto-CDF — invasive, breaks streaming tests until
toMicroBatchStreamlands.- Heuristic: skip the inner wrap when
_metadatais inreadDataSchema— works incidentally; same fragility called out above.- Explicit
CdcReadModedispatch:injectsCdcAtReaderLevel()is true only forSTREAMING.Chosen. Option 3. Both pathways stay live; the connector picks the right wrapper from the mode at the call site. Streaming behavior is unchanged from #6359.
Tbh, I was a bit overwhelmed on how to handle the streaming co-existence with our batch approach. I'm not sure if the chosen path is the correct one.
There was a problem hiding this comment.
Following on the comment on CdcReadMode: Consider Streaming write-time CDF a completely unrelated feature.
It's not possible to read both read-time CDF and write-time CDF in the same query, so it's not possible for the problem you describe to occur.
The important part here will be to make the distinction between the two features really clear to avoid confusion. This can simply be renaming isCDCRead to isWriteTimeCDCRead.
DeltaChangelogBatch will call createDeltaParquetReaderFactory passing isWriteTimeCDCRead=false
There was a problem hiding this comment.
Done. Renamed; DeltaChangelogBatch passes isWriteTimeCDCRead=false. Javadoc updated to make the read-time vs write-time distinction explicit.
| * classes (SparkTable, DeltaChangelog, SnapshotManagerFactory, V2SchemaUtils) which | ||
| * the sparkV1 module that hosts [[AbstractDeltaCatalog]] cannot depend on. | ||
| */ | ||
| abstract class DeltaCatalogChangelogSupport extends AbstractDeltaCatalog { |
There was a problem hiding this comment.
Why the loadChangelog override lives here, not in AbstractDeltaCatalog
**Problem.** The 1:1 port placed `loadChangelog` directly inside `AbstractDeltaCatalog`. That override references sparkV2 types (`SparkTable`, `DeltaChangelog`, `SnapshotManagerFactory`, `V2SchemaUtils`) — but `AbstractDeltaCatalog` lives in sparkV1, and sparkV2 already depends on sparkV1Filtered → sparkV1 transitively. A `sparkV1 → sparkV2` edge would close the dependency cycle and the module graph stops being a DAG.Options.
- Move the sparkV2 changelog classes (
DeltaChangelog,SparkTable,SnapshotManagerFactory, ...) into sparkV1 — pulls Kernel-based reader code into the V1 module against the Hybrid pattern introduced in [Spark] Hybrid Delta connector combining V1(default) and V2 connectors #5726. But I think we want to keep changes related to DsV2 to in V2, right?- Reflection lookup from sparkV1 — avoids compile-time deps but hides API contracts and is invisible to IDE tooling.
- Implement the override one level higher, in
spark-unified(which already depends on both V1 and V2), via an intermediate class.Chosen. Option 3 — this class. It mirrors how
DeltaCatalog.javaalready routesloadCatalogTable/loadPathTablebetween V1 and V2 fromspark-unified. The only structural difference from the internal-source port is where the override is defined; the implementation logic is unchanged.
I'm not sure if I set myself to limitations that do not exist. When I see V1 vs V2 I always think of clear separation. Is that the case? Or could I simply have implemented stuff in the sparkV1 world?
| vr.endingBoundInclusive()) | ||
| case tr: TimestampRange => | ||
| val catalogTable = resolveDeltaCatalogTable(ident) | ||
| val deltaLog = DeltaLog.forTable(spark, catalogTable) |
There was a problem hiding this comment.
Direct access to the delta log at this level of abstraction is a no-go imo.
buildChangelog is also doing a bit too much. For now, what you need is really just exposing the table schema. Later on surfacing containsCarryoverRows / ... may turn to be trickier and require more state to be loaded, but for now all the pre processing is a bit much.
A different structure will be simpler and delegate work to a later stage: have DeltaChangelog wrap a SparkTable (similar to https://docs.google.com/document/d/1NpJ9Q4wvU-tp3r67RC8MKnm-0J4Y0n7mSOC6iiiBJNo/edit?tab=t.0#heading=h.2gjhdgzfrm6g).
I would also just store the changelogInfo in DeltaChangelog, without trying to pre-process it. Making ChangelogSupport extension on the catalog really straightforward
We only support reading CDC here on DSv2 tables. That means SparkTable for Delta. You should reject reads on DeltaTableV2 below (which is a V1 table really, don't believe its name)
There was a problem hiding this comment.
Rejecting DeltaTableV2 reads with new Error DELTA_CHANGELOG_REQUIRES_V2_TABLE. Removed the direct DeltaLog access. Wrapping the SparkTable by deferring to the read path (DeltaChangelogScanBuilder.build and DeltaChangelogBatch.planInputPartitions)
| val engine = DefaultEngine.create(deltaLog.newDeltaHadoopConf()) | ||
| val snapshotManager = | ||
| SnapshotManagerFactory.create(tablePath, engine, Optional.of(catalogTable)) | ||
|
|
||
| // The latest snapshot is needed to default the ending version when the user did not | ||
| // specify one, and to surface a clear "start > latest" error before issuing a snapshot | ||
| // load that would otherwise fail with a low-level kernel error. | ||
| val latestVersion = snapshotManager.loadLatestSnapshot().getVersion() |
There was a problem hiding this comment.
You'll want to use the snapshotManager interface to resolve timestamps -> versions also, which is currently done by directly calling in DeltaLog.
E.g. startVersion probably through getActiveCommitAtTime.
Main point: you shouldn't create a snapshotManager yourself, and rely on the resolved SparkTable to do this. I'm not sure how well time-travel is implemented at this stage, so you may need to extend it.
E.g. we may want to start by resolving the SparkTable directly at startVersion / startTimestamp by calling loadTable(ident, version/timestamp)
There was a problem hiding this comment.
Currently no SnapshotManager is constructed inside the trait or DeltaChangelog. But I'm not using your approach through my own SnapshotManager (yet). I haven't yet switched to resolving the SparkTable directly at startVersion / startTimestamp via loadTable(ident, version/timestamp). The SparkTable resolved by the catalog gives us access to the snapshot manager for arbitrary version / timestamp lookups, so pinning the table to startVersion didn't seem necessary. I'll take a look at this on Monday.
| // CDCPartitionReaderFactory injects the CDC tail columns as constants instead. | ||
| Optional<CDCSchemaContext> cdcSchemaContext = | ||
| isCDCRead | ||
| cdcReadMode.injectsCdcAtReaderLevel() |
There was a problem hiding this comment.
Following on the comment on CdcReadMode: Consider Streaming write-time CDF a completely unrelated feature.
It's not possible to read both read-time CDF and write-time CDF in the same query, so it's not possible for the problem you describe to occur.
The important part here will be to make the distinction between the two features really clear to avoid confusion. This can simply be renaming isCDCRead to isWriteTimeCDCRead.
DeltaChangelogBatch will call createDeltaParquetReaderFactory passing isWriteTimeCDCRead=false
johanl-db
left a comment
johanl-db
left a comment
There was a problem hiding this comment.
Posting comments to be addressed in a follow up.
There's only one that needs to be addressed now (marked as such)
This core implementation is solid, I would recommend merging it, then:
PR 2: implementing support for tables with DVs
PR 3: updating the existing CDF tests to also run against that implementation
That way we get changes in before the next cut and have good confidence that things are working. We can use remaining time before or after the cut to do follow ups to improve if needed
| import org.apache.spark.unsafe.types.UTF8String; | ||
| import scala.Tuple2; | ||
|
|
||
| public class DeltaChangelogBatch implements Batch { |
There was a problem hiding this comment.
not super important, ss a follow up: it seems like DeltaChangelogBatch and other classes below don't need to be shims and can be in the main code paths since (at least on a quick look), they don't seem to use anything specific to Spark 4.2
Shims by definition tend to be rather small, only overriding the interfaces that actually change between versions.
Alternatively, this could live under spark-unified or another v2 specific location assuming there's a way to only build against Spark 4.2 (somehow doubt that)
| * | ||
| * <p>The catalog-driven `TableCatalog.loadChangelog` entrypoint and its supporting types | ||
| * (`Changelog`, `ChangelogInfo`, `ChangelogRange`) were introduced in Spark 4.2 via | ||
| * SPARK-56685. They do not exist in Spark 4.0/4.1, so the Auto-CDF wiring is compiled in only |
There was a problem hiding this comment.
nit: read-time CDF instead of Auto-CDF
| // Pre-check catches schema drift between start and end. The per-commit loop below catches | ||
| // in-range Metadata commits. | ||
| StructType startSchema = SchemaUtils.convertKernelSchemaToSparkSchema(snapshot.getSchema()); | ||
| if (!startSchema.equals(dataSchema)) { |
There was a problem hiding this comment.
As a follow up:
We do support reading changes across specific schema changes: adding a new column, relaxing nullability of a field.
These are additive schema changes: then end schema can still be used to read the table.
A utility like SchemaUtils.isReadCompatible would be better suited.
Also, using equals will consider the schema to differ if anything changed in the schema, including internal metadata attached to struct fields. isReadCompatible correctly ignores internal metadata.
Also might be worth calling dataSchema endDataSchema to disambiguate
| if (!commitSchema.equals(dataSchema)) { | ||
| DeltaErrors.throwChangelogSchemaChangeInRange(commit.getVersion()); | ||
| } | ||
| String rtValue = md.getConfiguration().get("delta.enableRowTracking"); |
There was a problem hiding this comment.
nit: there is a conf defined in DeltaConfig for row tracking, instead of using a literal string. And ideally a way of getting that as a boolean already, with validation backed in
| DeltaErrors.throwChangelogSchemaChangeInRange(commit.getVersion()); | ||
| } | ||
| String rtValue = md.getConfiguration().get("delta.enableRowTracking"); | ||
| // Absent key means the prior value persists (no change at this commit). |
There was a problem hiding this comment.
This comment must be addressed now:
I don't think that's true: the new Metadata superseds the previous one: all configs are explicitly listed
|
@SanJSp There are branch of java code under scala-shim folder/ can you make a seperate pr to move it to java-shim folder? |
|
Also why can't spark-unified/src/main/scala-shims/spark-4.0/org/apache/spark/sql/delta/catalog/ChangelogSupport.scala be in Java |
DeltaChangelogDirectBatchExecutionTest and DeltaChangelogCatalogIntegrationTest (from delta-io#6794) hit a ClassCastException on Spark 4.2 SNAPSHOT: DeltaCatalog cannot be cast to AbstractDeltaCatalog. This is a sparkV2 classpath issue with the hybrid DeltaCatalog from spark-unified — not related to Delta's Spark 4.2 shims. Gate on SNAPSHOT until the Changelog classpath is fixed.
DeltaChangelogDirectBatchExecutionTest and DeltaChangelogCatalogIntegrationTest (from delta-io#6794) hit a ClassCastException on Spark 4.2 SNAPSHOT: DeltaCatalog cannot be cast to AbstractDeltaCatalog. This is a sparkV2 classpath issue with the hybrid DeltaCatalog from spark-unified — not related to Delta's Spark 4.2 shims. Gate on SNAPSHOT until the Changelog classpath is fixed.
DeltaChangelogDirectBatchExecutionTest and DeltaChangelogCatalogIntegrationTest (from delta-io#6794) hit a ClassCastException on Spark 4.2 SNAPSHOT: DeltaCatalog cannot be cast to AbstractDeltaCatalog. This is a sparkV2 classpath issue with the hybrid DeltaCatalog from spark-unified — not related to Delta's Spark 4.2 shims. Gate on SNAPSHOT until the Changelog classpath is fixed.
<!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md 2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP] Your PR title ...'. 3. Be sure to keep the PR description updated to reflect all changes. 4. Please write your PR title to summarize what this PR proposes. 5. If possible, provide a concise example to reproduce the issue for a faster review. 6. If applicable, include the corresponding issue number in the PR title and link it in the body. --> #### Which Delta project/connector is this regarding? - [x] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description Follow-up to [#6794](#6794). Adds Deletion Vector (DV) support to the V2 changelog read path so `CHANGES FROM VERSION x TO VERSION y` returns correct results on tables with `delta.enableDeletionVectors=true`. **In short,** I apply change_type delete to every removeFile and change_type insert to every row in addFile per commit. A carry-over pair (delete+insert) will be filtered out as carry-over, but with the DVs we apply the DV as filter to the AddFile, leading to delete rows without insert partner, which represent actual delets. ### Implementation `DeltaChangelogBatch.java`: - `CDCInputPartition` carries a per-file `deletionVectorInfo: String` (the base64-serialized DV descriptor, or `null` when the file has no DV). - `planInputPartitions` extracts the DV per `AddFile`/`RemoveFile` via `DeletionVectorDescriptor::serializeToBase64` and threads it onto the partition. - `createReader` uses the new public `PartitionUtils.buildDvMetadata(...)` helper to set `FILE_ROW_INDEX_FILTER_ID_ENCODED` and `FILE_ROW_INDEX_FILTER_TYPE` together (with `RowIndexFilterType.IF_CONTAINED`) on the per-file metadata, so the existing `DropMarkedRowsFilter` machinery picks up the DV and produces the correct `delete` change rows. - Wrapper `RuntimeException` messages now include the inner cause's `getMessage()` so `AnalysisException` error classes (e.g. `DELTA_CHANGELOG_*`) reach the user instead of being hidden behind a generic "Failed to process CDC commit actions". `PartitionUtils.java`: - New public 1-arg `buildDvMetadata(String dvBase64)` overload that takes the base64 string directly and uses `IF_CONTAINED` semantics. Returns `java.util.Map` for ergonomic iteration from the changelog caller. Existing private overloads are unchanged. ### Tests `DeltaChangelogDirectBatchExecutionTest.java`: - `testDirectBatchExecutionWithExplicitExpectedRows` and `testUpdateProducesPairedDeleteAndInsert` are now parameterized via `@ParameterizedTest @valuesource(booleans = {false, true})` so each runs once with DVs off (rewrite-on-delete) and once with DVs on (DV-on-delete). The expected row sets are identical, so the same assertions cover both physical layouts. `DeltaChangelogDvTest.java` (new): DV-specific scenarios that don't fit the parameterized variants. - `test_mixedDvDelete_perFileBranching`: two `AddFile`s in one commit, a single `DELETE` touching both; asserts the per-file DV branch in `planInputPartitions` sets the DV constants independently per file. - `test_multiVersionDvDeletes_perCommitIsolation`: two `DELETE` commits in sequence on a DV-enabled table; asserts Catalyst's batch CDC post-processor partitions on `(row_id, version)` so each commit's carry-overs collapse independently. ## Note on Spark 4.2 lane The Spark 4.2 lane is non-blocking (`continue-on-error: true`); a follow-up PR ([#6830](#6830)) relocates these tests to `spark-unified/test` where the in-tree `DeltaCatalog` is on the classpath, making the 4.2 lane green for these tests too. ## How was this patch tested? Locally against Spark 4.0 / 4.1 (default lane, green) and Spark 4.2 (with PR #6830 applied to verify the tests pass end-to-end). 25 / 25 changelog tests pass with both PRs combined. ## Does this PR introduce _any_ user-facing changes? The V2 changelog path is gated behind `spark.databricks.delta.changelogV2.enabled` and not yet on a released code path. With this PR, when enabled, it now returns correct results on DV-enabled tables.
Replaces the "Auto-CDF" wording with "read-time CDF" across changelog comments, docs and config descriptions. Comment/string text only, no code identifiers changed. Reviewer comment: delta-io#6794 (comment)
The schema field holds the ending-version schema used as the read schema for the changelog range. Renames it to endDataSchema to disambiguate from per-commit schemas. Reviewer comment: delta-io#6794 (comment)
The changelog read path compared per-commit and start schemas to the end schema with strict equality. This rejects additive, read-compatible changes (new columns, relaxed nullability) and trips on internal field metadata. Replaces both checks with a SchemaUtils.isReadCompatible helper in the v2 schema utils that delegates to Delta's read-compatibility check (forbidTightenNullability = true, matching CDCReaderBase). Reviewer comment: delta-io#6794 (comment)
A Metadata action fully replaces the prior table configuration: all configs are listed explicitly. The changelog loop treated an absent row-tracking key as "inherit the prior value" and only failed on an explicit non-"true" value, so a Metadata commit that drops row tracking could slip through. Treats an absent key as the table default (disabled), and reads the config key from DeltaConfigs.ROW_TRACKING_ENABLED instead of a string literal. The in-range check now correctly rejects row tracking being disabled within the range; the boundary case is already caught earlier in DeltaChangelogScanBuilder. Reviewer comments: delta-io#6794 (comment) delta-io#6794 (comment) delta-io#6794 (comment)
Replaces the "Auto-CDF" wording with "read-time CDF" across changelog comments, docs and config descriptions. Comment/string text only, no code identifiers changed. Reviewer comment: delta-io#6794 (comment)
The schema field holds the ending-version schema used as the read schema for the changelog range. Renames it to endDataSchema to disambiguate from per-commit schemas. Reviewer comment: delta-io#6794 (comment)
The changelog read path compared per-commit and start schemas to the end schema with strict equality. This rejects additive, read-compatible changes (new columns, relaxed nullability) and trips on internal field metadata. Replaces both checks with a SchemaUtils.isReadCompatible helper in the v2 schema utils that delegates to Delta's read-compatibility check (forbidTightenNullability = true, matching CDCReaderBase). Reviewer comment: delta-io#6794 (comment)
A Metadata action fully replaces the prior table configuration: all configs are listed explicitly. The changelog loop treated an absent row-tracking key as "inherit the prior value" and only failed on an explicit non-"true" value, so a Metadata commit that drops row tracking could slip through. Treats an absent key as the table default (disabled), and reads the config key from DeltaConfigs.ROW_TRACKING_ENABLED instead of a string literal. The in-range check now correctly rejects row tracking being disabled within the range; the boundary case is already caught earlier in DeltaChangelogScanBuilder. Reviewer comments: delta-io#6794 (comment) delta-io#6794 (comment) delta-io#6794 (comment)
Replaces the "Auto-CDF" wording with "read-time CDF" across changelog comments, docs and config descriptions. Comment/string text only, no code identifiers changed. Reviewer comment: delta-io#6794 (comment)
The schema field holds the ending-version schema used as the read schema for the changelog range. Renames it to endDataSchema to disambiguate from per-commit schemas. Reviewer comment: delta-io#6794 (comment)
The changelog read path compared per-commit and start schemas to the end schema with strict equality. This rejects additive, read-compatible changes (new columns, relaxed nullability) and trips on internal field metadata. Replaces both checks with a SchemaUtils.isReadCompatible helper in the v2 schema utils that delegates to Delta's read-compatibility check (forbidTightenNullability = true, matching CDCReaderBase). Reviewer comment: delta-io#6794 (comment)
A Metadata action fully replaces the prior table configuration: all configs are listed explicitly. The changelog loop treated an absent row-tracking key as "inherit the prior value" and only failed on an explicit non-"true" value, so a Metadata commit that drops row tracking could slip through. Treats an absent key as the table default (disabled), and reads the config key from DeltaConfigs.ROW_TRACKING_ENABLED instead of a string literal. The in-range check now correctly rejects row tracking being disabled within the range; the boundary case is already caught earlier in DeltaChangelogScanBuilder. Reviewer comments: delta-io#6794 (comment) delta-io#6794 (comment) delta-io#6794 (comment)
<!-- Thanks for sending a pull request! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md 2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP] Your PR title ...'. 3. Be sure to keep the PR description updated to reflect all changes. 4. Please write your PR title to summarize what this PR proposes. 5. If possible, provide a concise example to reproduce the issue for a faster review. 6. If applicable, include the corresponding issue number in the PR title and link it in the body. --> #### Which Delta project/connector is this regarding? <!-- Please add the component selected below to the beginning of the pull request title For example: [Spark] Title of my pull request --> - [x] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description Follow-up cleanup for the read-time CDF (changelog) reader added in #6794 / #6830 / #6886, addressing reviewer feedback. The changelog V2 reader is gated behind the `changelogV2.enabled` SQL conf (default off) and only exists in the Spark 4.2 shim. Changes, each in its own commit: - **Terminology**: rename "Auto-CDF" to "read-time CDF" across changelog comments, docs and config descriptions (comment/string text only). ([comment](#6794 (comment))) - **Schema check**: replace strict schema equality in `DeltaChangelogBatch` with `SchemaUtils.isReadCompatible`, so additive, read-compatible changes (new columns, relaxed nullability) are accepted and internal field metadata is ignored. The check is wrapped in a small `isReadCompatible` helper on the v2 `SchemaUtils` (`forbidTightenNullability = true`, matching `CDCReaderBase`). Type changes are not read compatible and stay rejected. ([comment](#6794 (comment))) - **Naming**: rename the ambiguous `dataSchema` field to `endDataSchema`. ([comment](#6794 (comment))) - **Row-tracking detection**: a `Metadata` action fully replaces the prior table configuration, so an absent row-tracking key now means the table default (disabled) rather than an inherited value. The flag is read via the kernel `TableConfig.ROW_TRACKING_ENABLED` helper instead of a string literal. This makes the in-range row-tracking-disabled check correct; the boundary case is already caught earlier in `DeltaChangelogScanBuilder`. ([supersedes](#6794 (comment)), [config helper](#6794 (comment)), [in-range check](#6794 (comment))) - **Exceptions**: replace generic `RuntimeException` wrapping in the changelog read path with a `DELTA_CHANGELOG_READ_FAILED` error class (with `PLAN_INPUT_PARTITIONS` and `PROCESS_COMMIT_ACTIONS` sub-classes). Causes that already carry a Spark error class are rethrown unchanged so their user-facing class is preserved. ([comment](https://github.com/delta-io/delta/pull/6830/files#r3301759577), [comment](#6886 (comment))) - **Docs**: add a class-level comment describing the `DeltaChangelogBatch` read flow. ## How was this patch tested? Existing unit and integration tests in `DeltaChangelogCatalogIntegrationTest` and `SchemaUtilsTest`, extended for this change: - `SchemaUtilsTest` covers the `isReadCompatible` helper: additive column and relaxed nullability are accepted, dropped column and tightened nullability are rejected. - `DeltaChangelogCatalogIntegrationTest` was refactored onto managed tables (single `withTable` wrapper), and gains coverage for schema/row-tracking changes across the requested range: row tracking toggled off before or after the range is allowed, an additive schema change mid-range is allowed and the pre-change rows read back under the end schema, and a dropped column, tightened nullability, or type change mid-range is rejected with `DELTA_CHANGELOG_SCHEMA_CHANGE_IN_RANGE`. Positive tests assert the actual row values, not just the row count. ## Does this PR introduce _any_ user-facing changes? No <!-- If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible. If possible, please also clarify if this is a user-facing change compared to the released Delta Lake versions or within the unreleased branches such as master. If no, write 'No'. -->
4 participants
Please read my added comments in the conversation, they explain differences to the path I'd have expected to cleanly work.
Which Delta project/connector is this regarding?
Description
Guarded behing a feature flag, this PR Implements B.2 of the CDC SPIP, so the
Changeloginterface. Furthermore, it adds catalog-driven Auto-CDF (TableCatalog.loadChangelog) for the DSv2 connector, so the kernel-based V2 reader stack answersSELECT * FROM t CHANGES FROM VERSION/TIMESTAMP ...batch queries introduced by SPARK-55668 (apache/spark#55508).The streaming CDC entrypoint shipped in #6359 is preserved unchanged and coexists with the new batch path through an explicit
CdcReadModeflag.What the PR adds
DeltaCatalogChangelogSupport(new, inspark-unified) — abstract Scala class betweenAbstractDeltaCatalog(sparkV1) and the hybridDeltaCatalog(spark-unified). OverridesloadChangelog(ident, changelogInfo)and dispatches on theChangelogRangesubtype (version / timestamp / unbounded). Resolves the table throughloadTable(so it works for bothSparkTableandDeltaTableV2), loads the latest Kernel snapshot viaSnapshotManagerFactory, validates row tracking is enabled, applies bounds-inclusivity adjustments, and returns aDeltaChangelog.The class lives in
spark-unifiedbecause the implementation references sparkV2 classes (SparkTable,DeltaChangelog,SnapshotManagerFactory,V2SchemaUtils) andsparkV1cannot depend onsparkV2.DeltaCatalog.javanow extendsDeltaCatalogChangelogSupport.CdcReadMode(new enum,sparkV2) — replaces the priorboolean isCDCReadonPartitionUtils.createDeltaParquetReaderFactory:NONE— non-CDC scan (SparkBatch).STREAMING— opt-in streaming CDC (SparkMicroBatchStreamwhenreadChangeFeed = true).PartitionUtilsowns the CDC schema augmentation andCDCReadFunctionwrap, as before.BATCH_CHANGELOG— Auto-CDF (DeltaChangelogBatch).PartitionUtilsleaves schema and reader untouched;CDCPartitionReaderFactoryinDeltaChangelogBatchinjects_change_type/_commit_version/_commit_timestampas per-partition constants instead. This avoids the double-injection / schema-vs-reader misalignment that the previous sharedisCDCRead=truepath caused.Feature flag
DELTA_CHANGELOG_V2_ENABLED(changelogV2.enabled, internal, defaultfalse). When disabled,DeltaCatalogChangelogSupport.loadChangelogdelegates tosuper.loadChangelog, which surfaces the familiarUNSUPPORTED_FEATURE.CHANGE_DATA_CAPTUREerror. Lets the implementation land without changing user-visible behavior until tests catch up.Per-commit ordering in
DeltaChangelogBatch.planInputPartitions— emits allRemoveFilepartitions beforeAddFilepartitions within a single commit, so Spark's batch CDC post-processor (ResolveChangelogTable) sees preimage → postimage pairs regardless of the action order in the on-disk commit log. The Delta protocol does not contract action order; pinning the partition order here makes the test expectations stable across protocol implementations.Why this is split this way
The DSv2 Changelog interface (apache/spark#55426) covers both batch and streaming, but the two enter Delta through different paths today:
option("readChangeFeed", "true")→SparkMicroBatchStream→
PartitionUtilswithCdcReadMode.STREAMING. Existing surface from [kernel-spark][Part 4] CDC data reading: ReadFunc decorator, schema coordination, and reader factory wiring #6359; preserved by this PR.CHANGES FROM(SQL or DataFrame) → analyzer (SPARK-56686, SPARK-56687 / apache/spark#55637)→
TableCatalog.loadChangelog→DeltaCatalog→DeltaCatalogChangelogSupport.loadChangelog→DeltaChangelog→DeltaChangelogScan→DeltaChangelogBatch→PartitionUtilswithCdcReadMode.BATCH_CHANGELOG.The follow-up
DeltaChangelogScan.toMicroBatchStream()(so the catalog-driven Changelog also drives streaming reads, completing the surface that apache/spark#55637 builds on fordeduplicationMode = netChanges) is left as aTODOinDeltaChangelogScanfor a follow-up PR.Limitations / known follow-ups
requires row trackinganalysis error otherwise).UnboundedRangeis rejected withDELTA_CHANGELOG_UNBOUNDED_RANGE; Auto-CDF always operates over a bounded range.DeltaChangelogScandoes not yet implementtoMicroBatchStream()— see inlineTODO.DeltaChangelogBatch.planInputPartitionsstill goes throughStreamingHelper.getCommitActionsFromRangeUnsafe(markedTODOin the file). The helper is generic — only the class name is streaming-flavored. A separate rename / extract pass would be good.Build dependency
The new code uses the
Changelog/ChangelogInfo/ChangelogRangeinterfaces added in Spark 4.2 (apache/spark#55426). This PR is tested by cherry picking changes from #6657, which re-enables the Spark 4.2 snapshot cross-build row.How was this patch tested?
New tests (17 total, all green):
DeltaChangelogDirectBatchExecutionTest— exercises theDeltaChangelogScan→Batch→PartitionReaderpath directly, without going through SQL. Covers: initial insert + delete with paired DELETE/INSERT output, UPDATE-as-CoW producing paired preimage/postimage rows at the same commit, range slicing on non-zero start, and rowId / rowVersion field-reference contract.DeltaChangelogCatalogIntegrationTest— exercises the catalog-routed entrypoint (TableCatalog.loadChangelog) over both SQLCHANGES FROMand the DataFrame API. Covers all fourChangelogRangeshapes (version range, timestamp range, open-ended, exclusive bounds), boundary inclusivity, and the failure modes (timestamp before earliest commit, timestamp after latest commit, empty exclusive range, unbounded rejection).Pre-existing suites still pass:
SparkMicroBatchStreamCDCTest— confirms the streaming CDC entrypoint introduced in [kernel-spark][Part 4] CDC data reading: ReadFunc decorator, schema coordination, and reader factory wiring #6359 is unaffected by theCdcReadModerefactor.PartitionUtilsTest— updated to passCdcReadModeinstead ofboolean.Build / test command used locally:
Does this PR introduce any user-facing changes?
No, not while the feature flag is at its default.
With
changelogV2.enabled = true(internal flag, defaultfalse):SELECT * FROM <table> CHANGES FROM VERSION/TIMESTAMP ...and the DataFrame.changes(...)builder become functional on row-tracking-enabled Delta tables via the V2 connector. Previously the V2 catalog rejected these queries withUNSUPPORTED_FEATURE.CHANGE_DATA_CAPTURE.