metadata.iceberg.table |
(none) |
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
index f8a41f56f28e..82452ed056bd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
@@ -148,6 +148,7 @@ public class IcebergCommitCallback implements CommitCallback, TagCallback {
private final IndexFileHandler indexFileHandler;
private final boolean needAddDvToIceberg;
+ private final boolean syncFullHistory;
// -------------------------------------------------------------------------------------
// Public interface
@@ -202,6 +203,8 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) {
this.indexFileHandler = table.store().newIndexFileHandler();
this.needAddDvToIceberg = needAddDvToIceberg();
+ this.syncFullHistory =
+ table.coreOptions().toConfiguration().get(IcebergOptions.SYNC_FULL_HISTORY);
}
public static Path catalogTableMetadataPath(FileStoreTable table) {
@@ -442,7 +445,7 @@ private void createMetadata(
abandonedLastColumnId,
abandonedNextRowId);
} else {
- createMetadataWithoutBase(
+ recreateMetadata(
snapshotId, abandonedUuid, abandonedLastColumnId, abandonedNextRowId);
}
@@ -459,6 +462,253 @@ private void createMetadata(
}
}
+ /**
+ * Create Iceberg metadata when no usable base metadata exists: either the very first Iceberg
+ * commit for this table, or a recovery after the previous metadata became unusable (format
+ * version change, missing row lineage, Iceberg-layer commit failure).
+ *
+ * By default only the current snapshot is exposed to Iceberg. With {@link
+ * IcebergOptions#SYNC_FULL_HISTORY} the whole retained Paimon history is replayed instead, so
+ * Iceberg readers keep time travel and tags (see apache/paimon#6107).
+ */
+ private void recreateMetadata(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ if (syncFullHistory) {
+ rebuildFullHistory(snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor);
+ } else {
+ createMetadataWithoutBase(snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor);
+ }
+ }
+
+ /**
+ * Rebuild Iceberg metadata from every Paimon snapshot that is still retained, ending at {@code
+ * currentSnapshotId}: create metadata afresh for the earliest retained snapshot, then replay
+ * each following snapshot on top of its predecessor, exactly like live commits would have.
+ * Schemas, tags and (for format version 3) the row-id space therefore accumulate consistently
+ * across the whole replayed history.
+ *
+ *
Each replay step persists its metadata file, so an interrupted rebuild resumes from the
+ * newest already-written metadata on the next commit. Intermediate steps skip the version hint
+ * and the external catalog commit; only the final step publishes, so an external catalog sees a
+ * single transition. Replayed snapshots keep their original Paimon commit timestamps and are
+ * subject to the same retention policy ({@link CoreOptions#SNAPSHOT_NUM_RETAINED_MIN}, {@link
+ * CoreOptions#SNAPSHOT_TIME_RETAINED}, ...) that live commits apply.
+ */
+ private void rebuildFullHistory(
+ long currentSnapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Long earliest = snapshotManager.earliestSnapshotId();
+ long startId = earliest == null ? currentSnapshotId : Math.min(earliest, currentSnapshotId);
+
+ // Resume from the newest existing metadata below the current snapshot, if it is usable.
+ // Anything older than the newest existing file is stale by definition: live commits only
+ // ever read the immediately preceding metadata.
+ long baseId = -1;
+ for (long id = currentSnapshotId - 1; id >= startId; id--) {
+ Path metadataPath = pathFactory.toMetadataPath(id);
+ if (table.fileIO().exists(metadataPath)) {
+ try {
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(table.fileIO(), metadataPath);
+ if (isSameFormatVersion(metadata.formatVersion())
+ && (formatVersion < IcebergMetadata.FORMAT_VERSION_V3
+ || metadata.nextRowId() != null)
+ && coversRetainedPrefix(metadata, id, startId)) {
+ baseId = id;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read existing Iceberg metadata {}, rebuilding history from scratch",
+ metadataPath,
+ e);
+ }
+ break;
+ }
+ }
+
+ long firstWithBase;
+ boolean freshRebuild = baseId == -1;
+ StaleBuild staleBuild = null;
+ if (freshRebuild) {
+ // No usable base: the whole old build is stale. Nothing of it is deleted yet, so
+ // an external catalog that still points at the old metadata keeps a fully readable
+ // table during the entire replay; the old files are cleaned only after the final
+ // step has published. Their references are collected up front, tolerating
+ // unreadable files (that is what triggered some rebuilds in the first place).
+ staleBuild = collectStaleBuild(currentSnapshotId);
+ // a leftover file in the replay range must not survive as a replay step's output:
+ // it may match the step's commit identity (a regenerated build of the same Paimon
+ // snapshot does) while carrying another format or content, so each target is
+ // removed just before its replacement is written
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(startId));
+ createMetadataWithoutBase(
+ startId,
+ inheritUuid,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ startId != currentSnapshotId);
+ firstWithBase = startId + 1;
+ } else {
+ firstWithBase = baseId + 1;
+ }
+
+ for (long id = firstWithBase; id <= currentSnapshotId; id++) {
+ long snapshotId = id;
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ if (freshRebuild && snapshotId != currentSnapshotId) {
+ // see above; the final step replaces its twin through the regular write path
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(snapshotId));
+ }
+ createMetadataWithBase(
+ (removedFiles, addedFiles) ->
+ collectFileChanges(snapshotId, removedFiles, addedFiles),
+ indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX),
+ snapshot,
+ pathFactory.toMetadataPath(snapshotId - 1),
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ snapshotId != currentSnapshotId);
+ }
+
+ if (staleBuild != null) {
+ // only after the replay's final step may have published: a rebuild superseded by a
+ // newer commit leaves the old build in place for that commit's own rebuild
+ Long latestAfterReplay = table.snapshotManager().latestSnapshotId();
+ if (latestAfterReplay != null && latestAfterReplay == currentSnapshotId) {
+ deleteStaleBuild(staleBuild, currentSnapshotId, startId);
+ }
+ }
+ }
+
+ /** File names of the metadata chain being replaced by a from-scratch full-history replay. */
+ private static class StaleBuild {
+ private final List metadataPaths = new ArrayList<>();
+ private final Set manifestLists = new LinkedHashSet<>();
+ private final Set manifests = new LinkedHashSet<>();
+ }
+
+ private StaleBuild collectStaleBuild(long currentSnapshotId) throws IOException {
+ StaleBuild stale = new StaleBuild();
+ Iterator it =
+ pathFactory.getAllMetadataPathBefore(table.fileIO(), currentSnapshotId).iterator();
+ while (it.hasNext()) {
+ Path path = it.next();
+ stale.metadataPaths.add(path);
+ IcebergMetadata metadata;
+ try {
+ metadata = IcebergMetadata.fromPath(table.fileIO(), path);
+ } catch (Exception e) {
+ LOG.warn(
+ "Unreadable Iceberg metadata {} in the build being replaced; its "
+ + "manifests are left to orphan cleanup.",
+ path,
+ e);
+ continue;
+ }
+ for (IcebergSnapshot snapshot : metadata.snapshots()) {
+ String listName = new Path(snapshot.manifestList()).getName();
+ if (!stale.manifestLists.add(listName)) {
+ continue;
+ }
+ try {
+ for (IcebergManifestFileMeta meta : manifestList.read(listName)) {
+ stale.manifests.add(new Path(meta.manifestPath()).getName());
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Unreadable Iceberg manifest list {} in the build being replaced; "
+ + "its manifests are left to orphan cleanup.",
+ listName,
+ e);
+ }
+ }
+ }
+ return stale;
+ }
+
+ /**
+ * Delete what remains of the replaced build after the replay has published: manifests no
+ * replayed metadata references (the replay writes freshly named files, the name check is a
+ * safety net) and metadata files below the replay range, which no replay step overwrote.
+ */
+ private void deleteStaleBuild(StaleBuild stale, long currentSnapshotId, long startId)
+ throws IOException {
+ Set referenced = new HashSet<>();
+ try {
+ IcebergMetadata finalMetadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(), pathFactory.toMetadataPath(currentSnapshotId));
+ for (IcebergSnapshot snapshot : finalMetadata.snapshots()) {
+ referenced.add(new Path(snapshot.manifestList()).getName());
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read the replayed Iceberg metadata for snapshot {}; skipping "
+ + "cleanup of the replaced build.",
+ currentSnapshotId,
+ e);
+ return;
+ }
+ for (String listName : stale.manifestLists) {
+ if (referenced.contains(listName)) {
+ continue;
+ }
+ table.fileIO().deleteQuietly(pathFactory.toManifestListPath(listName));
+ }
+ for (String manifestName : stale.manifests) {
+ table.fileIO().deleteQuietly(pathFactory.toManifestFilePath(manifestName));
+ }
+ for (Path path : stale.metadataPaths) {
+ long version = metadataVersionOf(path);
+ if (version >= 0 && version < startId) {
+ table.fileIO().deleteQuietly(path);
+ }
+ }
+ }
+
+ private static long metadataVersionOf(Path path) {
+ String name = path.getName();
+ if (!name.startsWith("v") || !name.endsWith(".metadata.json")) {
+ return -1;
+ }
+ try {
+ return Long.parseLong(name.substring(1, name.indexOf('.')));
+ } catch (NumberFormatException e) {
+ return -1;
+ }
+ }
+
+ /**
+ * Whether a resume candidate for {@link #rebuildFullHistory(long)} really is the prefix of a
+ * full-history replay. Metadata written while full-history sync was off (e.g. single-snapshot
+ * metadata from a plain rebuild) also passes the format checks, but resuming from it would
+ * silently drop the retained snapshots it does not contain. The candidate is only usable if its
+ * history reaches back to the earliest retained snapshot, or if the newest snapshot it is
+ * missing was already expirable under the snapshot retention policy (i.e. the gap is legitimate
+ * retention trimming, not missing history).
+ */
+ private boolean coversRetainedPrefix(IcebergMetadata base, long baseSnapshotId, long startId) {
+ if (base.snapshots().isEmpty()) {
+ return false;
+ }
+ long oldestInBase =
+ base.snapshots().stream().mapToLong(IcebergSnapshot::snapshotId).min().getAsLong();
+ if (oldestInBase <= startId) {
+ return true;
+ }
+ Snapshot newestMissing = table.snapshotManager().snapshot(oldestInBase - 1);
+ return shouldExpire(newestMissing.id(), newestMissing.timeMillis(), baseSnapshotId);
+ }
+
// -------------------------------------------------------------------------------------
// Create metadata afresh
// -------------------------------------------------------------------------------------
@@ -478,6 +728,22 @@ private void createMetadataWithoutBase(
int lastColumnIdFloor,
long nextRowIdFloor)
throws IOException {
+ createMetadataWithoutBase(
+ snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor, false);
+ }
+
+ /**
+ * @param intermediate whether this metadata is an intermediate step of a {@link
+ * #rebuildFullHistory} replay; intermediate steps skip the version hint and the external
+ * catalog commit, which only the final step publishes.
+ */
+ private void createMetadataWithoutBase(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ boolean intermediate)
+ throws IOException {
SnapshotReader snapshotReader = table.newSnapshotReader().withSnapshot(snapshotId);
Snapshot paimonSnapshot = table.snapshotManager().snapshot(snapshotId);
SchemaCache schemaCache = new SchemaCache();
@@ -486,20 +752,32 @@ private void createMetadataWithoutBase(
SummaryMetrics metrics = new SummaryMetrics();
Set changedPartitions = new HashSet<>();
- List filteredDataSplits =
- snapshotReader.read().dataSplits().stream()
- .filter(DataSplit::rawConvertible)
- .collect(Collectors.toList());
- for (DataSplit dataSplit : filteredDataSplits) {
- changedPartitions.add(dataSplit.partition());
+ DataFilePathFactories dataFilePathFactories =
+ new DataFilePathFactories(fileStorePathFactory);
+ SkippedFiles skippedFiles = new SkippedFiles();
+ for (DataSplit dataSplit : snapshotReader.read().dataSplits()) {
dataSplitToManifestEntries(
- dataSplit, snapshotId, schemaCache, dataFileEntries, dvFileEntries);
-
- for (DataFileMeta paimonFileMeta : dataSplit.dataFiles()) {
- metrics.addedDataFiles++;
- metrics.addedRecords += paimonFileMeta.rowCount();
- metrics.addedFilesSize += paimonFileMeta.fileSize();
- }
+ dataSplit,
+ snapshotId,
+ schemaCache,
+ dataFilePathFactories,
+ dataFileEntries,
+ dvFileEntries,
+ metrics,
+ changedPartitions,
+ skippedFiles);
+ }
+ if (skippedFiles.fileCount > 0) {
+ LOG.warn(
+ "Iceberg metadata for Paimon snapshot {} was created from scratch, but "
+ + "{} data file(s) containing {} row(s) cannot be read without merging "
+ + "(level-0 files, or files shadowed by newer levels in buckets with "
+ + "overlapping key ranges) and were not exported to Iceberg. "
+ + "These rows will appear in Iceberg once compaction rewrites them; "
+ + "trigger a full compaction to export them immediately.",
+ snapshotId,
+ skippedFiles.fileCount,
+ skippedFiles.recordCount);
}
List dataManifestFileMetas = new ArrayList<>();
@@ -576,11 +854,16 @@ private void createMetadataWithoutBase(
// Tags can only be included in Iceberg if they point to an Iceberg snapshot that
// exists. Otherwise, an Iceberg client fails to parse the metadata and all reads fail.
- // Only the latest snapshot ID is added to Iceberg in this code path. Since this snapshot
- // has just been committed to Paimon, it is not possible for any Paimon tag to reference it
- // yet.
- // After https://github.com/apache/paimon/issues/6107 we can add tags here.
- Map refs = new HashMap<>();
+ // This metadata contains exactly one snapshot, so only tags pointing at it are eligible;
+ // that can happen when metadata is rebuilt for an existing snapshot (e.g. the start of a
+ // full history replay, see https://github.com/apache/paimon/issues/6107).
+ Map refs =
+ table.tagManager().tags().entrySet().stream()
+ .filter(entry -> entry.getKey().id() == snapshotId)
+ .collect(
+ Collectors.toMap(
+ entry -> entry.getValue().get(0),
+ entry -> new IcebergRef(entry.getKey().id())));
// keep the identity of the metadata this rebuild replaces, so already loaded readers
// and external catalogs keep refreshing the same table
@@ -633,9 +916,10 @@ private void createMetadataWithoutBase(
throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath);
}
// a delayed callback may still write its metadata (a newer commit extends it), but
- // only the current head may move the hint and the external catalog
+ // only the current head may move the hint and the external catalog; an intermediate
+ // replay step publishes nothing and cleans nothing, the final step covers both
Long latestAtPublish = table.snapshotManager().latestSnapshotId();
- if (latestAtPublish != null && latestAtPublish == snapshotId) {
+ if (!intermediate && latestAtPublish != null && latestAtPublish == snapshotId) {
table.fileIO()
.overwriteFileUtf8(
new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME),
@@ -647,28 +931,62 @@ private void createMetadataWithoutBase(
}
}
+ /** Files skipped by a from-scratch export because they cannot be read without merging. */
+ private static class SkippedFiles {
+ private long fileCount;
+ private long recordCount;
+ }
+
private void dataSplitToManifestEntries(
DataSplit dataSplit,
long snapshotId,
SchemaCache schemaCache,
+ DataFilePathFactories dataFilePathFactories,
List dataFileEntries,
- List dvFileEntries) {
- List rawFiles = dataSplit.convertToRawFiles().get();
+ List dvFileEntries,
+ SummaryMetrics metrics,
+ Set changedPartitions,
+ SkippedFiles skippedFiles) {
+ boolean rawConvertible = dataSplit.rawConvertible();
+ List rawFiles = rawConvertible ? dataSplit.convertToRawFiles().get() : null;
+ DataFilePathFactory dataFilePathFactory =
+ dataFilePathFactories.get(dataSplit.partition(), dataSplit.bucket());
for (int i = 0; i < dataSplit.dataFiles().size(); i++) {
DataFileMeta paimonFileMeta = dataSplit.dataFiles().get(i);
- RawFile rawFile = rawFiles.get(i);
+ String filePath;
+ String fileFormat;
+ if (rawConvertible) {
+ RawFile rawFile = rawFiles.get(i);
+ filePath = rawFile.path();
+ fileFormat = rawFile.format();
+ } else if (shouldAddFileToIceberg(paimonFileMeta)) {
+ // A split that cannot be read raw as a whole (it contains level-0 files or
+ // overlapping key ranges) can still contain files that the incremental commit
+ // path would have published; dropping the whole split would silently lose their
+ // rows until some future compaction happens to rewrite the files.
+ filePath = dataFilePathFactory.toPath(paimonFileMeta).toString();
+ fileFormat = paimonFileMeta.fileFormat();
+ } else {
+ skippedFiles.fileCount++;
+ skippedFiles.recordCount += paimonFileMeta.rowCount();
+ continue;
+ }
IcebergDataFileMeta fileMeta =
IcebergDataFileMeta.create(
IcebergDataFileMeta.Content.DATA,
- rawFile.path(),
- rawFile.format(),
+ filePath,
+ fileFormat,
dataSplit.partition(),
- rawFile.rowCount(),
- rawFile.fileSize(),
+ paimonFileMeta.rowCount(),
+ paimonFileMeta.fileSize(),
schemaCache.get(paimonFileMeta.schemaId()),
paimonFileMeta.valueStats(),
paimonFileMeta.valueStatsCols());
+ metrics.addedDataFiles++;
+ metrics.addedRecords += paimonFileMeta.rowCount();
+ metrics.addedFilesSize += paimonFileMeta.fileSize();
+ changedPartitions.add(dataSplit.partition());
dataFileEntries.add(
new IcebergManifestEntry(
IcebergManifestEntry.Status.ADDED,
@@ -689,7 +1007,7 @@ private void dataSplitToManifestEntries(
deletionFile.cardinality() != null,
"cardinality in DeletionFile is null, stop generating dv for iceberg. "
+ "dataFile path is {}, deletionFile is {}",
- rawFile.path(),
+ filePath,
deletionFile);
// We can not get the file size of the complete DV index file from the DeletionFile,
@@ -702,7 +1020,7 @@ private void dataSplitToManifestEntries(
dataSplit.partition(),
deletionFile.cardinality(),
-1,
- rawFile.path(),
+ filePath,
deletionFile.offset(),
deletionFile.length());
@@ -937,8 +1255,45 @@ private void createMetadataWithBase(
int lastColumnIdFloor,
long nextRowIdFloor)
throws IOException {
+ createMetadataWithBase(
+ fileChangesCollector,
+ indexFiles,
+ snapshot,
+ baseMetadataPath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ false);
+ }
+
+ /**
+ * @param intermediate whether this metadata is an intermediate step of a {@link
+ * #rebuildFullHistory} replay; intermediate steps skip the version hint and the external
+ * catalog commit, which only the final step publishes.
+ */
+ private void createMetadataWithBase(
+ FileChangesCollector fileChangesCollector,
+ List indexFiles,
+ Snapshot snapshot,
+ Path baseMetadataPath,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ boolean intermediate)
+ throws IOException {
long snapshotId = snapshot.id();
- IcebergMetadata baseMetadata = IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath);
+ IcebergMetadata baseMetadata;
+ try {
+ baseMetadata = IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath);
+ } catch (Exception e) {
+ // an unreadable base is an unusable base: recreate instead of failing the commit,
+ // so a corrupted metadata file self-heals like a structurally invalid one
+ LOG.warn(
+ "Unreadable base Iceberg metadata {}, recreating metadata.",
+ baseMetadataPath,
+ e);
+ recreateFromUnusableBase(
+ snapshotId, null, lastColumnIdFloor, nextRowIdFloor, intermediate);
+ return;
+ }
// row ids handed out by the base or by abandoned metadata must never be reused
long rowIdFloor =
Math.max(
@@ -955,32 +1310,35 @@ private void createMetadataWithBase(
return;
}
// keep the stale base's identity so external catalogs do not recreate the table
- createMetadataWithoutBase(
+ recreateFromUnusableBase(
snapshotId,
baseMetadata.tableUuid(),
Math.max(lastColumnIdFloor, baseMetadata.lastColumnId()),
- rowIdFloor);
+ rowIdFloor,
+ intermediate);
return;
}
if (!isSameFormatVersion(baseMetadata.formatVersion())) {
// we need to recreate iceberg metadata if format version changed
- createMetadataWithoutBase(
+ recreateFromUnusableBase(
snapshot.id(),
null,
Math.max(lastColumnIdFloor, baseMetadata.lastColumnId()),
- rowIdFloor);
+ rowIdFloor,
+ intermediate);
return;
}
if (formatVersion == IcebergMetadata.FORMAT_VERSION_V3
&& baseMetadata.nextRowId() == null) {
// v3 base metadata written before Paimon emitted row lineage; recreate to self-heal
- createMetadataWithoutBase(
+ recreateFromUnusableBase(
snapshot.id(),
baseMetadata.tableUuid(),
Math.max(lastColumnIdFloor, baseMetadata.lastColumnId()),
- rowIdFloor);
+ rowIdFloor,
+ intermediate);
return;
}
@@ -1001,11 +1359,12 @@ private void createMetadataWithBase(
: schemaCache.get(known.schemaId());
if (!known.equals(current)) {
// a re-evolution reused this id with different fields; rebuild from scratch
- createMetadataWithoutBase(
+ recreateFromUnusableBase(
snapshot.id(),
baseMetadata.tableUuid(),
Math.max(lastColumnIdFloor, baseMetadata.lastColumnId()),
- rowIdFloor);
+ rowIdFloor,
+ intermediate);
return;
}
}
@@ -1020,11 +1379,12 @@ private void createMetadataWithBase(
&& baseCurrent.schemaId()
== (int) snapshotManager.snapshot(snapshotId - 1).schemaId();
if (!pointerRollbackOnly) {
- createMetadataWithoutBase(
+ recreateFromUnusableBase(
snapshot.id(),
baseMetadata.tableUuid(),
Math.max(lastColumnIdFloor, baseMetadata.lastColumnId()),
- rowIdFloor);
+ rowIdFloor,
+ intermediate);
return;
}
}
@@ -1255,9 +1615,10 @@ private void createMetadataWithBase(
throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath);
}
// a delayed callback may still write its metadata (a newer commit extends it), but
- // only the current head may move the hint and the external catalog
+ // only the current head may move the hint and the external catalog; an intermediate
+ // replay step publishes nothing and cleans nothing, the final step covers both
Long latestAtPublish = table.snapshotManager().latestSnapshotId();
- if (latestAtPublish != null && latestAtPublish == snapshotId) {
+ if (!intermediate && latestAtPublish != null && latestAtPublish == snapshotId) {
table.fileIO()
.overwriteFileUtf8(
new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME),
@@ -1274,6 +1635,28 @@ private void createMetadataWithBase(
}
}
+ /**
+ * Recreate metadata when the base metadata of a commit turned out to be unusable. At the head
+ * of the history this honors {@link IcebergOptions#SYNC_FULL_HISTORY}; in the middle of a
+ * {@link #rebuildFullHistory(long)} replay (where an unusable base should be impossible, since
+ * the replay itself validates or writes every base) it falls back to single-snapshot metadata
+ * instead of recursing into another replay.
+ */
+ private void recreateFromUnusableBase(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ boolean intermediate)
+ throws IOException {
+ if (intermediate) {
+ createMetadataWithoutBase(
+ snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor, true);
+ } else {
+ recreateMetadata(snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor);
+ }
+ }
+
private interface FileChangesCollector {
boolean collect(
Map> removedFiles,
@@ -1560,16 +1943,18 @@ private List compactMetadataIfNeeded(
// -------------------------------------------------------------------------------------
private boolean shouldExpire(IcebergSnapshot snapshot, long currentSnapshotId) {
+ return shouldExpire(snapshot.snapshotId(), snapshot.timestampMs(), currentSnapshotId);
+ }
+
+ private boolean shouldExpire(long snapshotId, long timestampMs, long currentSnapshotId) {
Options options = new Options(table.options());
- if (snapshot.snapshotId()
- > currentSnapshotId - options.get(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN)) {
+ if (snapshotId > currentSnapshotId - options.get(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN)) {
return false;
}
- if (snapshot.snapshotId()
- <= currentSnapshotId - options.get(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX)) {
+ if (snapshotId <= currentSnapshotId - options.get(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX)) {
return true;
}
- return snapshot.timestampMs()
+ return timestampMs
< System.currentTimeMillis()
- options.get(CoreOptions.SNAPSHOT_TIME_RETAINED).toMillis();
}
@@ -1598,7 +1983,16 @@ private void expireAllBefore(long snapshotId) throws IOException {
while (it.hasNext()) {
Path path = it.next();
- IcebergMetadata metadata = IcebergMetadata.fromPath(table.fileIO(), path);
+ IcebergMetadata metadata;
+ try {
+ metadata = IcebergMetadata.fromPath(table.fileIO(), path);
+ } catch (Exception e) {
+ // an unreadable file must not fail expiration (rebuilds from corrupted
+ // metadata run through here); its manifests are left to orphan cleanup
+ LOG.warn("Deleting unreadable Iceberg metadata {} without expiring it.", path, e);
+ table.fileIO().deleteQuietly(path);
+ continue;
+ }
for (IcebergSnapshot snapshot : metadata.snapshots()) {
Path listPath = new Path(snapshot.manifestList());
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
index 819865066d12..b6c37515a657 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
@@ -95,6 +95,37 @@ public class IcebergOptions {
"The number of old metadata files to keep after each table commit. "
+ "For rest-catalog, it will keep 1 old metadata at least.");
+ public static final ConfigOption SYNC_FULL_HISTORY =
+ key("metadata.iceberg.sync-full-history")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When Iceberg metadata has to be created from scratch (for example, "
+ + "Iceberg compatibility is enabled on a table that already has "
+ + "snapshots, or the previous Iceberg metadata is unusable), "
+ + "rebuild it from all Paimon snapshots that are still retained "
+ + "instead of only the latest one, so Iceberg readers keep time "
+ + "travel and tags. The rebuild cost is proportional to the "
+ + "number of retained snapshots. Readers that resolve Iceberg "
+ + "metadata files (table-location, hadoop-catalog, "
+ + "hive-catalog) see the full replayed history; a rest-catalog "
+ + "only receives the final state.");
+ public static final ConfigOption REST_AUTO_RECREATE =
+ key("metadata.iceberg.rest-auto-recreate")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "When the Iceberg REST catalog state does not match the expected base "
+ + "metadata, automatically drop and recreate the catalog table "
+ + "(legacy recovery). This requires table-delete permission in "
+ + "the catalog (e.g. glue:DeleteTable) and replaces the catalog "
+ + "table's identity, breaking consumers that track it. When set "
+ + "to false, the committer instead reconciles the catalog table "
+ + "in place by replaying the snapshots it is missing from the "
+ + "locally generated metadata, and fails with a precise error "
+ + "when non-destructive reconciliation is impossible; it never "
+ + "drops the catalog table.");
+
public static final ConfigOption URI =
key("metadata.iceberg.uri")
.stringType()
@@ -216,6 +247,10 @@ public int previousVersionsMax() {
return options.get(METADATA_PREVIOUS_VERSIONS_MAX);
}
+ public boolean restAutoRecreate() {
+ return options.get(REST_AUTO_RECREATE);
+ }
+
/** Where to store Iceberg metadata. */
public enum StorageType implements DescribedEnum {
DISABLED("disabled", "Disable Iceberg compatibility support.", false),
diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergBootstrapNonRawSplitsTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergBootstrapNonRawSplitsTest.java
new file mode 100644
index 000000000000..8cfb796b67c4
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergBootstrapNonRawSplitsTest.java
@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
+import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * When Iceberg metadata is created from scratch for a primary key table, buckets with level-0 files
+ * or overlapping key ranges produce {@link org.apache.paimon.table.source.DataSplit}s that are not
+ * raw-convertible. Such splits must not be dropped wholesale: the files in them that the
+ * incremental commit path would have published (via {@code shouldAddFileToIceberg}) must still be
+ * exported, otherwise their rows silently vanish from Iceberg until some future compaction happens
+ * to rewrite the files.
+ */
+public class IcebergBootstrapNonRawSplitsTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileStoreTable table;
+ private TableWriteImpl write;
+ private TableCommitImpl commit;
+ private String commitUser;
+
+ @Test
+ public void testCreateFromScratchExportsCompactedFilesFromNonRawSplits() throws Exception {
+ createPrimaryKeyTableWithoutIceberg();
+ // snapshot 1: level-0 file {1, 2, 3}
+ writeCommit(1, false, GenericRow.of(1, 10), GenericRow.of(2, 20), GenericRow.of(3, 30));
+ // snapshot 2: full compaction, everything at max level
+ fullCompact(2);
+ // snapshot 3: level-0 file {1, 4} overlapping the max level file
+ writeCommit(3, false, GenericRow.of(1, 100), GenericRow.of(4, 40));
+
+ enableIceberg(false);
+ // snapshot 4: another level-0 file; triggers creating Iceberg metadata from scratch
+ writeCommit(4, false, GenericRow.of(5, 50));
+
+ // The bucket's files (max level + two level-0) form a split that is not raw-convertible.
+ // The max level file must still be exported: Iceberg sees the data as of the last full
+ // compaction, exactly like an incremental sync running since the table was created.
+ IcebergMetadata metadata = readMetadata(4);
+ assertThat(metadata.snapshots()).hasSize(1);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+
+ // a full compaction exports the remaining rows through the incremental path
+ fullCompact(5);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 100)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)");
+ }
+
+ @Test
+ public void testFullHistoryReplayWithNonRawSplits() throws Exception {
+ createPrimaryKeyTableWithoutIceberg();
+ writeCommit(1, false, GenericRow.of(1, 10), GenericRow.of(2, 20), GenericRow.of(3, 30));
+ fullCompact(2);
+ writeCommit(3, false, GenericRow.of(1, 100), GenericRow.of(4, 40));
+
+ enableIceberg(true);
+ writeCommit(4, false, GenericRow.of(5, 50));
+
+ // the replay mirrors live commits: every retained snapshot becomes an Iceberg snapshot
+ IcebergMetadata metadata = readMetadata(4);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L, 4L);
+
+ // snapshot 1 is a single level-0 file with no other files to merge with, so it is
+ // raw-convertible and fully visible; snapshots 3 and 4 add level-0 files, which stay
+ // invisible until compaction, exactly like live incremental commits
+ assertThat(
+ getIcebergResult(
+ icebergTable ->
+ IcebergGenerics.read(icebergTable).useSnapshot(1).build(),
+ Record::toString))
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+
+ fullCompact(5);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 100)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)");
+ }
+
+ // ------------------------------------------------------------------------
+ // Utils
+ // ------------------------------------------------------------------------
+
+ private void createPrimaryKeyTableWithoutIceberg() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.FILE_FORMAT, "avro");
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Arrays.asList("k"),
+ options.toMap(),
+ "");
+
+ try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) {
+ paimonCatalog.createDatabase("mydb", false);
+ Identifier paimonIdentifier = Identifier.create("mydb", "t");
+ paimonCatalog.createTable(paimonIdentifier, schema, false);
+ table = (FileStoreTable) paimonCatalog.getTable(paimonIdentifier);
+ }
+
+ commitUser = UUID.randomUUID().toString();
+ write = table.newWrite(commitUser);
+ commit = table.newCommit(commitUser);
+ }
+
+ private void enableIceberg(boolean syncFullHistory) throws Exception {
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), String.valueOf(syncFullHistory));
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+ }
+
+ private void writeCommit(long identifier, boolean waitCompaction, GenericRow... rows)
+ throws Exception {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(identifier, write.prepareCommit(waitCompaction, identifier));
+ }
+
+ private void fullCompact(long identifier) throws Exception {
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ writeCommit(identifier, true);
+ }
+
+ private IcebergMetadata readMetadata(long snapshotId) {
+ return IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(table.location(), "metadata/v" + snapshotId + ".metadata.json"));
+ }
+
+ private List getIcebergResult() throws Exception {
+ return getIcebergResult(
+ icebergTable -> IcebergGenerics.read(icebergTable).build(), Record::toString);
+ }
+
+ private List getIcebergResult(
+ Function> query,
+ Function icebergRecordToString)
+ throws Exception {
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString());
+ TableIdentifier icebergIdentifier = TableIdentifier.of("mydb.db", "t");
+ org.apache.iceberg.Table icebergTable = icebergCatalog.loadTable(icebergIdentifier);
+ CloseableIterable result = query.apply(icebergTable);
+ List actual = new ArrayList<>();
+ for (Record record : result) {
+ actual.add(icebergRecordToString.apply(record));
+ }
+ result.close();
+ return actual;
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergSyncFullHistoryTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergSyncFullHistoryTest.java
new file mode 100644
index 000000000000..1617908a79a5
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergSyncFullHistoryTest.java
@@ -0,0 +1,463 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
+import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link IcebergOptions#SYNC_FULL_HISTORY}: when Iceberg metadata is created from
+ * scratch, the whole retained Paimon history is replayed instead of only the latest snapshot. See
+ * apache/paimon#6107.
+ */
+public class IcebergSyncFullHistoryTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private static final String VERSION_HINT_FILENAME = "version-hint.text";
+
+ private FileStoreTable table;
+ private TableWriteImpl write;
+ private TableCommitImpl commit;
+ private String commitUser;
+
+ @Test
+ public void testDefaultRebuildOnlyExposesLatestSnapshot() throws Exception {
+ createAppendTableWithoutIceberg();
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+ writeCommit(3, GenericRow.of(3, 30));
+
+ enableIceberg(false);
+ writeCommit(4, GenericRow.of(4, 40));
+
+ IcebergMetadata metadata = readMetadata(4);
+ assertThat(metadata.snapshots()).hasSize(1);
+ assertThat(metadata.currentSnapshotId()).isEqualTo(4);
+ // even though it exposes only one Iceberg snapshot, it contains all live files
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)", "Record(2, 20)", "Record(3, 30)", "Record(4, 40)");
+ }
+
+ @Test
+ public void testSyncFullHistoryReplaysRetainedSnapshots() throws Exception {
+ createAppendTableWithoutIceberg();
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+ writeCommit(3, GenericRow.of(3, 30));
+ table.createTag("tag-2", 2);
+
+ enableIceberg(true);
+ writeCommit(4, GenericRow.of(4, 40));
+
+ IcebergMetadata metadata = readMetadata(4);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L, 4L);
+ assertThat(metadata.currentSnapshotId()).isEqualTo(4);
+
+ // replayed snapshots keep the original Paimon commit timestamps
+ for (IcebergSnapshot icebergSnapshot : metadata.snapshots()) {
+ Snapshot paimonSnapshot =
+ table.snapshotManager().snapshot(icebergSnapshot.snapshotId());
+ assertThat(icebergSnapshot.timestampMs()).isEqualTo(paimonSnapshot.timeMillis());
+ }
+
+ // a pre-existing tag becomes an Iceberg ref because its snapshot now exists
+ assertThat(metadata.refs()).containsOnlyKeys("tag-2");
+ assertThat(metadata.refs().get("tag-2").snapshotId()).isEqualTo(2);
+
+ // only the final replay step publishes the version hint
+ assertThat(
+ table.fileIO()
+ .readFileUtf8(
+ new Path(
+ table.location(),
+ "metadata/" + VERSION_HINT_FILENAME)))
+ .isEqualTo("4");
+
+ // an Iceberg client can read the current state, time travel, and resolve the tag
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)", "Record(2, 20)", "Record(3, 30)", "Record(4, 40)");
+ assertThat(
+ getIcebergResult(
+ icebergTable ->
+ IcebergGenerics.read(icebergTable).useSnapshot(2).build(),
+ Record::toString))
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)");
+ assertThat(
+ getIcebergResult(
+ icebergTable ->
+ IcebergGenerics.read(icebergTable)
+ .useSnapshot(
+ icebergTable
+ .refs()
+ .get("tag-2")
+ .snapshotId())
+ .build(),
+ Record::toString))
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)");
+ }
+
+ @Test
+ public void testInterruptedReplayResumesFromNewestMetadata() throws Exception {
+ createAppendTableWithoutIceberg();
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+ writeCommit(3, GenericRow.of(3, 30));
+
+ enableIceberg(true);
+ writeCommit(4, GenericRow.of(4, 40));
+ assertThat(readMetadata(4).snapshots()).hasSize(4);
+
+ // Simulate an interrupted replay / failed Iceberg commit: the newest metadata is missing,
+ // but earlier replay steps survived.
+ IcebergPathFactory pathFactory =
+ new IcebergPathFactory(new Path(table.location(), "metadata"));
+ String metadata3Before = table.fileIO().readFileUtf8(pathFactory.toMetadataPath(3));
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(4));
+
+ writeCommit(5, GenericRow.of(5, 50));
+
+ IcebergMetadata metadata = readMetadata(5);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L, 4L, 5L);
+ // metadata of already-replayed snapshots is reused, not rebuilt
+ assertThat(table.fileIO().readFileUtf8(pathFactory.toMetadataPath(3)))
+ .isEqualTo(metadata3Before);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)");
+ }
+
+ @Test
+ public void testResumeRejectsBaseWithoutRetainedPrefix() throws Exception {
+ createAppendTableWithoutIceberg();
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+ writeCommit(3, GenericRow.of(3, 30));
+
+ // Iceberg was first enabled WITHOUT full history sync: the metadata only contains the
+ // latest snapshot.
+ enableIceberg(false);
+ writeCommit(4, GenericRow.of(4, 40));
+ assertThat(readMetadata(4).snapshots()).hasSize(1);
+
+ // Full history sync is enabled later, and the newest metadata is lost (e.g. a failed
+ // Iceberg commit). The single-snapshot metadata of snapshot 4 is NOT a valid replay
+ // prefix: resuming from it would silently drop snapshots 1-3 forever.
+ Map options = new HashMap<>();
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), "true");
+ reopen(options);
+ writeCommit(5, GenericRow.of(5, 50));
+ IcebergPathFactory pathFactory =
+ new IcebergPathFactory(new Path(table.location(), "metadata"));
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(5));
+
+ writeCommit(6, GenericRow.of(6, 60));
+
+ IcebergMetadata metadata = readMetadata(6);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L, 4L, 5L, 6L);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)",
+ "Record(6, 60)");
+ }
+
+ @Test
+ public void testFormatVersionChangeRebuildsHistoryWithRowLineage() throws Exception {
+ // Iceberg (v2) is enabled from the start, with full history sync on.
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), "true");
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ createAppendTable(options);
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20), GenericRow.of(3, 30));
+ assertThat(readMetadata(2).formatVersion()).isEqualTo(IcebergMetadata.FORMAT_VERSION_V2);
+
+ // Switching to format version 3 makes the v2 base unusable, which triggers a full-history
+ // rebuild; the stale v2 metadata files must be cleaned up so the replay can write in their
+ // place.
+ Map upgrade = new HashMap<>();
+ upgrade.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ reopen(upgrade);
+ writeCommit(3, GenericRow.of(4, 40));
+
+ IcebergMetadata metadata = readMetadata(3);
+ assertThat(metadata.formatVersion()).isEqualTo(IcebergMetadata.FORMAT_VERSION_V3);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L);
+
+ // v3 row lineage accumulates consistently across the replayed history
+ List firstRowIds = new ArrayList<>();
+ for (IcebergSnapshot icebergSnapshot : metadata.snapshots()) {
+ assertThat(icebergSnapshot.firstRowId()).isNotNull();
+ assertThat(icebergSnapshot.addedRows()).isNotNull();
+ firstRowIds.add(icebergSnapshot.firstRowId());
+ }
+ assertThat(firstRowIds).containsExactly(0L, 1L, 3L);
+ assertThat(metadata.nextRowId()).isEqualTo(4);
+ }
+
+ @Test
+ public void testRebuildFromCorruptedMetadataSucceeds() throws Exception {
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), "true");
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ createAppendTable(options);
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+
+ // corrupt the newest metadata: the next commit finds no usable base and must rebuild
+ // from scratch, tolerating the unreadable file in every step of the rebuild
+ Path corrupted = new Path(table.location(), "metadata/v2.metadata.json");
+ table.fileIO().deleteQuietly(corrupted);
+ table.fileIO().overwriteFileUtf8(corrupted, "{ not json");
+
+ writeCommit(3, GenericRow.of(3, 30));
+
+ IcebergMetadata metadata = readMetadata(3);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+ }
+
+ @Test
+ public void testRebuildCleansOldBuildOnlyAfterPublication() throws Exception {
+ createAppendTableWithoutIceberg();
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+
+ // Iceberg enabled without full history: single-snapshot metadata, the "old build"
+ enableIceberg(false);
+ writeCommit(3, GenericRow.of(3, 30));
+ assertThat(readMetadata(3).snapshots()).hasSize(1);
+ List oldManifestLists = new ArrayList<>();
+ for (IcebergSnapshot snapshot : readMetadata(3).snapshots()) {
+ oldManifestLists.add(
+ new Path(
+ table.location(),
+ "metadata/" + new Path(snapshot.manifestList()).getName()));
+ }
+ assertThat(oldManifestLists).isNotEmpty();
+
+ // full history enabled later and the newest metadata lost: the next commit rebuilds
+ // from scratch (the single-snapshot candidate is not a valid replay prefix)
+ Map options = new HashMap<>();
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), "true");
+ reopen(options);
+ writeCommit(4, GenericRow.of(4, 40));
+ IcebergPathFactory pathFactory =
+ new IcebergPathFactory(new Path(table.location(), "metadata"));
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(4));
+ // the old build stays fully readable up to this point
+ for (Path listPath : oldManifestLists) {
+ assertThat(table.fileIO().exists(listPath)).isTrue();
+ }
+
+ writeCommit(5, GenericRow.of(5, 50));
+
+ // the rebuild published a full replacement chain ...
+ IcebergMetadata metadata = readMetadata(5);
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .containsExactly(1L, 2L, 3L, 4L, 5L);
+ // ... whose snapshots reference only freshly written manifest lists ...
+ List oldNames =
+ oldManifestLists.stream().map(Path::getName).collect(Collectors.toList());
+ for (IcebergSnapshot snapshot : metadata.snapshots()) {
+ assertThat(new Path(snapshot.manifestList()).getName()).isNotIn(oldNames);
+ }
+ // ... and only then were the old build's files removed
+ for (Path listPath : oldManifestLists) {
+ assertThat(table.fileIO().exists(listPath)).isFalse();
+ }
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)");
+ }
+
+ // ------------------------------------------------------------------------
+ // Utils
+ // ------------------------------------------------------------------------
+
+ private void createAppendTableWithoutIceberg() throws Exception {
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.DISABLED.toString());
+ createAppendTable(options);
+ }
+
+ private void createAppendTable(Map customOptions) throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+
+ Options options = new Options(customOptions);
+ options.set(CoreOptions.BUCKET, -1);
+ options.set(CoreOptions.FILE_FORMAT, "avro");
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options.toMap(),
+ "");
+
+ try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) {
+ paimonCatalog.createDatabase("mydb", false);
+ Identifier paimonIdentifier = Identifier.create("mydb", "t");
+ paimonCatalog.createTable(paimonIdentifier, schema, false);
+ table = (FileStoreTable) paimonCatalog.getTable(paimonIdentifier);
+ }
+
+ commitUser = UUID.randomUUID().toString();
+ write = table.newWrite(commitUser);
+ commit = table.newCommit(commitUser);
+ }
+
+ private void enableIceberg(boolean syncFullHistory) throws Exception {
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), String.valueOf(syncFullHistory));
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ reopen(options);
+ }
+
+ private void reopen(Map options) throws Exception {
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+ }
+
+ private void writeCommit(long identifier, GenericRow... rows) throws Exception {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(identifier, write.prepareCommit(false, identifier));
+ }
+
+ private IcebergMetadata readMetadata(long snapshotId) {
+ return IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(table.location(), "metadata/v" + snapshotId + ".metadata.json"));
+ }
+
+ private List getIcebergResult() throws Exception {
+ return getIcebergResult(
+ icebergTable -> IcebergGenerics.read(icebergTable).build(), Record::toString);
+ }
+
+ private List getIcebergResult(
+ Function> query,
+ Function icebergRecordToString)
+ throws Exception {
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString());
+ TableIdentifier icebergIdentifier = TableIdentifier.of("mydb.db", "t");
+ org.apache.iceberg.Table icebergTable = icebergCatalog.loadTable(icebergIdentifier);
+ CloseableIterable result = query.apply(icebergTable);
+ List actual = new ArrayList<>();
+ for (Record record : result) {
+ actual.add(icebergRecordToString.apply(record));
+ }
+ result.close();
+ return actual;
+ }
+}
diff --git a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestCatalogOutOfSyncException.java b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestCatalogOutOfSyncException.java
new file mode 100644
index 000000000000..45cb0e6910b0
--- /dev/null
+++ b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestCatalogOutOfSyncException.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+/**
+ * Thrown when the state of the external Iceberg REST catalog cannot be reconciled with the locally
+ * generated Iceberg metadata without a destructive operation (dropping or rewinding the catalog
+ * table), which the committer never performs automatically.
+ *
+ * The message always names the catalog table, the snapshot the catalog is at, the snapshot that
+ * was to be published, and the manual remediation.
+ */
+public class IcebergRestCatalogOutOfSyncException extends RuntimeException {
+
+ public IcebergRestCatalogOutOfSyncException(String message) {
+ super(message);
+ }
+}
diff --git a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
index adeec647f14f..4236f7498fa0 100644
--- a/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
+++ b/paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java
@@ -46,6 +46,8 @@
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.rest.Endpoint;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.types.Types;
@@ -58,11 +60,13 @@
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -140,6 +144,8 @@ public void commitMetadata(
IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) {
try {
commitMetadataImpl(newIcebergMetadata, baseIcebergMetadata);
+ } catch (IcebergRestCatalogOutOfSyncException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -215,27 +221,66 @@ private void commitMetadataImpl(
updateBuilder = updatesForCorrectBase(metadata, newMetadata, true);
} else {
boolean withBase = checkBase(metadata, newMetadata, baseIcebergMetadata);
+ if (withBase
+ && !icebergOptions.restAutoRecreate()
+ && !Objects.equals(
+ metadata.currentSnapshot().manifestListLocation(),
+ baseIcebergMetadata.currentSnapshot().manifestList())) {
+ // The catalog head has the expected snapshot id but not the expected
+ // content: the local metadata was regenerated (e.g. a full-history
+ // rebuild) and the catalog still holds entries of the previous build.
+ // The id-based check cannot see this; route through reconciliation so
+ // the stale entries are cleaned up instead of being appended to.
+ LOG.info(
+ "catalog head {} matches the expected base snapshot id but not "
+ + "its manifest list, reconciling.",
+ metadata.currentSnapshot().snapshotId());
+ withBase = false;
+ }
if (withBase) {
LOG.info("create updates with base metadata.");
updateBuilder = updatesForCorrectBase(metadata, newMetadata, false);
} else {
LOG.info(
- "create updates without base metadata. currentSnapshotId for base metadata: {}, for new metadata:{}",
+ "catalog state diverged from the expected base. currentSnapshotId"
+ + " in catalog: {}, in new metadata: {}",
metadata.currentSnapshot().snapshotId(),
newMetadata.currentSnapshot() != null
? newMetadata.currentSnapshot().snapshotId()
: "No snapshot");
- if (requiresRegistration(newIcebergMetadata)) {
+ if (newMetadata.currentSnapshot() != null
+ && sameSnapshot(
+ metadata.currentSnapshot(),
+ newMetadata.currentSnapshot())) {
+ // in every mode: recreating or reconciling the table here would
+ // only reproduce identical content under a new table identity
+ LOG.info(
+ "Iceberg table {} already contains snapshot {}; nothing to"
+ + " publish.",
+ icebergTableIdentifier,
+ newMetadata.currentSnapshot().snapshotId());
+ return;
+ }
+ if (!icebergOptions.restAutoRecreate()) {
+ updateBuilder = updatesForDivergedBase(metadata, newMetadata);
+ if (updateBuilder == null) {
+ // the catalog already holds the intended snapshot
+ return;
+ }
+ } else if (requiresRegistration(newIcebergMetadata)) {
LOG.info(
"the base metadata is incorrect, re-registering the iceberg"
+ " table from local metadata.");
registerAsCurrent(newIcebergMetadata, newMetadata, true);
return;
+ } else {
+ updateBuilder = updatesForIncorrectBase(newMetadata);
}
- updateBuilder = updatesForIncorrectBase(newMetadata);
}
}
}
+ } catch (IcebergRestCatalogOutOfSyncException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(
"Fail to create table or get table: " + icebergTableIdentifier, e);
@@ -247,15 +292,70 @@ private void commitMetadataImpl(
LOG.debug("updates:{}", updatesToString(updatedForCommit.changes()));
}
+ commitToRestCatalog(updatedForCommit);
+ }
+
+ /**
+ * Commit to the REST catalog, resolving ambiguous outcomes instead of failing pessimistically.
+ *
+ *
By the time this runs the Paimon snapshot and the local Iceberg metadata file are already
+ * durable, so an ambiguous catalog failure ({@link CommitStateUnknownException}, e.g. an AWS
+ * Glue timeout or 500 after the change was applied) must not be treated as fatal outright: the
+ * catalog is reloaded to check whether the commit actually landed, and if it verifiably did
+ * not, the same update set is retried once against the unchanged base. Only then is the
+ * ambiguity surfaced; a later commit still recovers by replaying the gap from the local
+ * metadata files through {@link #updatesForDivergedBase}, never by dropping the catalog table.
+ */
+ private void commitToRestCatalog(TableMetadata updatedForCommit) {
+ TableMetadata base = ((BaseTable) icebergTable).operations().current();
try {
- ((BaseTable) icebergTable)
- .operations()
- .commit(((BaseTable) icebergTable).operations().current(), updatedForCommit);
+ ((BaseTable) icebergTable).operations().commit(base, updatedForCommit);
+ return;
+ } catch (CommitStateUnknownException e) {
+ if (commitLanded(updatedForCommit)) {
+ LOG.info(
+ "Ambiguous commit to iceberg table {} verified as applied after reloading "
+ + "the catalog state.",
+ icebergTableIdentifier);
+ return;
+ }
+ LOG.warn(
+ "Commit to iceberg table {} finished in an unknown state and the catalog does "
+ + "not show it as applied; retrying once against the unchanged base.",
+ icebergTableIdentifier,
+ e);
+ try {
+ ((BaseTable) icebergTable).operations().commit(base, updatedForCommit);
+ return;
+ } catch (Exception retryFailure) {
+ e.addSuppressed(retryFailure);
+ }
+ throw e;
} catch (Exception e) {
throw new RuntimeException("Fail to commit metadata to rest catalog.", e);
}
}
+ /** Whether the catalog's current snapshot is exactly the one {@code updatedForCommit} sets. */
+ private boolean commitLanded(TableMetadata updatedForCommit) {
+ try {
+ Table reloaded = getTable();
+ org.apache.iceberg.Snapshot current = reloaded.currentSnapshot();
+ org.apache.iceberg.Snapshot intended = updatedForCommit.currentSnapshot();
+ boolean landed = current != null && intended != null && sameSnapshot(current, intended);
+ if (landed) {
+ icebergTable = reloaded;
+ }
+ return landed;
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to reload iceberg table {} while verifying an ambiguous commit.",
+ icebergTableIdentifier,
+ e);
+ return false;
+ }
+ }
+
private TableMetadata.Builder updatesForCorrectBase(
TableMetadata base, TableMetadata newMetadata, boolean isNewTable) {
TableMetadata.Builder updateBuilder = TableMetadata.buildFrom(base);
@@ -276,7 +376,7 @@ private TableMetadata.Builder updatesForCorrectBase(
updateBuilder.setDefaultPartitionSpec(newMetadata.defaultSpecId());
// add snapshot
- addNewSnapshot(newMetadata.currentSnapshot(), updateBuilder);
+ addNewSnapshot(base, newMetadata.currentSnapshot(), updateBuilder);
} else {
// add new schema if needed
@@ -293,7 +393,7 @@ private TableMetadata.Builder updatesForCorrectBase(
}
// add snapshot
- addNewSnapshot(newMetadata.currentSnapshot(), updateBuilder);
+ addNewSnapshot(base, newMetadata.currentSnapshot(), updateBuilder);
// remove snapshots not in new metadata
Set snapshotIdsToRemove = new HashSet<>();
@@ -312,6 +412,209 @@ private TableMetadata.Builder updatesForCorrectBase(
return updateBuilder;
}
+ /**
+ * Reconcile a catalog whose state is not exactly one snapshot behind the new metadata, without
+ * ever dropping the catalog table ({@link IcebergOptions#REST_AUTO_RECREATE} set to false). The
+ * local metadata file records that metadata was generated, not that it was published to
+ * the catalog, so after an ambiguous or failed publication the catalog may be an arbitrary
+ * number of snapshots behind; publication completion is derived here by loading the catalog
+ * state rather than tracked separately.
+ *
+ * Every Iceberg snapshot is self-contained through its manifest list, so a behind catalog is
+ * brought up to date by adding, in order, each retained snapshot it is missing and removing the
+ * entries the new metadata no longer retains, which is exactly the state that consecutive
+ * successful commits would have produced. Snapshots whose sequence number the catalog has
+ * already consumed (leftovers of a locally regenerated history) cannot be re-added, since
+ * Iceberg requires strictly increasing sequence numbers; they are removed without replacement,
+ * costing only catalog-side time travel to them.
+ *
+ *
The catalog table is never dropped or recreated. States that cannot be reconciled
+ * non-destructively (the catalog being ahead of the new metadata, or holding a different
+ * snapshot under the id being published) fail with {@link IcebergRestCatalogOutOfSyncException}
+ * instead.
+ *
+ * @return the updates to commit, or null if the catalog already holds the intended snapshot (a
+ * previously ambiguous commit actually landed) and there is nothing to publish
+ */
+ @Nullable
+ private TableMetadata.Builder updatesForDivergedBase(
+ TableMetadata currentMetadata, TableMetadata newMetadata) {
+ long catalogSnapshotId = currentMetadata.currentSnapshot().snapshotId();
+ Snapshot newCurrentSnapshot = newMetadata.currentSnapshot();
+ if (newCurrentSnapshot == null) {
+ throw new IcebergRestCatalogOutOfSyncException(
+ String.format(
+ "Cannot reconcile iceberg table %s: the newly generated metadata "
+ + "contains no snapshot while the catalog is at snapshot %s. "
+ + "Remediation: investigate why Paimon generated empty Iceberg "
+ + "metadata; the catalog table was left untouched.",
+ icebergTableIdentifier, catalogSnapshotId));
+ }
+ long newSnapshotId = newCurrentSnapshot.snapshotId();
+
+ if (catalogSnapshotId == newSnapshotId) {
+ if (sameSnapshot(currentMetadata.currentSnapshot(), newCurrentSnapshot)) {
+ LOG.info(
+ "Iceberg table {} already contains snapshot {}; a previously ambiguous "
+ + "commit landed, nothing to publish.",
+ icebergTableIdentifier,
+ newSnapshotId);
+ return null;
+ }
+ throw new IcebergRestCatalogOutOfSyncException(
+ String.format(
+ "Cannot reconcile iceberg table %s: the catalog is at snapshot %s "
+ + "with manifest list %s, but the metadata to publish carries "
+ + "a different snapshot under the same id (manifest list %s). "
+ + "The catalog head cannot be replaced in place because "
+ + "Iceberg sequence numbers are single-use. Remediation: the "
+ + "next Paimon commit publishes a higher snapshot and "
+ + "re-aligns the head; to discard the catalog state instead, "
+ + "drop or rename the catalog table manually. The catalog "
+ + "table was left untouched.",
+ icebergTableIdentifier,
+ catalogSnapshotId,
+ currentMetadata.currentSnapshot().manifestListLocation(),
+ newCurrentSnapshot.manifestListLocation()));
+ }
+
+ if (catalogSnapshotId > newSnapshotId) {
+ throw new IcebergRestCatalogOutOfSyncException(
+ String.format(
+ "Cannot reconcile iceberg table %s: the catalog is at snapshot %s, "
+ + "ahead of snapshot %s being published. The catalog table "
+ + "either belongs to a different (foreign or recreated) table "
+ + "history, or was advanced by another writer; rolling it "
+ + "back would be destructive. Remediation: verify the catalog "
+ + "table really mirrors this Paimon table, and drop or rename "
+ + "it manually if its history must be discarded. The catalog "
+ + "table was left untouched.",
+ icebergTableIdentifier, catalogSnapshotId, newSnapshotId));
+ }
+
+ // The catalog is behind: replay, in order, every retained snapshot it is missing. A
+ // snapshot the catalog already holds with identical content (e.g. after an external
+ // rollback) is re-activated by a plain ref move; a genuinely new snapshot must be added,
+ // which Iceberg only allows with a sequence number above the catalog's high-water mark.
+ // Leftovers of a regenerated history below that mark cannot be re-added and are skipped
+ // (their stale entries are removed below).
+ long lastSequenceNumber = currentMetadata.lastSequenceNumber();
+ Map catalogSnapshotsById =
+ currentMetadata.snapshots().stream()
+ .collect(Collectors.toMap(Snapshot::snapshotId, s -> s));
+ List toReplay =
+ newMetadata.snapshots().stream()
+ .filter(s -> s.snapshotId() > catalogSnapshotId)
+ .filter(
+ s -> {
+ Snapshot inCatalog = catalogSnapshotsById.get(s.snapshotId());
+ return inCatalog != null
+ ? sameSnapshot(s, inCatalog)
+ : s.sequenceNumber() > lastSequenceNumber;
+ })
+ .sorted(Comparator.comparingLong(Snapshot::snapshotId))
+ .collect(Collectors.toList());
+ if (toReplay.isEmpty() || toReplay.get(toReplay.size() - 1).snapshotId() != newSnapshotId) {
+ throw new IcebergRestCatalogOutOfSyncException(
+ String.format(
+ "Cannot reconcile iceberg table %s: the catalog is at snapshot %s "
+ + "with last sequence number %s, and snapshot %s to be "
+ + "published cannot be added because its sequence number was "
+ + "already consumed by the catalog. Remediation: the next "
+ + "Paimon commit publishes a higher snapshot and re-aligns "
+ + "the catalog; to discard the catalog state instead, drop or "
+ + "rename the catalog table manually. The catalog table was "
+ + "left untouched.",
+ icebergTableIdentifier,
+ catalogSnapshotId,
+ lastSequenceNumber,
+ newSnapshotId));
+ }
+
+ Map newSnapshotsById =
+ newMetadata.snapshots().stream()
+ .collect(Collectors.toMap(Snapshot::snapshotId, s -> s));
+ // Catalog entries the new metadata does not retain with identical content: snapshots
+ // expired locally, plus leftovers of a regenerated history (same id, different manifest
+ // list) whose manifest files may no longer exist.
+ Set staleSnapshotIds = new HashSet<>();
+ long replaced = 0;
+ for (Snapshot snapshot : currentMetadata.snapshots()) {
+ Snapshot inNew = newSnapshotsById.get(snapshot.snapshotId());
+ if (inNew == null || !sameSnapshot(snapshot, inNew)) {
+ staleSnapshotIds.add(snapshot.snapshotId());
+ if (inNew != null) {
+ replaced++;
+ }
+ }
+ }
+ if (replaced > 0) {
+ LOG.warn(
+ "Reconciling iceberg table {}: {} snapshot(s) in the catalog were generated "
+ + "by an earlier metadata build and their regenerated replacements "
+ + "cannot be re-added under already-consumed sequence numbers; they "
+ + "are removed from the catalog without replacement. Time travel to "
+ + "them through the catalog is unavailable; the full history remains "
+ + "in the file-based Iceberg metadata.",
+ icebergTableIdentifier,
+ replaced);
+ }
+ LOG.info(
+ "Reconciling iceberg table {} from snapshot {} to snapshot {}: replaying {} "
+ + "snapshot(s), removing {} stale snapshot(s).",
+ icebergTableIdentifier,
+ catalogSnapshotId,
+ newSnapshotId,
+ toReplay.size(),
+ staleSnapshotIds.size());
+
+ try {
+ TableMetadata.Builder updateBuilder = TableMetadata.buildFrom(currentMetadata);
+ if (newMetadata.formatVersion() > currentMetadata.formatVersion()) {
+ updateBuilder.upgradeFormatVersion(newMetadata.formatVersion());
+ }
+
+ int schemaId = icebergTable.schema().schemaId();
+ if (newMetadata.currentSchemaId() > schemaId) {
+ addAndSetCurrentSchema(
+ newMetadata.schemas().stream()
+ .filter(schema -> schema.schemaId() > schemaId)
+ .collect(Collectors.toList()),
+ newMetadata.currentSchemaId(),
+ updateBuilder);
+ }
+
+ for (Snapshot snapshot : toReplay) {
+ addNewSnapshot(currentMetadata, snapshot, updateBuilder);
+ }
+ removeSnapshots(staleSnapshotIds, updateBuilder);
+ updateProperties(updateBuilder);
+ // build eagerly so Iceberg's own validation failures (e.g. a version 3 row-id space
+ // that restarted behind the catalog's next-row-id watermark) surface as a precise
+ // out-of-sync error instead of a generic commit failure
+ updateBuilder.build();
+ return updateBuilder;
+ } catch (ValidationException e) {
+ throw new IcebergRestCatalogOutOfSyncException(
+ String.format(
+ "Cannot reconcile iceberg table %s from snapshot %s to snapshot %s: "
+ + "the replayed snapshots are not addable on top of the "
+ + "catalog state (%s). Remediation: drop or rename the "
+ + "catalog table manually to republish from scratch. The "
+ + "catalog table was left untouched.",
+ icebergTableIdentifier,
+ catalogSnapshotId,
+ newSnapshotId,
+ e.getMessage()));
+ }
+ }
+
+ /** Whether two snapshot entries describe the same physical snapshot. */
+ private static boolean sameSnapshot(Snapshot a, Snapshot b) {
+ return a.snapshotId() == b.snapshotId()
+ && Objects.equals(a.manifestListLocation(), b.manifestListLocation());
+ }
+
private TableMetadata.Builder updatesForIncorrectBase(TableMetadata newMetadata) {
LOG.info("the base metadata is incorrect, we'll recreate the iceberg table.");
icebergTable = recreateTable(newMetadata);
@@ -626,9 +929,16 @@ private void dropTable() {
// metadata updates
// -------------------------------------------------------------------------------------
- // add a new snapshot and point it as current snapshot
- private void addNewSnapshot(Snapshot newSnapshot, TableMetadata.Builder update) {
- update.setBranchSnapshot(newSnapshot, SnapshotRef.MAIN_BRANCH);
+ // point the main branch at the given snapshot, adding it first unless the base metadata
+ // already contains it (Iceberg's Snapshot-taking overload always adds and rejects an existing
+ // id, e.g. when re-activating a snapshot after an external rollback)
+ private void addNewSnapshot(
+ TableMetadata base, Snapshot newSnapshot, TableMetadata.Builder update) {
+ if (base.snapshot(newSnapshot.snapshotId()) != null) {
+ update.setBranchSnapshot(newSnapshot.snapshotId(), SnapshotRef.MAIN_BRANCH);
+ } else {
+ update.setBranchSnapshot(newSnapshot, SnapshotRef.MAIN_BRANCH);
+ }
}
// remove snapshots recorded in table metadata
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFullHistoryCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFullHistoryCompatibilityTest.java
new file mode 100644
index 000000000000..325ec204ee22
--- /dev/null
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFullHistoryCompatibilityTest.java
@@ -0,0 +1,333 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.core;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.FileSystemCatalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.iceberg.IcebergPathFactory;
+import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta;
+import org.apache.paimon.iceberg.manifest.IcebergManifestList;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
+import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end test for {@link IcebergOptions#SYNC_FULL_HISTORY} on an Iceberg v3 primary-key table
+ * with deletion vectors: enabling Iceberg compatibility on a table that already has snapshots must
+ * rebuild the full retained history with a consistent row-id space, readable (including time
+ * travel) by a real Apache Iceberg client. See apache/paimon#6107.
+ */
+public class IcebergFullHistoryCompatibilityTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void testEnableOnExistingV3DvTableRebuildsHistory() throws Exception {
+ FileStoreTable table = createPaimonTableWithoutIceberg();
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(row(RowKind.INSERT, 1, 1, "a"));
+ write.write(row(RowKind.INSERT, 1, 2, "b"));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ write.compact(partition(1), 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ write.write(row(RowKind.INSERT, 1, 3, "c"));
+ write.write(row(RowKind.DELETE, 1, 2, "b"));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ // no Iceberg metadata was produced so far
+ assertThat(table.fileIO().exists(new Path(table.location(), "metadata/v3.metadata.json")))
+ .isFalse();
+
+ // enable Iceberg v3 with full history sync; the next commit rebuilds everything
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ options.put(IcebergOptions.SYNC_FULL_HISTORY.key(), "true");
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ table = table.copy(options);
+ write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ commit = table.newCommit(commitUser);
+
+ // this compaction merges the delete and produces a deletion vector
+ write.compact(partition(1), 0, false);
+ commit.commit(4, write.prepareCommit(true, 4));
+ write.close();
+ commit.close();
+
+ long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(
+ table.location(),
+ "metadata/v" + latestSnapshotId + ".metadata.json"));
+
+ // the whole retained history is exposed
+ assertThat(metadata.formatVersion()).isEqualTo(IcebergMetadata.FORMAT_VERSION_V3);
+ List snapshotIds =
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList());
+ assertThat(snapshotIds)
+ .isEqualTo(
+ java.util.stream.LongStream.rangeClosed(1, latestSnapshotId)
+ .boxed()
+ .collect(Collectors.toList()));
+
+ // the v3 row-id space accumulates monotonically across the replayed history
+ Long previousFirstRowId = null;
+ for (IcebergSnapshot icebergSnapshot : metadata.snapshots()) {
+ assertThat(icebergSnapshot.firstRowId()).isNotNull();
+ assertThat(icebergSnapshot.addedRows()).isNotNull();
+ if (previousFirstRowId != null) {
+ assertThat(icebergSnapshot.firstRowId()).isGreaterThanOrEqualTo(previousFirstRowId);
+ }
+ previousFirstRowId = icebergSnapshot.firstRowId();
+ }
+ assertThat(metadata.nextRowId()).isNotNull();
+
+ // every replayed snapshot's data manifests carry a non-null first_row_id (required by
+ // strict v3 readers like Snowflake)
+ IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table, paths);
+ for (IcebergSnapshot icebergSnapshot : metadata.snapshots()) {
+ List metas =
+ manifestList.read(new Path(icebergSnapshot.manifestList()).getName());
+ assertThat(
+ metas.stream()
+ .filter(
+ m ->
+ m.content()
+ == IcebergManifestFileMeta.Content
+ .DATA))
+ .allMatch(m -> m.firstRowId() != null);
+ }
+
+ // a real Iceberg client sees the full history, reads the current state with the deletion
+ // vector applied, and can time travel to a replayed snapshot
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString());
+ Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t"));
+ assertThat(
+ java.util.stream.StreamSupport.stream(
+ icebergTable.snapshots().spliterator(), false)
+ .count())
+ .isEqualTo(latestSnapshotId);
+
+ assertThat(readIceberg(icebergTable, null)).containsExactlyInAnyOrder("1|1|a", "1|3|c");
+ // snapshot 2 is the first compaction: only the first two rows existed
+ assertThat(readIceberg(icebergTable, 2L)).containsExactlyInAnyOrder("1|1|a", "1|2|b");
+ }
+
+ @Test
+ public void testEnableOnUncompactedDvBucketExportsCompactedFiles() throws Exception {
+ FileStoreTable table = createPaimonTableWithoutIceberg();
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(row(RowKind.INSERT, 1, 1, "a"));
+ write.write(row(RowKind.INSERT, 1, 2, "b"));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ write.compact(partition(1), 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ write.write(row(RowKind.INSERT, 1, 3, "c"));
+ write.write(row(RowKind.DELETE, 1, 2, "b"));
+ commit.commit(3, write.prepareCommit(false, 3));
+
+ // this compaction produces a deletion vector against the max level file
+ write.compact(partition(1), 0, false);
+ commit.commit(4, write.prepareCommit(true, 4));
+
+ // a level-0 file on top of the compacted levels: the bucket's batch split is now NOT
+ // raw-convertible (level-0 file + overlapping key ranges + an active deletion vector)
+ write.write(row(RowKind.INSERT, 1, 4, "d"));
+ commit.commit(5, write.prepareCommit(false, 5));
+ write.close();
+ commit.close();
+
+ // enable Iceberg v3 WITHOUT full history sync; the next commit creates metadata from
+ // scratch while the bucket still has uncompacted level-0 files
+ Map options = new HashMap<>();
+ options.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ options.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ table = table.copy(options);
+ write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp"));
+ commit = table.newCommit(commitUser);
+
+ write.write(row(RowKind.INSERT, 1, 5, "e"));
+ commit.commit(6, write.prepareCommit(false, 6));
+
+ // The non-raw-convertible split must not be dropped wholesale: the files above level 0
+ // (with their deletion vector) are exactly what live incremental commits would have
+ // published, so Iceberg sees the data as of the last compaction. Only the level-0 rows
+ // (d, e) stay invisible until a compaction rewrites them.
+ HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString());
+ Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t"));
+ assertThat(readIceberg(icebergTable, null)).containsExactlyInAnyOrder("1|1|a", "1|3|c");
+
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(), new Path(table.location(), "metadata/v6.metadata.json"));
+ assertThat(metadata.nextRowId()).isNotNull();
+ IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table, paths);
+ assertThat(
+ manifestList
+ .read(new Path(metadata.currentSnapshot().manifestList()).getName())
+ .stream()
+ .filter(m -> m.content() == IcebergManifestFileMeta.Content.DATA))
+ .allMatch(m -> m.firstRowId() != null);
+
+ // a full compaction exports the level-0 rows through the incremental path
+ write.compact(partition(1), 0, true);
+ commit.commit(7, write.prepareCommit(true, 7));
+ write.close();
+ commit.close();
+
+ icebergTable.refresh();
+ assertThat(readIceberg(icebergTable, null))
+ .containsExactlyInAnyOrder("1|1|a", "1|3|c", "1|4|d", "1|5|e");
+ }
+
+ private static List readIceberg(Table icebergTable, Long snapshotId) throws Exception {
+ IcebergGenerics.ScanBuilder builder = IcebergGenerics.read(icebergTable);
+ if (snapshotId != null) {
+ builder = builder.useSnapshot(snapshotId);
+ }
+ List actual = new ArrayList<>();
+ try (CloseableIterable reader = builder.build()) {
+ // compare only the projected columns: Iceberg's generic reader may append materialized
+ // metadata columns (e.g. _pos while applying a deletion vector) to the output record
+ reader.forEach(
+ record ->
+ actual.add(record.get(0) + "|" + record.get(1) + "|" + record.get(2)));
+ }
+ return actual;
+ }
+
+ private FileStoreTable createPaimonTableWithoutIceberg() throws Exception {
+ RowType rowType =
+ new RowType(
+ Arrays.asList(
+ new DataField(0, "pt", DataTypes.INT().notNull()),
+ new DataField(1, "k", DataTypes.INT().notNull()),
+ new DataField(2, "v", DataTypes.STRING())));
+
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toString());
+
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, 1);
+ options.set(CoreOptions.FILE_FORMAT, "parquet");
+ options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofKibiBytes(32));
+ options.set(CoreOptions.DELETION_VECTORS_ENABLED, true);
+ options.set(CoreOptions.DELETION_VECTOR_BITMAP64, true);
+
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ options.toMap(),
+ "");
+
+ try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) {
+ paimonCatalog.createDatabase("mydb", false);
+ Identifier paimonIdentifier = Identifier.create("mydb", "t");
+ paimonCatalog.createTable(paimonIdentifier, schema, false);
+ return (FileStoreTable) paimonCatalog.getTable(paimonIdentifier);
+ }
+ }
+
+ private static GenericRow row(RowKind kind, int pt, int k, String v) {
+ return GenericRow.ofKind(kind, pt, k, BinaryString.fromString(v));
+ }
+
+ private static BinaryRow partition(int pt) {
+ BinaryRow partition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeInt(0, pt);
+ writer.complete();
+ return partition;
+ }
+}
diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
index b20aa47aa8c6..d96e892b01bb 100644
--- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
+++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java
@@ -32,6 +32,7 @@
import org.apache.paimon.iceberg.manifest.IcebergManifestList;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergSnapshot;
+import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.SchemaChange;
@@ -1763,6 +1764,308 @@ private static void removeRegisterTableEndpoint(IcebergRestMetadataCommitter com
endpointsField.set(sessionCatalog, endpoints);
}
+ /**
+ * The REST server (and the catalog table {@code mydb.t} on it) is shared across the tests in
+ * this class; in auto-recreate mode a leftover table from another test is silently dropped
+ * during the first commit, but the reconcile-mode tests refuse to touch a foreign table, so
+ * they must start clean.
+ */
+ private void dropLeftoverRestTable() {
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ if (restCatalog.tableExists(identifier)) {
+ restCatalog.dropTable(identifier, false);
+ }
+ }
+
+ @Test
+ public void testReconcileWhenRestCatalogBehind() throws Exception {
+ // REST publication misses several snapshots (here: sidecar generation continues while the
+ // committer is off); with auto-recreate disabled the committer must replay the missing
+ // snapshots onto the existing catalog table instead of dropping it.
+ dropLeftoverRestTable();
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ Map customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.REST_AUTO_RECREATE.key(), "false");
+ customOptions.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE_LOCATION.key(), "catalog-location");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ BaseTable icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ String tableUuid = icebergTable.operations().current().uuid();
+
+ // keep generating local metadata but stop publishing to the REST catalog
+ Map options = new HashMap<>();
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "table-location");
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.write(GenericRow.of(4, 40));
+ commit.commit(4, write.prepareCommit(false, 4));
+
+ // re-enable REST publication; the catalog is now three snapshots behind
+ options.put(IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "rest-catalog");
+ table = table.copy(options);
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(5, 50));
+ commit.commit(5, write.prepareCommit(false, 5));
+
+ icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(5);
+ assertThat(icebergTable.operations().current().uuid()).isEqualTo(tableUuid);
+ assertThat(ImmutableList.copyOf(icebergTable.snapshots()).size()).isEqualTo(5);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder(
+ "Record(1, 10)",
+ "Record(2, 20)",
+ "Record(3, 30)",
+ "Record(4, 40)",
+ "Record(5, 50)");
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testRepublishAlreadyPublishedSnapshotIsNoOp() throws Exception {
+ // Re-driving a publication that already landed (e.g. after an ambiguous commit or a
+ // restart) must not change the catalog table in either mode; in particular the legacy
+ // auto-recreate mode must not drop and recreate it just to reproduce identical content.
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ Collections.emptyMap());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.close();
+ commit.close();
+
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ BaseTable icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ String tableUuid = icebergTable.operations().current().uuid();
+
+ IcebergMetadata latestMetadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(catalogTableMetadataPath(table), "v2.metadata.json"));
+
+ // legacy mode (auto-recreate enabled, the default)
+ new IcebergRestMetadataCommitter(table).commitMetadata(latestMetadata, null);
+ icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(icebergTable.operations().current().uuid()).isEqualTo(tableUuid);
+
+ // reconcile mode
+ FileStoreTable reconcileTable =
+ table.copy(
+ Collections.singletonMap(IcebergOptions.REST_AUTO_RECREATE.key(), "false"));
+ new IcebergRestMetadataCommitter(reconcileTable).commitMetadata(latestMetadata, null);
+ icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(icebergTable.operations().current().uuid()).isEqualTo(tableUuid);
+ }
+
+ @Test
+ public void testRestCatalogAheadFailsWithoutDrop() throws Exception {
+ // Publishing stale metadata against a catalog that is already ahead must fail with a
+ // precise error and leave the catalog table untouched when auto-recreate is disabled.
+ dropLeftoverRestTable();
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ Collections.singletonMap(IcebergOptions.REST_AUTO_RECREATE.key(), "false"));
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ BaseTable icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(3);
+ String tableUuid = icebergTable.operations().current().uuid();
+
+ IcebergMetadata staleMetadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(catalogTableMetadataPath(table), "v2.metadata.json"));
+
+ IcebergRestMetadataCommitter committer = new IcebergRestMetadataCommitter(table);
+ assertThatThrownBy(() -> committer.commitMetadata(staleMetadata, null))
+ .isInstanceOf(IcebergRestCatalogOutOfSyncException.class)
+ .hasMessageContaining("ahead");
+
+ icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(3);
+ assertThat(icebergTable.operations().current().uuid()).isEqualTo(tableUuid);
+ }
+
+ @Test
+ public void testReconcileRegeneratedHistory() throws Exception {
+ // Locally regenerated metadata (deleted metadata directory + full-history rebuild) must
+ // be publishable onto the existing catalog table without dropping it: the head snapshot
+ // is replayed, stale entries whose sequence numbers cannot be reused are removed.
+ dropLeftoverRestTable();
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ Map customOptions = new HashMap<>();
+ customOptions.put(IcebergOptions.REST_AUTO_RECREATE.key(), "false");
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ BaseTable icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ String tableUuid = icebergTable.operations().current().uuid();
+
+ // force a from-scratch rebuild of the local Iceberg metadata
+ table.fileIO().delete(catalogTableMetadataPath(table), true);
+ table =
+ table.copy(
+ Collections.singletonMap(IcebergOptions.SYNC_FULL_HISTORY.key(), "true"));
+ write.close();
+ write = table.newWrite(commitUser);
+ commit.close();
+ commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(3, 30));
+ commit.commit(3, write.prepareCommit(false, 3));
+ write.close();
+ commit.close();
+
+ icebergTable = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(3);
+ assertThat(icebergTable.operations().current().uuid()).isEqualTo(tableUuid);
+ // snapshots 1 and 2 were regenerated under already-consumed sequence numbers, so the
+ // catalog keeps only the replayed head
+ assertThat(ImmutableList.copyOf(icebergTable.snapshots()).size()).isEqualTo(1);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)", "Record(3, 30)");
+ }
+
+ @Test
+ public void testRetryRepublishesExistingMetadata() throws Exception {
+ // A retried commit whose metadata file already exists must still re-drive the REST
+ // publication: the file only proves the metadata was generated, not that it reached the
+ // catalog.
+ dropLeftoverRestTable();
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ -1,
+ "avro",
+ Collections.emptyMap());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.close();
+ commit.close();
+
+ TableIdentifier identifier = TableIdentifier.of("mydb", "t");
+ Table icebergTable = restCatalog.loadTable(identifier);
+ assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(2);
+ String tableUuid = ((BaseTable) icebergTable).operations().current().uuid();
+
+ // simulate a publication that never reached the catalog for the latest snapshot
+ icebergTable.manageSnapshots().rollbackTo(1).commit();
+ assertThat(restCatalog.loadTable(identifier).currentSnapshot().snapshotId()).isEqualTo(1);
+
+ // a Flink-style retry of the last committable republishes from the existing files
+ try (IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser)) {
+ callback.retry(new ManifestCommittable(2L));
+ }
+
+ BaseTable reloaded = (BaseTable) restCatalog.loadTable(identifier);
+ assertThat(reloaded.currentSnapshot().snapshotId()).isEqualTo(2);
+ assertThat(reloaded.operations().current().uuid()).isEqualTo(tableUuid);
+ assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(2, 20)");
+ }
+
private static class TestRecord {
private final BinaryRow partition;
private final GenericRow record;