Delta: Updating Delta to Iceberg conversion

[vladislav-sidorovich]

Current PRs contains initial version of the code to update of the existing functionality: https://iceberg.apache.org/docs/1.4.3/delta-lake-migration/ to the recent Delta Lake version (read: 3, write: 7). The motivation of the PR is to receive the earliest feedback from the community.

Note: The PR doesn't remove the old logic but adds new Interface implementation, so it will be easier to compare/review. Also base on the usage scenario of the module, such approach will not introduce any issues.
More detailed development doc.

The PR scope:

1. Support existing interface
2. Uses Delta Lake kernel library instead of deprecated Delta Lake standalone
3. Contains the basic flow
4. Converts all data types
5. Converts table schema and partitions spec
6. Support only INSERT operation (Delta Lake Add action)
7. Support UPDATES and DELETS (Delta Lake Remove action)
8. Support Delta VACUUM scenario
9. Support DVs

Future steps:

1. Support All Delta Lake actions
2. Support All Delta Lake features (column mapping, generated columns and so on)
3. Handle Edge cases for partitions and Generated columns
4. Handle Schema evolution
5. Incremental Conversion (from/to a specific Delta Version)

Tests:
Unit-tests: contains all supported datatypes including complex arrays and structures.
Integration-tests: contains inserts only scenario with Spark 3.5. The test must be updated for newer Delta Lake version once the previous solution will be deleted from the code.

In the following PRs, I will add all the tables from: Delta golden tables

[anoopj]

Thank you for the PR. Moving to the Delta kernel is a great improvement. Here is my initial feedback.

[anoopj]

We are using internal APIs of the kernel. This is fragile - can we refactor this to use the public APIs instead? Snapshot, Table etc. Or are we doing this because we are trying to preserve the table history during the conversion? I would try to avoid this as much as possible.

[vladislav-sidorovich]

No, there are no public API available for these purposes we need.

Yes, I want go through table history step by step, so we will have exactly the same granularity in the history.

At the same time it's quite safe to use an internal API because it's depends on the Delta protocol which is stable.

[anoopj]

The internal APIs can change or disappear without any notice. I would think hard about avoiding dependencies on internal APIs, including changing semantics. (e.g. not preserving all the history by default).

[vladislav-sidorovich]

I moved all the internal delta API usage to InternalDeltaKernelUtils so it will be easier to refactor it in the future.

Additionally, InternalDeltaKernelUtils now represents all the internal API that missed in public api and can be used as the list for contributions to delta kernel lib.

[HonahX]

Thanks for the great effort in moving to delta kernel given that delta standalone is archived and deprecated! Left some comments, please let me know WDYT!

[HonahX]

Does the log clean-up always to happen with the VACUUM? Will there be case where the old logs are not cleaned up and our conversion fails because of the missing data file (deleted by VACUUM)?
Seems this indicate some requirement for conversion to work when VACUUM is called on the source table, could you please elaborate on the scope of supporting vacuum in this PR? Thanks!

[vladislav-sidorovich]

The short answers:

1. No, logs clean-up is not the same as VACUUM
2. The conversation will not failed.

Explanations:
The logs clean-up and VACUUM are 2 parallel and different processes. In this test I intentionally delete some logs (not all) to make the delta versions not re-creatable and not continuous (started from table creation). For example in this test Delta versions 3-9 are exist but are not re-creatable because the data files were deleted by VACUUM.
So the first re-creatable version in this test is 10, this version used as initial version in the conversion and the test is for that scenario.

To sum-up:

1. VACUUM and logs clean-up are 2 different processes.
2. In this test I first execute VACUUM with RETAIN 0 HOURS.
3. After some DML operation I simulated (as written in the comment) logs clean-up.
4. Verify that conversion works as expected.

[HonahX]

In the old implementation, the commitInitialDeltaSnapshotToIcebergTransaction, will first to find the actual earliest reconstructable version that have all the data files available. That could help reduce the conversion failure due to VACUUM cleaned up files and breaks time travel ability. What's the design choice behind not using that in the new impl?

[vladislav-sidorovich]

