[kernel-spark] Add getStartingVersion() for obtaining the initial offset for DSv2 streaming #5356

zikangh · 2025-10-17T00:57:51Z

Which Delta project/connector is this regarding?

Description

This PR is Part I of implementing SparkMicroBatchStream.initialSnapshot() to support Kernel-based dsv2 Delta streaming (M1 milestone). This PR handles the read option startingVersion.

How was this patch tested?

Parameterized tests verifying parity between DSv1 (DeltaSource) and DSv2 (SparkMicroBatchStream).

Does this PR introduce any user-facing changes?

No

zikangh · 2025-10-17T00:58:41Z

Hi, could I get a review from @huan233usc @gengliangwang @jerrypeng @tdas please? Thanks!

huan233usc · 2025-10-20T17:43:24Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+   * <p>This is the DSv2 Kernel-based implementation of DeltaSource.getStartingVersion.
+   */
+  Optional<Long> getStartingVersion() {
+    if (options == null) {


Can we check non null in the constructor? Objects.requireNonNull

I'm leaning toward continuing to support null options -- it should represent optional configurations, not required (e.g. some tests pass in zero options). wdyt?

What's the difference between null options and new DeltaOptions(Map$.MODULE$.empty(), spark.sessionState().conf());

Was trying to avoid assigning null if possible to reduce risks of NPE

When would "options" be null?

it should represent optional configurations, not required (e.g. some tests pass in zero options)

Java supports an "Optional" API. Why not use that so it is clear to the reader this variable is optional.

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

huan233usc · 2025-10-20T17:48:53Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+      // so we must check the message. See DeltaErrors.unsupportedTableFeature,
+      // DeltaErrors.unsupportedReaderFeatures, and DeltaErrors.unsupportedWriterFeatures.
+      String exceptionMessage = e.getMessage();
+      if (exceptionMessage != null


Maybe only keep exceptionMessage.contains("Unsupported Delta reader features")?

Good call. what about "Unsupported Delta table feature"?

"Unsupported Delta table feature"
Let's added that, good catch

huan233usc · 2025-10-20T17:51:46Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+      return true;
+    } catch (KernelException e) {
+      // Check if it's an unsupported table feature exception
+      // Kernel throws plain KernelException (not a subclass) for unsupported features,


I think there is a improvement tasks for kernel to have finer-grain error throwing cc @raveeram-db

That would be great. created an issue: #5369

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

huan233usc · 2025-10-20T17:59:36Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+            streamingHelper.checkVersionExists(
+                version, /* mustBeRecreatable= */ false, /* allowOutOfRange= */ false);
+          } catch (Exception e) {
+            throw new RuntimeException("Failed to validate starting version: " + version, e);


Wondering if it helps to avoid this try-catch block if we make the not found exception extend AnalysisException

Created #5369

Can we throw the same exception class as DSv1 which would be AnalysisException?

@jerrypeng yes we can throw the DSv1 version of this Exception

Although we should still wrap this with RuntimeException because (2a). the caller should not have to handle this exception again and (2b). AnalysisException is also checked unfortunately.

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java

huan233usc · 2025-10-20T22:20:21Z

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java

+    deltaLog.checkpoint();
+
+    // Delete log files for versions 1-5 to make them non-recreatable
+    // Note: Version 0 is kept because it contains the table schema


I think this produces a state of table that will never be reached?

So basically the log file clean up, we will by timestamp so always delete 0-5 (as 0 is earlier), for the schema, we will make sure there is always a checkpoint at the earlist version

-- so to setup the test, let's create 5.checkpoint, 10.checkpoint then remove log 0-5

Fixed. Thanks!

jerrypeng · 2025-10-21T18:03:48Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+    // DeltaSQLConf.DELTA_CDF_ALLOW_OUT_OF_RANGE_TIMESTAMP.
+    if (options.startingVersion().isDefined()) {
+      DeltaStartingVersion startingVersion = options.startingVersion().get();
+      if (startingVersion.equals(StartingVersionLatest$.MODULE$)) {


why not use instanceOf instead of equals?

jerrypeng · 2025-10-21T18:12:47Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+      String exceptionMessage = e.getMessage();
+      if (exceptionMessage != null
+          && exceptionMessage.contains("Unsupported Delta reader features")) {
+        throw new RuntimeException(e);


Why do we need to wrap the KernelException in a RuntimeException? Also DSv1 throws "DeltaUnsupportedTableFeatureException". Can we throw the same exception to maintain the same error handling behavior?

You are right, KernelException is unchecked; I removed the wrapping

it's challenging to convert the KernelException to dsv1's DeltaUnsupportedTableFeatureException in this case as it doesn't expose any error params. I agree we should try to throw the same exceptions as DSv1, but sometimes it's challenging to do so. Is it really that important to keep the Exception types consistent if we are failing the stream anyway?

jerrypeng · 2025-10-21T18:16:27Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+      // Suppress other KernelExceptions
+      logger.warn("Protocol validation failed at version {} with: {}", version, e.getMessage());
+      return false;
+    } catch (Exception e) {


I don't think we want to suppress all other exception here for example InterruptedException. We should just follow what DSv1 is doing and only suppress NonFatal Exceptions.

what we are doing here is equivalent to the scala NonFatal

We won't catch VirtualMachineError, ThreadDeath, LinkageError (extends Error)

We won't catch ControlThrowable (extends Throwable)

We don't need to catch InterruptedException because it won't be thrown by the kernel.

jerrypeng · 2025-10-21T18:43:10Z

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java

+        Arguments.of(/* startingVersion= */ "3", /* expectedVersion= */ Optional.of(3L)),
+        Arguments.of(/* startingVersion= */ "5", /* expectedVersion= */ Optional.of(5L)),
+        Arguments.of(/* startingVersion= */ "latest", /* expectedVersion= */ Optional.of(6L)),
+        Arguments.of(/* startingVersion= */ null, /* expectedVersion= */ Optional.empty()));


What about these cases:

startingVersion is not set

startingVersion set to Optional.empty

(1) is already covered by testGetStartingVersion. I added (2): testGetStartingVersion_NoOptions

jerrypeng · 2025-10-21T18:48:12Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+   * Validate the protocol at a given version. If the snapshot reconstruction fails for any other
+   * reason than unsupported feature exception, we suppress it. This allows fallback to previous
+   * behavior where the starting version/timestamp was not mandatory to point to reconstructable
+   * snapshot.


This allows fallback to previous behavior where the starting version/timestamp was not mandatory to point to reconstructable snapshot.

Where is this code?

Please see

if (!validateProtocolAt(spark, tablePath, engine, version)) { // When starting from a given version, we don't require that the snapshot of this // version can be reconstructed, even though the input table is technically in an // inconsistent state. If the snapshot cannot be reconstructed, then the protocol // check is skipped, so this is technically not safe, but we keep it this way for // historical reasons. try { streamingHelper.checkVersionExists( version, /* mustBeRecreatable= */ false, /* allowOutOfRange= */

huan233usc · 2025-10-21T23:18:12Z

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java

+        /* catalogTableOpt= */ Option.empty(),
+        options,
+        /* snapshotAtSourceInit= */ snapshot,
+        /* metadataPath= */ tablePath + "/_checkpoint",


Is it the location of streaming checkpoint?

This is the location of PersistedMetadata for advanced schema evolution.

huan233usc · 2025-10-27T04:55:37Z

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java

+      // Attempt to construct a snapshot at the startingVersion to validate the protocol
+      // If snapshot reconstruction fails, fall back to old behavior where the only
+      // requirement was for the commit to exist
+      TableManager.loadSnapshot(tablePath).atVersion(version).build(engine);


Actually we should define a method in streaming helper (maybe called LoadSnapshotAt)

startingversion

9e4b99d

huan233usc requested review from gengliangwang, huan233usc and tdas October 17, 2025 00:59

zikangh changed the title ~~[kernel-spark] Add getStartingVersion() to support Kernel-based DSv2 streaming (InitialSnapshot Part I)~~ [kernel-spark] Add getStartingVersion() for obtaining the initial offset for DSv2 streaming Oct 17, 2025

huan233usc reviewed Oct 20, 2025

View reviewed changes

gengliangwang reviewed Oct 20, 2025

View reviewed changes

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java Show resolved Hide resolved

huan233usc reviewed Oct 20, 2025

View reviewed changes

gengliangwang reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Show resolved Hide resolved

gengliangwang reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Outdated Show resolved Hide resolved

gengliangwang reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Outdated Show resolved Hide resolved

gengliangwang reviewed Oct 20, 2025

View reviewed changes

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkMicroBatchStream.java Show resolved Hide resolved

huan233usc reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Show resolved Hide resolved

zikangh mentioned this pull request Oct 20, 2025

[Feature Request] Improve Kernel error handling #5369

Open

8 tasks

address comments

6674764

zikangh requested review from gengliangwang and huan233usc October 20, 2025 20:23

gengliangwang reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Outdated Show resolved Hide resolved

gengliangwang approved these changes Oct 20, 2025

View reviewed changes

huan233usc reviewed Oct 20, 2025

View reviewed changes

...rk/src/test/java/io/delta/kernel/spark/read/SparkMicroBatchStreamGetStartingVersionTest.java Outdated Show resolved Hide resolved

huan233usc reviewed Oct 20, 2025

View reviewed changes

jerrypeng reviewed Oct 21, 2025

View reviewed changes

address comments

8fb11d8

zikangh requested review from huan233usc and jerrypeng October 21, 2025 22:07

compilation issue

1acba48

huan233usc reviewed Oct 21, 2025

View reviewed changes

huan233usc approved these changes Oct 21, 2025

View reviewed changes

huan233usc requested review from jerrypeng and raveeram-db and removed request for jerrypeng October 24, 2025 17:25

huan233usc reviewed Oct 27, 2025

View reviewed changes

Uh oh!

[kernel-spark] Add getStartingVersion() for obtaining the initial offset for DSv2 streaming #5356

Are you sure you want to change the base?

[kernel-spark] Add getStartingVersion() for obtaining the initial offset for DSv2 streaming #5356

Uh oh!

Conversation

zikangh commented Oct 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

Uh oh!

zikangh commented Oct 17, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

huan233usc Oct 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

zikangh commented Oct 17, 2025 •

edited

Loading

huan233usc Oct 20, 2025 •

edited

Loading