[kernel-spark] Create v2 adapters for metadata and protocol#6546
Merged
TimothyW553 merged 2 commits intoMay 5, 2026
Conversation
PorridgeSwim
requested review from
TimothyW553,
huan233usc,
murali-db,
raveeram-db and
tdas
as code owners
PorridgeSwim
changed the title
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
ae2f693 to
f380f30
Compare
Collaborator
Author
Range-diff: master (ae2f693 -> f380f30)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
f380f30 to
7713d81
Compare
Collaborator
Author
Range-diff: master (f380f30 -> 7713d81)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
7713d81 to
08b4adc
Compare
Collaborator
Author
Range-diff: master (7713d81 -> 08b4adc)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
08b4adc to
27798f7
Compare
Collaborator
Author
Range-diff: master (08b4adc -> 27798f7)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
27798f7 to
6205321
Compare
Collaborator
Author
Range-diff: master (27798f7 -> 6205321)
Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
6205321 to
4e8ae79
Compare
Collaborator
Author
Range-diff: master (6205321 -> 4e8ae79)
Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
4e8ae79 to
d1be3cc
Compare
This was referenced Apr 12, 2026
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
d1be3cc to
ca71e8f
Compare
Collaborator
Author
Range-diff: master (d1be3cc -> ca71e8f)
Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
ca71e8f to
71534ed
Compare
Collaborator
Author
Range-diff: master (ca71e8f -> 71534ed)
Reproduce locally: |
This was referenced Apr 14, 2026
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
71534ed to
19b49ba
Compare
Collaborator
Author
Range-diff: master (71534ed -> 19b49ba)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
19b49ba to
c803c1a
Compare
Collaborator
Author
Range-diff: master (19b49ba -> c803c1a)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
c803c1a to
a50c9d2
Compare
Collaborator
Author
Range-diff: master (c803c1a -> a50c9d2)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
a50c9d2 to
dfe61d9
Compare
Collaborator
Author
Range-diff: master (a50c9d2 -> dfe61d9)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
TimothyW553
reviewed
Apr 30, 2026
Comment on lines
+111
to
+118
| @Override | ||
| public StructType partitionSchema() { | ||
| if (cachedPartitionSchema == null) { | ||
| cachedPartitionSchema = AbstractMetadata.super.partitionSchema(); | ||
| } | ||
| return cachedPartitionSchema; | ||
| } | ||
| } |
Collaborator
There was a problem hiding this comment.
can we add a test where partitionColumns = ["Part1"] and the schema field is part1? Note the difference in capitalization.
Comment on lines
+96
to
+109
| public DeltaColumnMappingMode columnMappingMode() { | ||
| ColumnMapping.ColumnMappingMode kernelMode = | ||
| ColumnMapping.getColumnMappingMode(kernelMetadata.getConfiguration()); | ||
| switch (kernelMode) { | ||
| case NONE: | ||
| return NoMapping$.MODULE$; | ||
| case ID: | ||
| return IdMapping$.MODULE$; | ||
| case NAME: | ||
| return NameMapping$.MODULE$; | ||
| default: | ||
| throw new UnsupportedOperationException("Unsupported column mapping mode: " + kernelMode); | ||
| } | ||
| } |
Collaborator
There was a problem hiding this comment.
V1 already has DeltaColumnMappingMode.apply(String) that does this exact mapping. can we reuse it instead of maintaining two tables?
Collaborator
Author
There was a problem hiding this comment.
Thank you for the info, reused the v1 apply
Comment on lines
+77
to
+82
| cachedPartitionColumns = | ||
| CollectionConverters.asScala( | ||
| VectorUtils.toJavaList(kernelMetadata.getPartitionColumns()).stream() | ||
| .map(Object::toString) | ||
| .collect(Collectors.toList())) | ||
| .toSeq(); |
Collaborator
There was a problem hiding this comment.
toJavaList already returns List here. The .stream().map(Object::toString) does nothing and just hides type bugs.
Suggested change
| cachedPartitionColumns = | |
| CollectionConverters.asScala( | |
| VectorUtils.toJavaList(kernelMetadata.getPartitionColumns()).stream() | |
| .map(Object::toString) | |
| .collect(Collectors.toList())) | |
| .toSeq(); | |
| List<String> rawCols = VectorUtils.toJavaList(kernelMetadata.getPartitionColumns()); | |
| cachedPartitionColumns = CollectionConverters.asScala(rawCols).toSeq(); |
Collaborator
Author
There was a problem hiding this comment.
I was using stream to cast type from Object to String, but you code looks better
Comment on lines
+50
to
+59
| public Option<Set<String>> readerFeatures() { | ||
| if (cachedReaderFeatures == null) { | ||
| cachedReaderFeatures = | ||
| kernelProtocol.supportsReaderFeatures() | ||
| ? Option.apply( | ||
| CollectionConverters.asScala(kernelProtocol.getReaderFeatures()).toSet()) | ||
| : Option.empty(); | ||
| } | ||
| return cachedReaderFeatures; | ||
| } |
Collaborator
There was a problem hiding this comment.
volatile + check-then-set seems racy (two threads can both compute)
Collaborator
Author
There was a problem hiding this comment.
kernelProtocol is declared to be a constant, so it is fine
| case NONE: | ||
| return NoMapping$.MODULE$; | ||
| case ID: | ||
| return IdMapping$.MODULE$; |
Collaborator
There was a problem hiding this comment.
Only NoMapping and NameMapping are tested -- IdMapping is uncovered. same with unknown mode.
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
dfe61d9 to
f3359c7
Compare
This was referenced May 1, 2026
Collaborator
Author
Range-diff: master (dfe61d9 -> f3359c7)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
TimothyW553
approved these changes
May 1, 2026
Collaborator
TimothyW553
left a comment
TimothyW553
left a comment
There was a problem hiding this comment.
Hi @PorridgeSwim approved with one comment. Please confirm, update the branch, and ping me to merge when CI is green.
| public void testProtocolAdapterWithTableFeatures() { | ||
| // Reader features: supported but empty (version >= 3 means features are supported, even with | ||
| // an empty set). Writer features: supported and populated. | ||
| Set<String> readerFeatures = Collections.emptySet(); |
Collaborator
There was a problem hiding this comment.
Please include v2Checkpoint here too, because it is a reader-writer feature and this test should use a valid protocol.
Suggested change
| Set<String> readerFeatures = Collections.emptySet(); | |
| Set<String> readerFeatures = Collections.singleton("v2Checkpoint"); |
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
f3359c7 to
7bbf917
Compare
Collaborator
Author
Range-diff: master (f3359c7 -> 7bbf917)
... (truncated, output exceeded 60000 bytes) Reproduce locally: |
tdas
reviewed
May 4, 2026
| * Adapter from {@link io.delta.kernel.internal.actions.Metadata} to {@link | ||
| * org.apache.spark.sql.delta.v2.interop.AbstractMetadata}. | ||
| */ | ||
| public class KernelMetadataAdapter implements AbstractMetadata { |
Contributor
There was a problem hiding this comment.
why is there both interop and adapters directories?
whats the difference?
Collaborator
Author
There was a problem hiding this comment.
interop is under org.apache.spark.sql define the AbstractMetadata interface that v1 MetadataAction extends, while adapters is under io.delta.spark.internal.v2 that adapts v2 MetadataAction into AbstractMetadata to bridge between v1 and v2 MetadataAction.
zikangh
reviewed
May 5, 2026
| @Override | ||
| public DeltaColumnMappingMode columnMappingMode() { | ||
| String mode = kernelMetadata.getConfiguration().get(ColumnMapping.COLUMN_MAPPING_MODE_KEY); | ||
| return mode == null ? NoMapping$.MODULE$ : DeltaColumnMappingMode$.MODULE$.apply(mode); |
Collaborator
There was a problem hiding this comment.
should we cache columnMappingMode for faster access?
zikangh
approved these changes
May 5, 2026
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
7bbf917 to
6070b93
Compare
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
6070b93 to
84f5ce6
Compare
Co-authored-by: Isaac
PorridgeSwim
force-pushed
the
stack/SparkMetadataAdapter
branch
from
97651ca to
9271a62
Compare
TimothyW553
pushed a commit
that referenced
this pull request
May 6, 2026
…#6550) ## 🥞 Stacked PR Use this [link](https://github.com/delta-io/delta/pull/6550/files) to review incremental changes. - [stack/SparkMetadataAdapter](#6546) [[Files changed](https://github.com/delta-io/delta/pull/6546/files)] [MERGED] - [**stack/RefactorMetadataTrackingLog**](#6550) [[Files changed](https://github.com/delta-io/delta/pull/6550/files)] - [stack/RefactorDeltaSourceMetadataEvolutionSupport](#6562) [[Files changed](https://github.com/delta-io/delta/pull/6562/files/953f137f8c4ce46d8b8a9605b0c7bed898e30df4..027984b6edcbad0f4731e560425c2ed9bcf8fc27)] - [stack/MetadataEvolutionHandler2](#6563) [[Files changed](https://github.com/delta-io/delta/pull/6563/files/027984b6edcbad0f4731e560425c2ed9bcf8fc27..ada845895139edcb2727a87b39922c8e16837a99)] - [stack/NonAdditiveSchemaEvolution2](#6570) [[Files changed](https://github.com/delta-io/delta/pull/6570/files/ada845895139edcb2727a87b39922c8e16837a99..476762fde7b9cb9b9bc3e416c86a260cd29806ed)] - [stack/NonAdditiveSchemaEvolution3](#6697) [[Files changed](https://github.com/delta-io/delta/pull/6697/files/476762fde7b9cb9b9bc3e416c86a260cd29806ed..13395a7f2a49db4962091e8ee919bebdab5bd4e2)] - [stack/consecutiveSchemaChangesMerger](#6698) [[Files changed](https://github.com/delta-io/delta/pull/6698/files/13395a7f2a49db4962091e8ee919bebdab5bd4e2..f22ba063eaf35ab69d653a2d5faefdc52f35eab5)] --------- #### Which Delta project/connector is this regarding? - [X] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description PR 2/7 in the non-additive schema evolution for V2 streaming connector stack. Decouple `DeltaSourceMetadataTrackingLog` and `PersistedMetadata` from V1-specific types so the schema log can be reused by the V2 connector. - Replace `SnapshotDescriptor` parameter in `create()` with plain `sourceTableId` and `sourceDataPath` strings - Unify `PersistedMetadata.apply` to accept `AbstractMetadata`/`AbstractProtocol` instead of V1 `Metadata`/`Protocol` - Extract the consecutive schema changes merger (V1-specific, depends on `DeltaLog`) out of the companion object into `DeltaSourceMetadataEvolutionSupport`, and inject it as a function parameter so V2 can provide its own implementation - Remove `Protocol`'s `private` constructor modifier to allow construction from abstract protocol fields All changes are structural refactors with no behavioral change. ## How was this patch tested? Existing tests in `DeltaSourceSchemaEvolutionSuite` updated to use the new API. No behavioral changes. ## Does this PR introduce _any_ user-facing changes? No.
murali-db
pushed a commit
that referenced
this pull request
May 6, 2026
…seable in v2 (#6562) ## 🥞 Stacked PR Use this [link](https://github.com/delta-io/delta/pull/6562/files) to review incremental changes. - [stack/SparkMetadataAdapter](#6546) [[Files changed](https://github.com/delta-io/delta/pull/6546/files)] [MERGED] - [stack/RefactorMetadataTrackingLog](#6550) [[Files changed](https://github.com/delta-io/delta/pull/6550/files)] [MERGED] - [**stack/RefactorDeltaSourceMetadataEvolutionSupport**](#6562) [[Files changed](https://github.com/delta-io/delta/pull/6562/files)] - [stack/MetadataEvolutionHandler2](#6563) [[Files changed](https://github.com/delta-io/delta/pull/6563/files/ed92a0fa2051432b6bc5784034df0b7949bbfb98..e5b2c3295843ec85753e07dc0010aa5ccebaabb7)] - [stack/NonAdditiveSchemaEvolution2](#6570) [[Files changed](https://github.com/delta-io/delta/pull/6570/files/e5b2c3295843ec85753e07dc0010aa5ccebaabb7..7c66bf11a0f1b651cda32ed7f529f552dd9dbfcb)] - [stack/NonAdditiveSchemaEvolution3](#6697) [[Files changed](https://github.com/delta-io/delta/pull/6697/files/7c66bf11a0f1b651cda32ed7f529f552dd9dbfcb..14956ea304c93d2343ccd7eb89a112966f07f906)] - [stack/consecutiveSchemaChangesMerger](#6698) [[Files changed](https://github.com/delta-io/delta/pull/6698/files/14956ea304c93d2343ccd7eb89a112966f07f906..8101b335b892a6a5b6d6fe11f4a202d14102721c)] --------- #### Which Delta project/connector is this regarding? - [X] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description PR 3/7 in the non-additive schema evolution for V2 streaming connector stack. Refactor `DeltaSourceMetadataEvolutionSupport` and `DeltaColumnMapping` so the schema change detection logic can be called from V2 without depending on V1 instance state. **`DeltaSourceMetadataEvolutionSupport`:** - Extract instance methods (`validateAndResolveMetadataEvolution`, `checkColumnMappingSchemaChangesDuringStreaming`, `resolveMetadataEvolutionForCommitRange`, etc.) to companion object statics that accept explicit parameters instead of accessing V1 `DeltaSource` via `this` - V1 trait methods now delegate to the companion object statics **`DeltaColumnMapping`:** - Widen `hasNoColumnMappingSchemaChanges` from V1 `Metadata` to `AbstractMetadata` so V2 can call it via the adapter layer - Extract `assignColumnIdAndPhysicalNameToSchema(StructType, Map)` from `assignColumnIdAndPhysicalName(Metadata, Metadata, ...)` — needed for simulating column mapping upgrades during NoMapping-to-NameMapping transitions All changes are structural refactors with no behavioral change. ## How was this patch tested? Existing tests in `DeltaSourceSchemaEvolutionSuite` continue to pass. No behavioral changes. ## Does this PR introduce _any_ user-facing changes? No.
Draft
murali-db
pushed a commit
that referenced
this pull request
May 11, 2026
## 🥞 Stacked PR Use this [link](https://github.com/delta-io/delta/pull/6563/files) to review incremental changes. - [stack/SparkMetadataAdapter](#6546) [[Files changed](https://github.com/delta-io/delta/pull/6546/files)] [MERGED] - [stack/RefactorMetadataTrackingLog](#6550) [[Files changed](https://github.com/delta-io/delta/pull/6550/files)] [MERGED] - [stack/RefactorDeltaSourceMetadataEvolutionSupport](#6562) [[Files changed](https://github.com/delta-io/delta/pull/6562/files)] [MERGED] - [**stack/MetadataEvolutionHandler2**](#6563) [[Files changed](https://github.com/delta-io/delta/pull/6563/files)] - [stack/NonAdditiveSchemaEvolution2](#6570) [[Files changed](https://github.com/delta-io/delta/pull/6570/files/a20f1f3ab452a75fc954e15c57c17327e0cb9267..0e07f87285becd6be416450ae084df454d9c94a9)] - [stack/NonAdditiveSchemaEvolution3](#6697) [[Files changed](https://github.com/delta-io/delta/pull/6697/files/0e07f87285becd6be416450ae084df454d9c94a9..73e1aa7f4162a3e1480ffd2b88b9ca79d852f2fe)] - [stack/consecutiveSchemaChangesMerger](#6698) [[Files changed](https://github.com/delta-io/delta/pull/6698/files/73e1aa7f4162a3e1480ffd2b88b9ca79d852f2fe..5e5d260b64d45cc11bcfdb58e5aab1b2d2637b33)] - [stack/V1V2MixTest](#6759) [[Files changed](https://github.com/delta-io/delta/pull/6759/files/5e5d260b64d45cc11bcfdb58e5aab1b2d2637b33..738379713040986c74f98dbebfdc6c83ec1d3f16)] --------- #### Which Delta project/connector is this regarding? - [X] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description PR 4/7 in the non-additive schema evolution for V2 streaming connector stack. Introduce `MetadataEvolutionHandler`, a Java class that implements the V1 barrier protocol for schema evolution in the V2 connector. In V1 this logic lives in `DeltaSourceMetadataEvolutionSupport`, a Scala trait mixed into `DeltaSource` that accesses stream state via `this`. Since V2's `SparkMicroBatchStream` is Java and cannot use Scala trait mixins, `MetadataEvolutionHandler` receives all dependencies via constructor injection instead. The handler covers the full schema evolution lifecycle: - **Stream start**: eager metadata tracking log initialization on first batch - **Offset generation**: injects `METADATA_CHANGE_INDEX` / `POST_METADATA_CHANGE_INDEX` barrier sentinels into the file change iterator - **Pending schema offsets**: returns barrier offsets for in-progress schema changes - **Batch commit**: updates the schema log and throws `DELTA_STREAMING_METADATA_EVOLUTION` to trigger stream restart - **Batch planning on restart**: validates and re-initializes the schema log All detection logic delegates to the shared `DeltaSourceMetadataEvolutionSupport$` companion object statics (refactored in PR 3/7). V2-specific orchestration is limited to wiring the barrier protocol into the `CloseableIterator<IndexedFile>` pipeline and collecting metadata/protocol from Kernel commit ranges via `StreamingHelper`. Also extends `StreamingHelper` with `getMetadataAndProtocolForVersionRange` to collect metadata and protocol actions from a range of Kernel commits. ## How was this patch tested? Unit tests in `MetadataEvolutionHandlerTest.java` covering: barrier protocol (METADATA_CHANGE_INDEX / POST_METADATA_CHANGE_INDEX offset generation), tracking state transitions, initialization lifecycle, offset arithmetic, pending schema change handling, and commit-time evolution exception. ## Does this PR introduce _any_ user-facing changes? No.
5 tasks
murali-db
pushed a commit
that referenced
this pull request
May 16, 2026
## 🥞 Stacked PR Use this [link](https://github.com/delta-io/delta/pull/6570/files) to review incremental changes. - [stack/SparkMetadataAdapter](#6546) [[Files changed](https://github.com/delta-io/delta/pull/6546/files)] [MERGED] - [stack/RefactorMetadataTrackingLog](#6550) [[Files changed](https://github.com/delta-io/delta/pull/6550/files)] [MERGED] - [stack/RefactorDeltaSourceMetadataEvolutionSupport](#6562) [[Files changed](https://github.com/delta-io/delta/pull/6562/files)] [MERGED] - [stack/MetadataEvolutionHandler2](#6563) [[Files changed](https://github.com/delta-io/delta/pull/6563/files)] [MERGED] - [**stack/NonAdditiveSchemaEvolution2**](#6570) [[Files changed](https://github.com/delta-io/delta/pull/6570/files)] - [stack/NonAdditiveSchemaEvolution3](#6697) [[Files changed](https://github.com/delta-io/delta/pull/6697/files/b7f6c8ebfc0882e7e2cc580f09f376be23a8d43d..dbb6246c14be1ab7f017ad9fc26455ae599ee676)] - [stack/consecutiveSchemaChangesMerger](#6698) [[Files changed](https://github.com/delta-io/delta/pull/6698/files/dbb6246c14be1ab7f017ad9fc26455ae599ee676..4bf2fa3fa828bcab0b56c4c26ca51ee9cc40b482)] - [stack/SchemaTrackingWithCDC](#6801) [[Files changed](https://github.com/delta-io/delta/pull/6801/files/4bf2fa3fa828bcab0b56c4c26ca51ee9cc40b482..a78a4ac2bc9a52605278a36b98804230258c12a2)] - [stack/V1V2MixTest](#6759) [[Files changed](https://github.com/delta-io/delta/pull/6759/files/7f9b7f2724b2245ab7380908616303cf7ea95fca..e146cdc9ebb0572e8b0a928cc6dd3bfdc198d984)] --------- #### Which Delta project/connector is this regarding? - [X] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description PR 5/7 in the non-additive schema evolution for V2 streaming connector stack. Wire schema tracking into V2's analysis path so the analyzed plan reflects the persisted (evolved) schema instead of the live snapshot schema. - `DeltaAnalysis.verifyDeltaSourceSchemaLocation`: extend the duplicate-schema-location check to also visit `StreamingRelationV2`, keyed on the V2 `Table.name`. - `SparkTable`: open `DeltaSourceMetadataTrackingLog` once during construction (gated on `mergeConsecutiveSchemaChanges`) and seed `SchemaProvider` from the persisted metadata, so analysis-time `schema()` matches what the stream will read at runtime. - `ApplyV2ReadOptions` (renamed from `ApplyV2Streaming`): generalize the CDC-only rebuild to also fire when `schemaTrackingLocation` arrives via `extraOptions` on the catalog `readStream.table()` path; rebuild `SparkTable` with merged options so the schema-log lookup actually fires. - `MetadataEvolutionHandler.getMetadataTrackingLogForMicroBatchStream`: V2 port of V1's helper, reused by `SparkTable` (analysis) and `SparkScan` (execution). ## How was this patch tested? `SparkTableTest`, `MetadataEvolutionHandlerTest`, `ApplyV2ReadOptionsSuite`. Unified `DeltaV2SourceSchemaEvolutionSuite` updated. ## Does this PR introduce _any_ user-facing changes? No.
murali-db
pushed a commit
that referenced
this pull request
May 16, 2026
…6697) ## 🥞 Stacked PR Use this [link](https://github.com/delta-io/delta/pull/6697/files) to review incremental changes. - [stack/SparkMetadataAdapter](#6546) [[Files changed](https://github.com/delta-io/delta/pull/6546/files)] [MERGED] - [stack/RefactorMetadataTrackingLog](#6550) [[Files changed](https://github.com/delta-io/delta/pull/6550/files)] [MERGED] - [stack/RefactorDeltaSourceMetadataEvolutionSupport](#6562) [[Files changed](https://github.com/delta-io/delta/pull/6562/files)] [MERGED] - [stack/MetadataEvolutionHandler2](#6563) [[Files changed](https://github.com/delta-io/delta/pull/6563/files)] [MERGED] - [stack/NonAdditiveSchemaEvolution2](#6570) [[Files changed](https://github.com/delta-io/delta/pull/6570/files)] [MERGED] - [**stack/NonAdditiveSchemaEvolution3**](#6697) [[Files changed](https://github.com/delta-io/delta/pull/6697/files)] - [stack/consecutiveSchemaChangesMerger](#6698) [[Files changed](https://github.com/delta-io/delta/pull/6698/files/f96643aa3cc01e7f70cc13a18b82dc27f277f11d..f612628ad931ec35c237801109f01b6fbd1379f7)] - [stack/SchemaTrackingWithCDC](#6801) [[Files changed](https://github.com/delta-io/delta/pull/6801/files/f612628ad931ec35c237801109f01b6fbd1379f7..4aeacfb120b33e9cdfe124352290b72f53f7cf89)] - [stack/V1V2MixTest](#6759) [[Files changed](https://github.com/delta-io/delta/pull/6759/files/f612628ad931ec35c237801109f01b6fbd1379f7..0c818ee431ab417a4f2ffbcc609930be09d25031)] --------- #### Which Delta project/connector is this regarding? - [X] Spark - [ ] Standalone - [ ] Flink - [ ] Kernel - [ ] Other (fill in here) ## Description PR 6/7 in the non-additive schema evolution for V2 streaming connector stack. Wire `MetadataEvolutionHandler` into `SparkMicroBatchStream` and `SparkScan` so V2 streaming reads honor non-additive schema evolution (column rename/drop, type widening). - `SparkMicroBatchStream`: take `metadataTrackingLog` + `metadataPath` as constructor inputs; when a persisted entry exists, layer it onto the freshly loaded `snapshotAtSourceInit` to derive `readSnapshotAtSourceInit` (mirrors V1's `readSnapshotDescriptor`). Integrate the schema-evolution barrier protocol into `latestOffset` / `commit` / `planInputPartitions`. Skip the on-restart schema-validation check when schema tracking is active — the schema-log evolution exception covers it. - `SparkScan.toMicroBatchStream`: reload latest snapshot (the analysis-time `initialSnapshot` can be stale by stream start), open the tracking log via `MetadataEvolutionHandler.getMetadataTrackingLogForMicroBatchStream` with `mergeConsecutiveSchemaChanges=false` (the merger only runs at analysis), and pass it through with the checkpoint location. - `SparkScan` option allow-list: move `allowSourceColumnDrop` / `Rename` / `TypeChange` out of the unsupported list now that they are honored. ## How was this patch tested? `SparkMicroBatchStreamTest`, `MetadataEvolutionHandlerTest`. Unified suites (`DeltaV2SourceSchemaEvolutionSuite`, `TypeWideningStreamingV2SourceSuite`, `RemoveColumnMappingStreamingReadV2Suite`) move non-merger evolution scenarios from `shouldFailTests` to `shouldPassTests`; merger-dependent tests remain pending until PR 7/7. ## Does this PR introduce _any_ user-facing changes? No.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
🥞 Stacked PR
Use this link to review incremental changes.
Which Delta project/connector is this regarding?
Description
PR 1/7 in the non-additive schema evolution for V2 streaming connector stack.
The shared V1 Scala utilities (
DeltaColumnMapping,DeltaSourceMetadataEvolutionSupport) operate onAbstractMetadata/AbstractProtocol, but V2 holds Kernel types. This PR creates two adapter classes that bridge the gap:KernelMetadataAdapter: KernelMetadata→AbstractMetadata(schema conversion viaSchemaUtils, partition columns and configuration converted to Scala collections)KernelProtocolAdapter: KernelProtocol→AbstractProtocol(maps reader/writer features toOption[Set[String]])Also adds
columnMappingModeandpartitionSchemato theAbstractMetadatatrait — V1'sMetadataalready had these fields, the trait just didn't expose them.How was this patch tested?
Unit tests in
ActionAdaptersTest.java: table-features protocol, legacy protocol, full metadata round-trip, null optional fields, and null constructor rejection.Does this PR introduce any user-facing changes?
No.