The intention is to support VACUUM operation as well and there is a dedicated test for it org.apache.iceberg.delta.TestSnapshotDeltaLakeKernelTable#testConversionAfterVacuum.

The logic is the very similar to the previous one:

1. Find first retractable commit => Create Iceberg commit.
2. Go 1 by 1 per Delta version after the recreatable commit.

The diff is in commitDeltaSnapshotToIcebergTransaction vs convertEachDeltaVersion. While methods are similar, the way how to collect required data files is different.

[HonahX]

Shall we add tests for unpartitioned table and a table with possible all data types? Also would be good if we have some check similar to checkDataFilePathsIntegrity to verify that data file path is absolute and matches those in delta table.

[vladislav-sidorovich]

There is unit-test to cover all possible data types TestDeltaLakeKernelTypeToType including structs.
Unpartitioned table also covered by unit-test in TestBaseSnapshotDeltaLakeKernelTableAction. Actually, I tried to move more logic to unit tests.
I will also add more tables from https://github.com/delta-io/delta/tree/master/connectors/golden-tables/src , so unit-test will be robust. I removed these tables from the initial PR to reduce number of files in the PR.

Regarding checkDataFilePathsIntegrity I'm not 100% sure. There is no simply utility method that will help to collect all data files, so it means we will need to introduce a logic into the test similar to one we have in the conversion. At the same time this test has checkSnapshotIntegrityForQuery which verify full data for the table per tag/version. Could you clarify what test scenarios is missing and what do you want to test more?

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[laskoviymishka]

@vladislav-sidorovich is this ready to one more round of review?

[vladislav-sidorovich]

@vladislav-sidorovich is this ready to one more round of review?

The initial inputs from you were addressed. But let me double check everything this weekends to avoid double work from you and other reviewers.

[vladislav-sidorovich]

I went though the comments and the code. Let me sum-up the current state:

1. The code was/is ready for next review cycle. Some comments are not resolved because I would like an aka (and resolve) from the author.
2. There is backlog to add more features to the module according to the plan in the PR description.
3. The current version works and functional (with known limitations). I believe it's more pragmatic to get an alignment with community on the current version before increasing the code base with additional logic.

@anoopj

1. The concern regarding usage of internal delta API is actual and valid. As we agreed yearly, I will finish full conversion logic and then collect the full list of the internal API. With the list we will decide if it will be contribution to DeltaKernel or some in-house replacement.

@HonahX

1. I left the comments regarding VACUUM tests. The tests are robust
2. I left the comment regarding VACUUM logic handling
3. All data types (including Variant) tests in unit test. I left the comment.

@laskoviymishka

1. Topic about validate-from snapshot to currentSnapshot() - is not completely clear to me. So, it can be some work here.
2. Testing of deletion vectors per version explained (there is integration test).
3. The blockers from the big summary comment:
   1.1 DV-update - resolved
   1.2 Per-ColumnarBatch commits break - fixed
   1.3 Unhandled metaData/protocol actions - fixed
   1.4 DV test asserts presence - explained

@anoopj @HonahX @laskoviymishka I would love to get next cycle of your review and input. Thank you for the contributions to the PR.

[laskoviymishka]

Thanks for sticking with the review iterations. I went back through the existing threads before adding anything here, so I’m not reopening what’s already settled.

I like the overall plan: land the Kernel path as an experimental, opt-in utility (snapshotDeltaLakeKernelTable) while keeping the standalone action as the default. I’m also fine with the M1–M7 direction.

For this PR, my merge bar: every path reachable through the new utility either works or fails clearly. Nothing should silently write the wrong table.

Things I’d fix here:

• The DV OutputFileFactory is fixed at (1, 1) and reused for a new BaseDVFileWriter per AddFile. With multiple DVs, files can collide and overwrite each other. The factory is also never closed.
• execute() returns a method-reference lambda instead of ImmutableSnapshotDeltaLakeTable.Result. Besides the int vs long mismatch, it diverges from the Immutables contract used by the standalone implementation.
• icebergPropertiesBuilder is an instance ImmutableMap.Builder, so duplicate keys throw when user properties overlap Delta config, or when execute() runs twice. A local map with later-wins semantics is safer.
• The golden-table test only checks that execute() doesn’t throw. It should assert at least schema, row count, and partitioning, or be disabled with a follow-up issue.

