[Kernel]Provide api to return per commit action iterator in get change api #5290

huan233usc · 2025-10-02T22:34:38Z

Which Delta project/connector is this regarding?

Description

This PR introduces api to return per commit action iterator in kernel get change api, it will help to implementing current structure streaming logic using kernel

Case 1: need to find commit boundary and inject offset

Case 2: skip commits without read through iterator: no need to iterate all action in 2.json and directly go to 3.json

How was this patch tested?

Unit

Does this PR introduce any user-facing changes?

No

allisonport-db · 2025-10-15T22:53:41Z

kernel/kernel-api/src/main/java/io/delta/kernel/CommitActions.java

+   * <p>This will also close the actions iterator if it hasn't been fully consumed.
+   */
+  @Override
+  void close() throws Exception;


Do we need this separate close()? is anything allocated other than the iterator?

Or is this just so it can be AutoCloseable?

CommitAction will own the action iterator and transfer the ownership to caller when getActions. I am trying to handle the logic of closing this action iterator if getActions not called.

I made AutoCloseable only for impl class as it is a bit strange if we put it in the interface

allisonport-db · 2025-10-15T22:55:01Z

kernel/kernel-api/src/main/java/io/delta/kernel/CommitActions.java

+   *
+   * @return a {@link CloseableIterator} over columnar batches containing this commit's actions
+   */
+  CloseableIterator<ColumnarBatch> getActions();


I think it needs to be clear whether you can re-call this multiple times (new iterator each time) or if it always returns the same one. I think maybe this is a potential design decision? From an implementation standpoint either should be do-able.

Yes this is a good point, I rethink a bit what will be the best to serve the use case ---

The pseudo code of the actual use case is
try (CommitActions commitActions = ...) {
// First pass: validate and decide
try (CloseableIterator iter = commitActions.getActions()) {
boolean shouldSkip = validateCommitAndDecideSkipping(iter);
}

// Second pass: process the data (re-reads from file)
if (!shouldSkip) {
try (CloseableIterator iter = commitActions.getActions()) {
processActions(iter);
}
}
}

So basically three options

Only allow 1 time call (this implementation)
+Simple in kernel
-Difficult in connector: Connector needs to create a list to collect all add files so that they could do a second pass. "Create a list to collect all add files" may lead to memory pressure if commit file is large and there's no way to get around it.
-The complexity/memory pressure is just pushed to the connector layer

2.1 Allow multiple time call, implementation: kernel collects all batches in a List in CommitAction

+Simple in connector
-Memory pressure in kernel: All batches for a commit must be kept in memory

For large commits with many actions, this could be problematic

The memory is held even if caller only needs one pass
2.2 Allow multiple time call, implementation: kernel re-reads the commit file on each getActions() call

+Memory efficient: No need to cache batches in memory
+Flexible: Caller can choose to call once or multiple times based on their needs
+Supports the two-pass use case naturally
-Slight I/O overhead: Re-reads file if getActions() called multiple times
-More code complex implementation in kernel

Slight I/O overhead could actually be further improved by connector implementing a cache or if needed, we could add a hint parameter like
CommitActions.Builder.withCaching(boolean) or size threshold to let kernel
decide whether to cache in memory or re-read from file.

With those --- seems that 2.2 is the preferred one?

I think 2.2 is preferred. We don't want to hold memory in Kernel, that's something the connector should implement.

In theory, 2.2 should be easy to support, but i'm not sure if it is actually that way considering we reuse internal classes that maybe don't make that sort of practice easy.

allisonport-db · 2025-10-15T22:58:53Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeImpl.java

