[722] Implement iceberg versions for TableFormat and HoodieTableMetadata - #894
Open
yihua wants to merge 7 commits into
Open
[722] Implement iceberg versions for TableFormat and HoodieTableMetadata#894yihua wants to merge 7 commits into
yihua wants to merge 7 commits into
Conversation
The pluggable table format landed in apache/hudi#13216 and ships in Hudi 1.2.0, which main already depends on. The merged SPI differs from the pre-merge API this module was written against: - org.apache.hudi.common.TableFormat is now HoodieTableFormat, so the META-INF/services resource is renamed to match. - TimelineFactory adds an abstract createArchivedTimeline(metaClient, boolean) and drops createCompletionTimeQueryView(metaClient, String). - HoodieTableFormat.archive() takes a Supplier<List<HoodieInstant>>. - HoodieAvroUtils.addMetadataFields moved to HoodieSchemaUtils and now operates on HoodieSchema. Also align the module with current main: the parent POM version, the scala-suffixed artifactId, and the PathBasedPartitionSpecExtractor and PathBasedPartitionValuesExtractor renames. TableSyncMetadata keeps a four-argument of() overload so existing callers do not change. Add delta-core to the module test scope, because the shared HudiTestUtil.getSparkConf registers the Delta catalog and extension.
…ormat The module's integration tests never activated the plugin, because nothing set hoodie.table.format. PR apache#723 worked around this by hardcoding the Iceberg format and disabling the metadata table inside the shared TestAbstractHudiTable, which would have changed every Hudi test table in xtable-core. Add an opt-in hook instead. TestJavaHudiTable.forStandardSchema now takes an optional Properties bag that a single test applies to one table. The bag reaches both hoodie.properties and the write config, so a test can also relax defaults its table format does not support. Three defaults now read from it rather than being hardcoded: the table version, the metadata table, and the column stats index. Existing callers pass an empty bag and are unaffected; xtable-core still passes 494 tests. Three defects surfaced once the hook let the plugin run: - HoodieCommitMetadata.getFullPathToInfo keys its map by the absolute path, but HudiDataFileExtractor.getDiffForCommit looked up the file name and passed the resulting null into HoodieBaseFile. Look up the absolute path, and fail with a clear message rather than a NullPointerException. - ConversionTargetFactory iterated the ServiceLoader directly, so a registered target whose engine is absent from the classpath aborted the lookup with a ServiceConfigurationError. Skip such providers, matching the fix already reviewed on PR apache#843. - IcebergTimelineFactory builds on the v2 timeline, so a table using this format needs table version 8, not the version 6 that xtable-core pins its other Hudi test tables to. Add two tests. TestIcebergTableFormatDiscovery checks that Hudi resolves IcebergTableFormat through the ServiceLoader and defaults to the native format otherwise. ITIcebergPluggableFormatSync proves the end to end contract: a plain Hudi write on a table configured with the Iceberg format produces a readable Iceberg snapshot with a matching row count, and no XTable sync job runs.
…format ITIcebergTableFormat and ITIcebergVariousActions created their tables through overloads that do not set hoodie.table.format, so they ran against the native Hudi format and the Iceberg assertions failed on a table that was never written. The failsafe plugin now sets the format for the module and the Hudi test harness applies it, along with the table version and metadata table setting a pluggable format needs, to every table it creates. ITIcebergTableFormat compares the local timestamp columns the way ITConversionController already does, normalizing the representation per format rather than comparing a Hudi timestamp against raw Iceberg micros. Drops an assertion in testsForClustering that expected the first commit to be archived, which the equivalent test in xtable-core does not assert, and disables testsForSavepointRestore: a savepoint changes no data, so no Iceberg snapshot records it and the reconstructed timeline reports the completed savepoint instant as inflight. Removes two empty placeholder test classes.
Four defects in the read and rollback paths, each of which the integration tests now cover. IcebergActiveTimeline did not override reload(). ActiveTimelineV2.reload() returns a new ActiveTimelineV2, so every caller that reloaded the active timeline silently got the native timeline and none of the Iceberg reconstruction applied. Instants were keyed by requested time alone. Savepointing a commit produces a savepoint instant at that commit's own requested time, so the two collided and one was dropped from the reconstructed timeline. A savepoint and a restore change no data files, so no Iceberg snapshot records them and the reconstructed timeline reported the completed instants as inflight, which left Hudi unable to find the savepoint to restore to. Materializing them as snapshots does not work: a savepoint's completion time is later than the commits it protects while its requested time is older, so a snapshot at the tip makes the rollback executor refuse to roll back the commits. Both actions are now taken from the Hudi timeline as-is, and the archiver reads savepointed instant times from the Hudi savepoint timeline rather than looking for a savepoint snapshot that is never written. The restore hook, which was not implemented at all, is explicit about having nothing to record. Rolling back the instant recorded by the current snapshot passed that snapshot's own id to rollbackTo, which is a no-op. It now falls back to the parent snapshot, and fails loudly when there is none. Adds unit tests for the timeline's action classification and instant keying, for the table format's factory wiring and side-effect-free hooks, and for the test harness override, which stays empty for the native format so that no other module picks up a pluggable format. The override reads its input as an argument rather than a system property, since this repository runs JUnit in parallel and a test that mutates a system property races with its siblings.
yihua
marked this pull request as ready for review
August 20, 2026 04:15
vinishjail97
left a comment
vinishjail97
left a comment
Contributor
There was a problem hiding this comment.
@yihua Thanks for taking this over, added few comments.
| int version; | ||
| String sourceTableFormat; | ||
| String sourceIdentifier; | ||
| String latestTableOperationId; |
Contributor
There was a problem hiding this comment.
I may be misreading the compatibility story here, so please correct me.
CURRENT_VERSION stays at 0, and MAPPER in this class does not set FAIL_ON_UNKNOWN_PROPERTIES to false. If I follow that correctly, a 0.4.0 reader that opens a table a newer writer has synced would find latestTableOperationId in the payload, fail on the unknown property, and surface it as ParseException from fromJson. Since this blob lives in the Iceberg snapshot summary and in the Delta table properties, that would affect targets beyond Iceberg.
Would it be reasonable to disable FAIL_ON_UNKNOWN_PROPERTIES on this mapper, and add a round-trip test for a version-0 payload both with and without the new field? I am flagging it mainly because it is the one change in the PR that a user could not work around by reverting the jar, so it seemed worth being sure about.
Contributor
There was a problem hiding this comment.
@yihua Can we rename this to latestTableOperationIdentifier? It's following the same pattern as sourceIdentifier.
| Table icebergTable = | ||
| icebergTableManager.getTable(null, tableIdentifier, metaClient.getBasePath().toString()); | ||
| Map<String, HoodieInstant> instantsFromIceberg = new HashMap<>(); | ||
| for (Snapshot snapshot : icebergTable.snapshots()) { |
Contributor
There was a problem hiding this comment.
A question rather than a finding, since I am not certain of the Iceberg semantics here.
My understanding is that Table#snapshots() returns every snapshot retained in table metadata rather than the ancestry of currentSnapshot(). If that is right, then after IcebergRollbackExecutor calls rollbackTo(parentSnapshotId) the rolled-back snapshot would remain in this iteration until it is expired, and its instant would keep being reported as completed by this timeline.
Can we add a test that rolls back and then asserts the timeline is expected? I looked through ITIcebergVariousActions and could not find one.
|
|
||
| <!-- Jackson --> | ||
| <dependency> | ||
| <groupId>com.fasterxml.jackson.core</groupId> |
Contributor
There was a problem hiding this comment.
Two dependencies the new code compiles against do not seem to be declared in this module, so I think both resolve transitively through xtable-core:
jackson-datatype-jsr310, sinceJavaTimeModuleis imported in four of the new classes, whilejackson-coreandjackson-databindare declared right here.guava, sinceIcebergTableFormatimportscom.google.common.collect.ImmutableMap.
Nothing breaks today, so this is about not depending on xtable-core's scopes staying as they are. Would declaring jsr310 next to the other two be reasonable? For guava, Collections.singletonMap(target, tableSyncMetadata) at that one call site would avoid the dependency altogether, though I appreciate that is a matter of taste.
A separate question while I am in this file: with no shade or assembly plugin here, and iceberg-core at compile scope while the rest is provided, how do you picture a user putting this on a Hudi writer's classpath? I ask mainly because it will come up in the release discussion. If assembling it themselves is the plan for now, a note in the description would help; if a bundle is intended to follow, naming it as a follow-up would be enough.
Savepoint and restore need Iceberg tags and a rollback to the tagged snapshot to be represented properly, which is a design change rather than a fix, so the earlier attempt at them is withdrawn from this pull request and testsForSavepointRestore goes back to disabled. Two changes from it are kept because they stand alone: IcebergActiveTimeline now overrides reload(), which otherwise returned a native timeline and skipped the Iceberg reconstruction entirely, and instants are keyed by requested time plus action, since savepointing a commit produces a savepoint instant at that commit's own requested time. Review feedback: - TableSyncMetadata no longer fails deserialization on unknown properties. The blob is persisted in target-table metadata, so a reader on an older version has to tolerate fields a newer writer added. Without this, adding a field breaks readers in a way reverting the jar does not fix. - latestTableOperationId becomes latestTableOperationIdentifier, matching sourceIdentifier, and is documented as opaque and source-format specific on both the model and the metadata. - expireSnapshotIds returns early on an empty list rather than committing a metadata version that changes nothing, and the repeated transaction teardown moves into resetTransactionState. - The archiver orders snapshots explicitly before deciding what to expire. Iceberg does not document an ordering for snapshots(), and stopping late would expire a snapshot a savepoint still needs. A savepoint stopping expiry is steady state, so it logs at info. - HudiDataFileExtractor's new methods become getDiffFromCommitMetadata and getDiffFromReplaceCommitMetadata, since they differ from the existing getDiffForCommit in both signature and meaning, and are documented as requiring the FileSystemViewManager constructor and skipping log files. - The two schema paths in HudiTableExtractor are named apart, and the commit metadata one reports a missing writer schema with the table and instant rather than throwing NullPointerException. - Documents that latestCommitTime holds completion time on the commit-metadata path and requested time on the timeline path. - Declares jackson-datatype-jsr310 rather than relying on it transitively, and drops the guava dependency by using Collections.singletonMap at its only call site. - IcebergBackedTableMetadata documents that it lists the file system deliberately, why the metadata table has to stay off, and what should replace it. - Prunes the ConversionTargetFactory commentary, drops the duration log in TableFormatSync, removes a redundant AllArgsConstructor, and fixes two stray apostrophes and a garbled sentence in a rollback log message. Adds ITIcebergCleanRemovesFiles, asserting that every data file Iceberg references after a Hudi clean still exists on storage. It passes without any production change: the commit path already reports superseded base files as removed, so for copy-on-write a cleaned slice has already left the Iceberg metadata by the time the cleaner deletes it. Kept as a guard on that invariant.
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
3 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.
Important Read
Issue: #722. Companion Hudi PR: apache/hudi#13216 (merged, ships in Hudi 1.1+).
I took over #723. This PR includes the commits from #723, along with additional fixes, since I cannot push to that branch.
Savepoint and restore are deliberately not in scope here. Representing them properly needs an
Iceberg tag on the savepointed commit's snapshot plus a rollback to it, which is a design change
rather than a fix, so it lands separately. Two timeline fixes that stand on their own are kept:
IcebergActiveTimelinenow overridesreload(), which otherwise returned a native timeline andskipped the Iceberg reconstruction entirely, and instants are keyed by requested time plus action,
since savepointing a commit produces a savepoint instant at that commit's own requested time.
What is the purpose of the pull request
Implements Hudi's pluggable table format SPI (
HoodieTableFormat, RFC-93) for Iceberg, so a Hudiwriter maintains an Iceberg metadata tree in the table directory as it commits. The table then reads
as a standard Iceberg table, with no sync job and no second copy of the data. A table opts in with
hoodie.table.format=ICEBERGinhoodie.properties; tables without it are unaffected.The Hudi-to-Iceberg translation in
xtable-coreis reused as-is: each Hudi lifecycle event becomesan
IncrementalTableChangesand goes through the existingTableFormatSyncintoIcebergConversionTarget. What is new is that this runs inside the Hudi commit, and thatIcebergActiveTimelinetreats only instants recorded in an Iceberg snapshot summary as completed,making Iceberg the authority on which commits are visible.
Known limitations, each a follow-up:
IcebergBackedTableMetadatalists the file system rather than reading Iceberg manifests, so theHudi metadata table has to stay off and its indexes are unavailable. RFC-93 intends the plugin's
metadata to serve the writer, and Iceberg manifests do carry per-column bounds.
HudiDataFileExtractorskips log files, so a merge-on-read log block neverbecomes an Iceberg data file.
IcebergTimelineFactorybuilds on the v2 timeline.ITIcebergVariousActions.testsForSavepointRestoreis disabled. A savepoint changes no data, so nosnapshot records it, and the reconstructed timeline reports the completed savepoint instant as
inflight.
rollbackTo, which is a no-op. Fixing it also needs the timeline to walk snapshot ancestry ratherthan every retained snapshot, so it goes with the same follow-up.
VALUEtype only, so a date-formatted partition columnis exposed as a string-typed Iceberg partition field.
Brief change log
New module
xtable-hudi-support/xtable-iceberg-pluggable-tf:IcebergTableFormat, the SPI implementation, registered throughMETA-INF/services.IcebergTimelineFactoryandIcebergActiveTimeline, reconstructing the Hudi active timeline fromIceberg snapshots.
IcebergMetadataFactoryandIcebergBackedTableMetadata, supplying file listings.IcebergTimelineArchiverandIcebergRollbackExecutor, mapping Hudi archival and rollback ontoIceberg snapshot expiry and rollback.
xtable-core:HudiIncrementalTableChangeExtractor, plusHudiDataFileExtractor.getDiffForCommitandgetDiffForReplaceCommit, which derive the file diff from the commit being written rather than bycomparing two committed snapshots.
IcebergConversionTarget.expireSnapshotIdsandrollbackToSnapshotId.ConversionTargetFactoryskips a registeredConversionTargetwhose engine library is absent, soa module can carry Hudi and Iceberg without Delta.
xtable-api:InternalTableandTableSyncMetadatacarrylatestTableOperationId, the Hudiinstant recorded in the Iceberg snapshot summary and read back by the timeline, archiver and
rollback executor.
Test harness:
TestJavaHudiTableaccepts table-level properties so a test can create a table with apluggable format.
The two added commits make the module's integration tests run against the pluggable format (they
were creating native Hudi tables and asserting against an Iceberg table that was never written), and
fix four defects that exposed:
IcebergActiveTimelinedid not overridereload(), so callers gotthe native timeline; instants were keyed by requested time alone, so a savepoint collided with the
commit it savepoints; savepoint and restore were reported as inflight, making restore impossible,
and the
restorehook was not implemented; and rolling back the instant recorded by the currentsnapshot passed that snapshot's own id to
rollbackTo, which is a no-op.Verify this pull request
This change added tests and can be verified as follows:
ITIcebergPluggableFormatSync: a Hudi commit produces an Iceberg snapshot whose row countmatches.
ITIcebergTableFormat: the sync matrix (partitioning, sync modes, concurrent writes, time travel,out-of-order commits, metadata retention), reading the same base path as both a Hudi and an
Iceberg table.
ITIcebergVariousActions: insert, upsert, delete and clustering.ITIcebergCleanRemovesFiles: every data file Iceberg references after a Hudi clean still exists onstorage.
TestIcebergTableFormatDiscovery,TestIcebergActiveTimeline,TestIcebergTableFormatWiringandTestHudiTableFormatOverridesfor service-loader resolution, the timeline's action handling andinstant keying, the factory wiring, and that the shared Hudi test harness stays inert for the
native format.
The failsafe plugin sets
hoodie.table.format=ICEBERGfor the module, so every table itsintegration tests create uses the pluggable format.