One question: does testConversionAfterVacuum cover a Remove whose data file was physically vacuumed, or only log cleanup? If the file itself is gone, that’s the case I’d want covered because it exercises the exists() / metrics read path.

The typo, Set.of vs relocated ImmutableSet, and unused @TempDir are minor.

For the larger follow-ups, I’d move them out of the dev doc and into GitHub issues:

• migrate off io.delta.kernel.internal.*; I’d make this the gate for turning the Kernel path into the default
• retire the standalone implementation and wire the Kernel path into DeltaLakeToIcebergMigrationActionsProvider
• track the deferred work: VACUUM/Remove via AddFile.stats, full DV update semantics, and schema evolution

Once the correctness issues are fixed and the follow-ups are tracked, happy to take another pass and approve.

[laskoviymishka]

The factory is pinned at partitionId=1, taskId=1, and convertDeltaDVsToIcebergDVs spins up a fresh BaseDVFileWriter per AddFile against this same factory.

Since the factory's counter is the only thing varying the filename and each writer starts from the same fixed seed, the Puffin DV files can land on colliding paths and overwrite each other across files — so we can't really claim correct DV conversion without fixing this alongside the DV double-delete above. Sharing one factory so the counter increments across all DV writers is what the code looks like it intended. OutputFileFactory is also Closeable and this one is never closed, worth handling in the same pass.

[vladislav-sidorovich]

I'm not sure I get an issue with the files colliding paths. This method used for each DV puffin file generation: https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java#L106

So, fileCount.incrementAndGet() will always generate next sequential (and unique) file name. It seems that partitionId=1, taskId=1 are need for execution in cluster (on different workers), with the current conversion scenario the usage seems valid. Do I miss anything?

[laskoviymishka]

The schema-evolution guard only runs inside this isCommitInfo(row) branch — so a version that carries no commitInfo row skips the check entirely. Delta doesn't guarantee a commitInfo per version, and the schema is fixed once from initialDeltaSnapshot (line 194), so an ADD/RENAME/DROP COLUMN landing in a commitInfo-less version would convert silently against the wrong schema and write a truncated Iceberg table.

That's the silent-corruption case I'd want closed before merge. I'd lift the schema-consistency check out of the per-row commitInfo parsing and verify the schema is stable across the whole [minimalAvailableDeltaVersion, latestDeltaVersion] range up front — that way the guard doesn't depend on an action Delta may not have written. wdyt?

[vladislav-sidorovich]

Good point, let me rewrite it to rely on metaData

[laskoviymishka]

icebergPropertiesBuilder is an instance field that accumulates across calls, and ImmutableMap.Builder.build() throws on duplicate keys.

buildTablePropertiesWithDelta puts the internal props and then putAll(configuration), on top of whatever the user passed via tableProperties/tableProperty. Any overlap — a user prop that also appears in the Delta config, or a second execute() on the same action — throws IllegalArgumentException. I'd make this a local HashMap inside the build method, populate user props → Delta config → required Iceberg props so later wins, and return ImmutableMap.copyOf(...).

[laskoviymishka]

goldenDeltaTableConversion only asserts that execute() doesn't throw — every golden table passes as long as conversion completes, regardless of whether the result is correct.

For an experimental path that's especially risky, because green CI then implies coverage that isn't actually there. I'd either load the resulting Iceberg table and assert schema, row count, and partitioning against the source, or mark it @Disabled("follow-up #<issue>") so it's honest about what it does and doesn't verify. This is the natural place to catch a silently-wrong DV or Remove conversion.

[vladislav-sidorovich]

Thanks for sticking with the review iterations. I went back through the existing threads before adding anything here, so I’m not reopening what’s already settled.

.....

Once the correctness issues are fixed and the follow-ups are tracked, happy to take another pass and approve.

@laskoviymishka Thank you for the review and the summary comment.

1. The DV OutputFileFactoryis fixed at(1, 1)...

The following sequence of action

1. Add(file_A, DV_1)
2. Remove(file_A, DV_1)
3. Add(file_A, DV_2)