-        engine, dataPath.toString(), getDeltaFiles(), actionSet);
+  }
+
+  private io.delta.kernel.CommitActions convertToCommitActions(


This seems like a valid approach that will work. But doesn't it probably make more sense to wrap in the other direction? (just a larger refactor to existing code?)

common code - returns a Iterator<CommitActions>
former method - wraps this and flattens the iterators + adds the timestamp/version columns

Updated the implementation to make underlying internal library returns Iterator.
Then let getActions depends on getCommits

huan233usc · 2025-10-21T17:15:30Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeImpl.java

  public CloseableIterator<ColumnarBatch> getActions(
      Engine engine, Snapshot startSnapshot, Set<DeltaLogActionUtils.DeltaAction> actionSet) {
+    validateParameters(engine, startSnapshot, actionSet);
+    // Build on top of getCommits() by flattening and adding version/timestamp columns


getActions now depends on getCommits for the future API consolidation.

raveeram-db · 2025-10-22T19:31:11Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/CommitActionsImpl.java

+    CloseableIterator<ActionWrapper> wrappers =
+        new ActionsIterator(
+            engine, Collections.singletonList(commitFile), readSchema, Optional.empty());
+    ;


nit: extra semi-colon

raveeram-db · 2025-10-22T20:10:34Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/CommitActionsImpl.java

+                  () -> new RuntimeException("timestamp should always exist for Delta File"));
+
+      // Process first batch and cache the iterator
+      ColumnarBatch firstBatch =


Can this be done lazily in getActions? Just get a reference to the firstWrapper here but build out the batch in the getActions?

raveeram-db · 2025-10-22T20:10:38Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/CommitActionsImpl.java

+
+  @Override
+  public synchronized CloseableIterator<ColumnarBatch> getActions() {
+    if (!firstCallDone) {


Given that this is essentially about caching the iterator and transferring ownership, should we use that terminology instead of firstCallDone and firstReadIterator? Perhaps something like cachedIterator and cachedIteratorTransferred ? Would be good to add comments to the member vars above as well as to what they mean.

I reorg this file, feel like using supplier might be cleaner

allisonport-db

Overall approach lgtm, left some suggestions

allisonport-db · 2025-10-22T23:56:08Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/commitrange/CommitRangeImpl.java

+  }
+
+  @Override
+  public CloseableIterator<CommitActions> getCommits(


Can we name this getCommitActions or something actually? This is misleading, I thought maybe it was getting a list of commit files or something

allisonport-db · 2025-10-23T00:15:57Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableChangesUtils.java

+   * @param shouldDropCommitInfoColumn whether to drop the commitInfo column
+   * @return the processed batch
+   */
+  public static ColumnarBatch processAndDropColumns(


validateProtocolAndDropInternalColumns? more descriptive

allisonport-db · 2025-10-23T00:16:47Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableChangesUtils.java

+
+    // Validate protocol
+    int protocolIdx = batch.getSchema().indexOf("protocol");
+    if (protocolIdx >= 0) {


what's the scenario where the protocol column was not read? does that exist (before I think it was impossible)?

I think no, so I make this utils inside CommitActionsImpl and add precondition.

allisonport-db · 2025-10-23T00:20:11Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableChangesUtils.java

+   * @param commits the iterator of CommitActions to flatten
+   * @return an iterator of ColumnarBatch with version and timestamp columns added
+   */
+  public static CloseableIterator<ColumnarBatch> flattenCommitsAndAddMetadata(


I wonder if this would be simpler if we added a util for CloseableIterator<CloseableIterator<T>>::flatMap (or something along those lines)? With unit tests. Then this code should be easier, and I think we could reuse this elsewhere. I really don't like us re-implementing common iterator concepts like flatMap in multiple places, seems dangerous for little bugs

allisonport-db · 2025-10-23T00:27:51Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableChangesUtils.java

+        .withNewColumn(0, new StructField("version", LongType.LONG, false), commitVersionVector)
+        .withNewColumn(
+            1, new StructField("timestamp", LongType.LONG, false), commitTimestampVector);


nit - still use the static fields as before?

allisonport-db · 2025-10-23T00:33:14Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/TableChangesSuite.scala

      testGetChangesVsSpark(tablePath, 0, 2, FULL_ACTION_SET)
    }
  }
+


I think we are missing a lot of important test cases based on the code

call CommitActions::getCommits multiple times, we correctly get a new iterator each time and it contains the same rows

empty commit file? (test the code on CommitActionImpl line 90-95 where there is no actions), also make sure this works for the traditional getActions API to make sure the flatMap iterator is fine for empty iterators

wondering if we should re-run some of the existing tests using this API? Or is it sufficient since we use it underneath for the current one?

Added cases for first 2.
For the third one, I think it is sufficient since we use it underneath for the current one

- Fix typo in comment: 'will null' -> 'with null' - Fix double period in JavaDoc - Add exception handling in hasNext() to ensure cleanup on error - Update test assertions to use === for consistency - Add test case for exception handling during iteration - Verify that outer iterator is closed when exception occurs

…ocumentation - Rename flatMap() to flatten() for more accurate semantics - Add important note in JavaDoc about calling close() even if iterator is not fully consumed - Remove unnecessary 'throws IOException' from close() method - Update all test case names from 'flatMap' to 'flatten' Addresses reviewer feedback: 1. Method name should be 'flatten' not 'flatMap' (semantics) 2. Callers need to know they must call close() for resource cleanup 3. Exception handling in next() is correct as-is

Sync with the rename from flatMap to flatten in Utils.java

huan233usc · 2025-10-23T19:58:12Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/Utils.java

    return new FilteredBatchToRowIter(sourceBatches);
  }

+  /**


from #5388, will merge that one first

#### Which Delta project/connector is this regarding?  - [ ] Spark - [ ] Standalone - [ ] Flink - [x] Kernel - [ ] Other (fill in here) ## Description  Adds a `Utils.flatMap` utility method for flattening nested `CloseableIterator<CloseableIterator<T>>` structures into `CloseableIterator<T>`. It will be helpful for use case like #5290 ## How was this patch tested?  Unit tests ## Does this PR introduce _any_ user-facing changes?  No

huan233usc · 2025-10-23T22:28:31Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/DeltaLogActionUtils.java

      String tablePath,
      List<FileStatus> commitFiles,
      Set<DeltaLogActionUtils.DeltaAction> actionSet) {
-    // Create a new action set which is a super set of the requested actions.


Move to CommitAction's constructor

allisonport-db · 2025-10-23T23:02:03Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/CommitActionsImpl.java

-              .combine(
-                  wrappers.map(
+      // Create supplier that will build the iterator on demand
+      this.iteratorSupplier =


I'm not sure I like this pattern any better, in fact I think I find it quite confusing. Maybe can we come up with some good internal class that abstracts all this away?

PeekableActionIterator(commitFile) { getTimestamp getVersion getNewIterator // this can privately implement a peaking iterator? }

WDYT?

PeekableIterator is a good idea and we could build commit action based on that.

PeekableActionIterator specifically seems a bit over kill.

Updated the impl

huan233usc added 11 commits October 1, 2025 22:44

poc

af96306

poc

63b0bb6

reuse code

49a55e2

Merge branch 'master' into kernel-api

c2479bb

merge master

b3861fe

fix build

5368f96

fix build

0991c4e

fmt

0829218

add test

64f9725

fix

2d1dc9f

add doc

7bd9d76

huan233usc changed the title ~~[POC]Get change api for streaming~~ [Kernel]Provide api to return per commit action iterator in get change api Oct 14, 2025

huan233usc added 3 commits October 14, 2025 13:38

fmt

ad5ccec

doc

35b03da

fix

c22cae7

huan233usc self-assigned this Oct 14, 2025

huan233usc requested review from allisonport-db, raveeram-db and scottsand-db October 14, 2025 20:43

huan233usc added 2 commits October 14, 2025 13:46

clean up

74fc278

fix test

a3e4347

allisonport-db reviewed Oct 15, 2025

View reviewed changes

huan233usc added 6 commits October 20, 2025 16:33

refactor

bd7cd34

refactor

2c1cd78

refactor

5d078dc

simp

811c3cc

fix

4afbe1f

fix

7223a13

huan233usc requested a review from allisonport-db October 21, 2025 02:38

huan233usc commented Oct 21, 2025

View reviewed changes

raveeram-db reviewed Oct 22, 2025

View reviewed changes

allisonport-db reviewed Oct 23, 2025

View reviewed changes

save

e2c2350

huan233usc mentioned this pull request Oct 23, 2025

[Kernel]Introduce flatmap utility method for Closable Iterator #5388

Merged

5 tasks

huan233usc added 6 commits October 23, 2025 11:28

simplify

a280387

Update flatMap to flatten in TableChangesUtils

683b6b8

Sync with the rename from flatMap to flatten in Utils.java

Merge branch 'master' into kernel-api

b3bad95

empty

6b01de6

huan233usc commented Oct 23, 2025

View reviewed changes

huan233usc added 2 commits October 23, 2025 14:38

resourcr

a676055

simplify

9086d87

huan233usc added 2 commits October 23, 2025 15:25

simplify

e79d41f

Merge remote-tracking branch 'origin/master' into kernel-api

b8c0d90

huan233usc commented Oct 23, 2025

View reviewed changes

define static

f37120d

huan233usc requested review from allisonport-db and raveeram-db October 23, 2025 22:35

allisonport-db reviewed Oct 23, 2025

View reviewed changes

huan233usc added 2 commits October 23, 2025 20:47

fix

c866e52

fix

cbfaa61

huan233usc requested a review from allisonport-db October 24, 2025 03:58

Uh oh!

[Kernel]Provide api to return per commit action iterator in get change api #5290

Are you sure you want to change the base?

[Kernel]Provide api to return per commit action iterator in get change api #5290

Uh oh!

Conversation

huan233usc commented Oct 2, 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!

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!

huan233usc Oct 17, 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!

allisonport-db left a comment

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!

allisonport-db Oct 23, 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

huan233usc commented Oct 2, 2025 •

edited

Loading

huan233usc Oct 17, 2025 •

edited

Loading

allisonport-db Oct 23, 2025 •

edited

Loading