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..f7632d3ecd3c 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,397 @@ 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.
+ *
+ *
Every replay step writes to a uniquely named staging path; the published chain, the
+ * version hint and the external catalog stay untouched until the final staged metadata is
+ * durable, and only then is the staged chain promoted into place. A rebuild that fails at any
+ * point therefore leaves the previously published metadata fully readable, and its staged
+ * leftovers are removed by the next rebuild. 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);
+
+ deleteStagedLeftovers();
+
+ // 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 (isUsableReplayBase(metadata, id, startId)) {
+ baseId = id;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read existing Iceberg metadata {}, rebuilding history from scratch",
+ metadataPath,
+ e);
+ }
+ break;
+ }
+ }
+
+ String rebuildUuid = UUID.randomUUID().toString();
+ long firstStagedId;
+ boolean freshRebuild = baseId == -1;
+ StaleBuild staleBuild = null;
+ if (freshRebuild) {
+ // No usable base: the whole old build is stale. Nothing of it is touched during the
+ // replay, so an external catalog that still points at the old metadata keeps a fully
+ // readable table; the old files are replaced and cleaned only by the promotion. Their
+ // references are collected up front, tolerating unreadable files (that is what
+ // triggered some rebuilds in the first place).
+ staleBuild = collectStaleBuild(currentSnapshotId);
+ firstStagedId = startId;
+ createMetadataWithoutBase(
+ startId,
+ inheritUuid,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, startId));
+ } else {
+ firstStagedId = baseId + 1;
+ }
+
+ for (long id = firstStagedId == startId ? startId + 1 : firstStagedId;
+ id <= currentSnapshotId;
+ id++) {
+ long snapshotId = id;
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ Path basePath =
+ snapshotId - 1 < firstStagedId
+ ? pathFactory.toMetadataPath(snapshotId - 1)
+ : stagedMetadataPath(rebuildUuid, snapshotId - 1);
+ createMetadataWithBase(
+ (removedFiles, addedFiles) ->
+ collectFileChanges(snapshotId, removedFiles, addedFiles),
+ indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX),
+ snapshot,
+ basePath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, snapshotId));
+ }
+
+ boolean promoted =
+ promoteStagedReplay(rebuildUuid, firstStagedId, currentSnapshotId, startId);
+ if (promoted && staleBuild != null) {
+ deleteStaleBuild(staleBuild, currentSnapshotId, startId);
+ }
+ }
+
+ /**
+ * Whether an existing metadata file can serve as the resume base of a full-history replay. On
+ * top of the structural checks, the base must describe the live timeline: an abandoned base (a
+ * rolled-back Paimon snapshot reused the id) or one carrying re-evolved schema definitions
+ * would be rejected again by the first replay step, and a rebuild must never select a base its
+ * own replay refuses to extend.
+ */
+ private boolean isUsableReplayBase(IcebergMetadata metadata, long baseId, long startId) {
+ if (!isSameFormatVersion(metadata.formatVersion())) {
+ return false;
+ }
+ if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3 && metadata.nextRowId() == null) {
+ return false;
+ }
+ if (!coversRetainedPrefix(metadata, baseId, startId)) {
+ return false;
+ }
+ SnapshotManager snapshotManager = table.snapshotManager();
+ if (!snapshotManager.snapshotExists(baseId)
+ || !metadataMatchesSnapshot(metadata, snapshotManager.snapshot(baseId))) {
+ return false;
+ }
+ SchemaCache schemaCache = new SchemaCache();
+ long latestSchemaId = schemaCache.getLatestSchemaId();
+ for (IcebergSchema known : metadata.schemas()) {
+ if (known.schemaId() > latestSchemaId
+ || !known.equals(schemaCache.get(known.schemaId()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private Path stagedMetadataPath(String rebuildUuid, long snapshotId) {
+ return new Path(
+ pathFactory.metadataDirectory(),
+ String.format("rebuild-%s-v%d.metadata.json", rebuildUuid, snapshotId));
+ }
+
+ /** Staged files of crashed or superseded rebuilds; only ever garbage. */
+ private void deleteStagedLeftovers() throws IOException {
+ FileStatus[] statuses;
+ try {
+ statuses = table.fileIO().listStatus(pathFactory.metadataDirectory());
+ } catch (FileNotFoundException e) {
+ return;
+ }
+ for (FileStatus status : statuses) {
+ String name = status.getPath().getName();
+ if (name.startsWith("rebuild-") && name.endsWith(".metadata.json")) {
+ table.fileIO().deleteQuietly(status.getPath());
+ }
+ }
+ }
+
+ /**
+ * Switch the published chain to the staged one, then the version hint and the external catalog.
+ * The staged head is durable and validated before any published path is touched; the remaining
+ * window is the per-file replacement of each chain position, ending one below the head, and
+ * every path serves complete metadata again as soon as its replacement lands.
+ *
+ * @return false if a newer commit superseded this rebuild; the staged files are discarded and
+ * the published chain is left for that commit's own rebuild
+ */
+ private boolean promoteStagedReplay(
+ String rebuildUuid, long firstStagedId, long currentSnapshotId, long startId)
+ throws IOException {
+ IcebergMetadata head =
+ IcebergMetadata.fromPath(
+ table.fileIO(), stagedMetadataPath(rebuildUuid, currentSnapshotId));
+ Preconditions.checkState(
+ head.currentSnapshotId() == currentSnapshotId,
+ "Staged replay head is at snapshot %s instead of %s",
+ head.currentSnapshotId(),
+ currentSnapshotId);
+ // a truncated chain must never be promoted, no matter what produced it
+ Preconditions.checkState(
+ coversRetainedPrefix(head, currentSnapshotId, startId),
+ "Staged replay head for snapshot %s does not cover the retained history",
+ currentSnapshotId);
+
+ Long latest = table.snapshotManager().latestSnapshotId();
+ if (latest == null || latest != currentSnapshotId) {
+ for (long id = firstStagedId; id <= currentSnapshotId; id++) {
+ table.fileIO().deleteQuietly(stagedMetadataPath(rebuildUuid, id));
+ }
+ return false;
+ }
+
+ for (long id = firstStagedId; id <= currentSnapshotId; id++) {
+ Path staged = stagedMetadataPath(rebuildUuid, id);
+ Path published = pathFactory.toMetadataPath(id);
+ if (table.fileIO().exists(published)) {
+ table.fileIO().deleteQuietly(published);
+ }
+ if (!table.fileIO().rename(staged, published)) {
+ table.fileIO().deleteQuietly(published);
+ if (!table.fileIO().rename(staged, published)) {
+ throw new IllegalStateException(
+ "Failed to promote staged Iceberg metadata " + staged);
+ }
+ }
+ }
+
+ Path headPath = pathFactory.toMetadataPath(currentSnapshotId);
+ table.fileIO()
+ .overwriteFileUtf8(
+ new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME),
+ String.valueOf(currentSnapshotId));
+ commitToExternalCatalog(head, headPath, null, null);
+ deleteApplicableMetadataFiles(currentSnapshotId);
+ return true;
+ }
+
+ /** Identity recovered from the newest readable metadata below an unreadable base. */
+ private static class RecoveredIdentity {
+ private @Nullable String uuid;
+ private int lastColumnId;
+ }
+
+ private RecoveredIdentity recoverIdentityBelow(long snapshotId) {
+ RecoveredIdentity recovered = new RecoveredIdentity();
+ try {
+ long newestReadable = -1;
+ Iterator it =
+ pathFactory.getAllMetadataPathBefore(table.fileIO(), snapshotId).iterator();
+ while (it.hasNext()) {
+ Path path = it.next();
+ long version = metadataVersionOf(path);
+ if (version <= newestReadable) {
+ continue;
+ }
+ try {
+ IcebergMetadata metadata = IcebergMetadata.fromPath(table.fileIO(), path);
+ newestReadable = version;
+ recovered.uuid = metadata.tableUuid();
+ recovered.lastColumnId =
+ Math.max(recovered.lastColumnId, metadata.lastColumnId());
+ } catch (Exception ignored) {
+ // unreadable files are exactly what this recovery works around
+ }
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to scan older Iceberg metadata for identity recovery.", e);
+ }
+ return recovered;
+ }
+
+ /** 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 +872,21 @@ private void createMetadataWithoutBase(
int lastColumnIdFloor,
long nextRowIdFloor)
throws IOException {
+ createMetadataWithoutBase(snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor, null);
+ }
+
+ /**
+ * @param stagedTarget when non-null, this metadata is a step of a {@link #rebuildFullHistory}
+ * replay: it is written to this staging path instead of the published one, and nothing is
+ * published; the replay promotes the staged chain after its final step.
+ */
+ private void createMetadataWithoutBase(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ @Nullable Path stagedTarget)
+ throws IOException {
SnapshotReader snapshotReader = table.newSnapshotReader().withSnapshot(snapshotId);
Snapshot paimonSnapshot = table.snapshotManager().snapshot(snapshotId);
SchemaCache schemaCache = new SchemaCache();
@@ -486,20 +895,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 +997,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
@@ -618,6 +1044,16 @@ private void createMetadataWithoutBase(
nextRowId,
refs);
+ if (stagedTarget != null) {
+ // a replay step: only the staging path is touched, the published chain stays
+ // intact until the replay promotes; staging names are unique, so a failed write
+ // is a hard error, never a twin
+ if (!table.fileIO().tryToWriteAtomic(stagedTarget, metadata.toJson())) {
+ throw new IllegalStateException(
+ "Failed to write staged Iceberg metadata " + stagedTarget);
+ }
+ return;
+ }
Path metadataPath = pathFactory.toMetadataPath(snapshotId);
// atomic-first: where rename overwrites, a stale twin is replaced with no window at
// all; otherwise fall back to delete-then-write, the smallest window available
@@ -647,28 +1083,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 +1159,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 +1172,7 @@ private void dataSplitToManifestEntries(
dataSplit.partition(),
deletionFile.cardinality(),
-1,
- rawFile.path(),
+ filePath,
deletionFile.offset(),
deletionFile.length());
@@ -937,8 +1407,63 @@ private void createMetadataWithBase(
int lastColumnIdFloor,
long nextRowIdFloor)
throws IOException {
+ createMetadataWithBase(
+ fileChangesCollector,
+ indexFiles,
+ snapshot,
+ baseMetadataPath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ null);
+ }
+
+ /**
+ * @param stagedTarget when non-null, this metadata is a step of a {@link #rebuildFullHistory}
+ * replay: it is written to this staging path instead of the published one, and nothing is
+ * published; the replay promotes the staged chain after its final step.
+ */
+ private void createMetadataWithBase(
+ FileChangesCollector fileChangesCollector,
+ List indexFiles,
+ Snapshot snapshot,
+ Path baseMetadataPath,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ @Nullable Path stagedTarget)
+ 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) {
+ if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3) {
+ // the unreadable base may already have issued row ids that no other file
+ // records; rebuilding without its high-water mark would reuse them, so fail
+ // until the file is repaired or its history is removed together with it
+ // (the external catalog's watermark could lift this in the future)
+ throw new IllegalStateException(
+ "Base Iceberg metadata "
+ + baseMetadataPath
+ + " is unreadable and format version 3 forbids rebuilding "
+ + "without its row-id high-water mark. Repair or remove the "
+ + "Iceberg metadata directory to republish from scratch.",
+ e);
+ }
+ // v2 has no row lineage to reset: recreate instead of failing the commit, keeping
+ // the identity of the newest readable metadata so catalogs track the same table
+ LOG.warn(
+ "Unreadable base Iceberg metadata {}, recreating metadata.",
+ baseMetadataPath,
+ e);
+ RecoveredIdentity recovered = recoverIdentityBelow(snapshotId);
+ recreateFromUnusableBase(
+ snapshotId,
+ recovered.uuid,
+ Math.max(lastColumnIdFloor, recovered.lastColumnId),
+ nextRowIdFloor,
+ stagedTarget != null);
+ return;
+ }
// row ids handed out by the base or by abandoned metadata must never be reused
long rowIdFloor =
Math.max(
@@ -955,32 +1480,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,
+ stagedTarget != null);
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,
+ stagedTarget != null);
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,
+ stagedTarget != null);
return;
}
@@ -1001,11 +1529,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,
+ stagedTarget != null);
return;
}
}
@@ -1020,11 +1549,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,
+ stagedTarget != null);
return;
}
}
@@ -1241,6 +1771,14 @@ private void createMetadataWithBase(
nextRowId,
refs);
+ if (stagedTarget != null) {
+ // a replay step: only the staging path is touched; see the no-base path
+ if (!table.fileIO().tryToWriteAtomic(stagedTarget, metadata.toJson())) {
+ throw new IllegalStateException(
+ "Failed to write staged Iceberg metadata " + stagedTarget);
+ }
+ return;
+ }
Path metadataPath = pathFactory.toMetadataPath(snapshotId);
// atomic-first: see the no-base path
boolean written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson());
@@ -1274,6 +1812,34 @@ 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 insideReplay)
+ throws IOException {
+ if (insideReplay) {
+ // the base of a replay step was written or validated by the replay itself, so an
+ // unusable one means interference or corruption; writing reduced metadata here
+ // would silently truncate the replayed history, and recursing into another
+ // replay from inside this one does not terminate. Fail the commit instead: the
+ // published chain is untouched and the next commit restarts the rebuild.
+ throw new IllegalStateException(
+ "The base metadata of a full-history replay step for snapshot "
+ + snapshotId
+ + " is unusable; failing the rebuild so it restarts cleanly.");
+ }
+ recreateMetadata(snapshotId, inheritUuid, lastColumnIdFloor, nextRowIdFloor);
+ }
+
private interface FileChangesCollector {
boolean collect(
Map> removedFiles,
@@ -1560,16 +2126,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 +2166,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..89cf5b11a53b 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,22 @@ 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 URI =
key("metadata.iceberg.uri")
.stringType()
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..232a6a37c2d5
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergSyncFullHistoryTest.java
@@ -0,0 +1,541 @@
+/*
+ * 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;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * 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 testCorruptedV3BaseFailsClosed() 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.FORMAT_VERSION.key(), "3");
+ options.put(IcebergOptions.METADATA_DELETE_AFTER_COMMIT.key(), "false");
+ createAppendTable(options);
+ writeCommit(1, GenericRow.of(1, 10));
+ writeCommit(2, GenericRow.of(2, 20));
+ assertThat(readMetadata(2).nextRowId()).isEqualTo(2);
+ String metadata1Before =
+ table.fileIO()
+ .readFileUtf8(new Path(table.location(), "metadata/v1.metadata.json"));
+
+ // the unreadable base may have issued row ids no other file records; rebuilding
+ // without its high-water mark would reuse them, so the commit must fail instead
+ Path corrupted = new Path(table.location(), "metadata/v2.metadata.json");
+ table.fileIO().deleteQuietly(corrupted);
+ table.fileIO().overwriteFileUtf8(corrupted, "{ not json");
+
+ assertThatThrownBy(() -> writeCommit(3, GenericRow.of(3, 30)))
+ .hasStackTraceContaining("forbids rebuilding");
+ // nothing was reset or replaced
+ assertThat(
+ table.fileIO()
+ .readFileUtf8(
+ new Path(table.location(), "metadata/v1.metadata.json")))
+ .isEqualTo(metadata1Before);
+ assertThat(table.fileIO().exists(new Path(table.location(), "metadata/v3.metadata.json")))
+ .isFalse();
+ }
+
+ @Test
+ public void testAbandonedResumeCandidateDoesNotRecurse() throws Exception {
+ // a rollback plus an Iceberg-disabled commit reuses snapshot id 2 on a new timeline,
+ // leaving the old v2 metadata abandoned; the rebuild must reject it as a resume base
+ // instead of selecting it and recursing when the replay refuses to extend it
+ 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));
+
+ table.rollbackTo(1);
+
+ Map disable = new HashMap<>();
+ disable.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.DISABLED.toString());
+ reopen(disable);
+ writeCommit(3, GenericRow.of(22, 220));
+
+ Map enable = new HashMap<>();
+ enable.put(
+ IcebergOptions.METADATA_ICEBERG_STORAGE.key(),
+ IcebergOptions.StorageType.TABLE_LOCATION.toString());
+ reopen(enable);
+ writeCommit(4, GenericRow.of(3, 30));
+
+ IcebergMetadata metadata = readMetadata(table.snapshotManager().latestSnapshotId());
+ assertThat(
+ metadata.snapshots().stream()
+ .map(IcebergSnapshot::snapshotId)
+ .collect(Collectors.toList()))
+ .contains(1L)
+ .doesNotHaveDuplicates();
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1, 10)", "Record(22, 220)", "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/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;
+ }
+}