is possible in multiple Delta commits and handled in io.delta.kernel.internal.replay.ActiveAddFilesIterator by UniqueFileActionTuple. At the same time this sequence is impossible in the scope of single Delta snapshot and handled by DMLWithDeletionVectorsHelper.scala for writes, and ConflictChecker.scala for reads (see lazy val addedFilePathToActionMap: Map[String, AddFile]). The conversion works snapshot by snapshot, so it's correct.

The factory is not closable, but DVFileWriter dvWriter is closed in try-with-resource section in convertDeltaDVsToIcebergDVs.

2. execute() returns a method-reference lambda instead ...

Done.

3. icebergPropertiesBuilder is an instance ImmutableMap.Builder ...

throw on duplicate keys is an expected behavior. It’s never happened in a proper execution but will throw an exception for invalid usage. This strategy also aligns with your previous comments regarding fail with explicit message vs silent corruption.

4. The golden-table test only checks that execute() doesn’t throw ....

It's actually an interesting topic to discus. My ideas was the following: use unit-tests TestBaseSnapshotDeltaLakeKernelTableAction to verify conversion logic like supported / not supported features, valid/invalid parameters, basically logic I wrote for the conversion. Testing the table data is done in the integration tests and moving it to unit-tests will make them much slower. So, I see the following options:

1. Keep golden tables in unit-tests and follow the pattern to throw an exception instead of silent corruption.
2. Move golden tables to integration-tests and verify data

What do you think?

Other topics:

1. VACUUM - I'm not 100% aligned there. I believe VACUUM logic is covered and work properly. Please let me know if I missed anything.
2. testConversionAfterVacuum - tests to processes VACUUM and logs compaction. VACUUM deletes data files, and log compaction deletes logs before check points. So, in this test we have commits pointing to deleted files but it works properly because conversion support VACUUM and starts from the latest checkpoint.
3. Let me go again through all the comments to double check that I didn't miss anything.
4. I will move the implementation doc to issues as was requested.
5. I will move all the usage of internal Delta API to helper classes to have it in 1 place because it was highlighted by at least 2 reviewers .

[laskoviymishka]

Thanks @vladislav-sidorovich for write up!

Dived into the DV-update path I flagged as a blocker last round: Add+Remove of the same path, the core DELETE/MERGE/UPDATE path on DV-enabled tables.

That path now produces a valid snapshot. I traced it: each Delta DV-delete is remove(file) + add(file) with a cumulative DV, and the converter's removeRows(file) drops the prior DV while the new cumulative DV replaces it — so the earlier "addRows + removeRows on the same file" concern is reconciled in practice, even if implicitly rather than via an explicit removeDeletes.

To convince myself, I built a repro: single data file (range(0, 10000, 1, 1)), DELETE id = 1 then DELETE id = 2 as two commits. This is the exact case my last note said the tests couldn't catch — row content can't see it, because the v1 DV {1} is a subset of the v2 DV {1,2}, so the union Iceberg applies is correct (9998 rows) whether or not the stale DV is cleaned up. The assertion that actually pins it down is the delete-file count:

assertThat(spark.sql("SELECT * FROM " + table + ".delete_files").count()).isEqualTo(1L);

Green: exactly one DV on the file in the latest snapshot, no dangling superseded DV, V3 one-DV-per-file invariant holds. The other blockers from last round look addressed too (per-Delta-version commit atomicity, schema-evolution handling).

Stepping back, though, this is incremental, not a finish line, and the plan for the rest already exists: the Future steps in the description (all Delta actions; all Delta features like column mapping / generated columns; partition + generated-column edge cases; schema evolution; incremental conversion) plus the golden-tables suite and the dev doc. My ask is to make that plan visible and trackable in the open, a tracking issue with sub-issues per step, so the staged work is reviewable as it lands instead of living only in the PR body.

Net: the blocking correctness concern on the central DV path is resolved and verified. Treating this as a staged step, good to keep moving, with the remaining scope tracked openly via the tracking issue + sub-issues rather than silently closed.

Before merging will wait for (@anoopj | @huan233usc) && (@kevinjqliu | @HonahX)