From a53351c3b9680caf9a023d84267560a9f3b770b4 Mon Sep 17 00:00:00 2001 From: Jiajia Li Date: Wed, 5 Aug 2026 06:54:14 -0400 Subject: [PATCH] [iceberg] Fix Iceberg metadata after a rollback --- .../paimon/iceberg/IcebergCommitCallback.java | 501 +++++++++-- .../paimon/table/AbstractFileStoreTable.java | 3 + .../iceberg/IcebergCompatibilityTest.java | 851 ++++++++++++++++++ .../RecordingIcebergMetadataCommitter.java | 78 ++ .../org.apache.paimon.factories.Factory | 1 + .../iceberg/IcebergRestMetadataCommitter.java | 55 +- .../IcebergRestMetadataCommitterTest.java | 231 +++++ 7 files changed, 1648 insertions(+), 72 deletions(-) create mode 100644 paimon-core/src/test/java/org/apache/paimon/iceberg/RecordingIcebergMetadataCommitter.java 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 336b0fe8652c..4b6776d3396b 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 @@ -25,6 +25,7 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.factories.FactoryException; import org.apache.paimon.factories.FactoryUtil; +import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.iceberg.manifest.IcebergConversions; import org.apache.paimon.iceberg.manifest.IcebergDataFileMeta; @@ -79,6 +80,7 @@ import javax.annotation.Nullable; +import java.io.FileNotFoundException; import java.io.IOException; import java.io.UncheckedIOException; import java.util.ArrayList; @@ -296,6 +298,23 @@ private void createMetadata( List indexFiles) { long snapshotId = snapshot.id(); try { + // a stale callback outliving a rollback (or drop and recreate) must do nothing; + // read the snapshot file directly - snapshot caches may predate the rollback + Snapshot current; + try { + current = + SnapshotManager.tryFromPath( + table.fileIO(), table.snapshotManager().snapshotPath(snapshotId)); + } catch (FileNotFoundException e) { + return; + } + if (!commitIdentity(current).equals(commitIdentity(snapshot))) { + return; + } + // the snapshot cache may still hold a rolled-back timeline under reused ids; + // every by-id read below must see the disk state the guard just verified + table.snapshotManager().invalidateCache(); + if (snapshotId == Snapshot.FIRST_SNAPSHOT_ID) { // If Iceberg metadata is stored separately in another directory, dropping the table // will not delete old Iceberg metadata. So we delete them here, when the table is @@ -303,8 +322,84 @@ private void createMetadata( table.fileIO().delete(pathFactory.metadataDirectory(), true); } + String abandonedUuid = null; + int abandonedLastColumnId = 0; if (table.fileIO().exists(pathFactory.toMetadataPath(snapshotId))) { - return; + if (metadataMatchesSnapshot(snapshotId, snapshot)) { + // a retry repairs hint, pointer and files only while this snapshot is + // still the head; a replay of an older committable must not move them back + Long latestForRepair = table.snapshotManager().latestSnapshotId(); + if (latestForRepair == null || latestForRepair != snapshotId) { + return; + } + if (readVersionHint() != snapshotId) { + table.fileIO() + .overwriteFileUtf8( + new Path( + pathFactory.metadataDirectory(), + VERSION_HINT_FILENAME), + String.valueOf(snapshotId)); + } + if (metadataCommitter != null) { + // recommit the pointer (previous version as base, so a lagging + // catalog advances) before retiring files it may still reference + Path existingPath = pathFactory.toMetadataPath(snapshotId); + Path basePath = pathFactory.toMetadataPath(snapshotId - 1); + IcebergMetadata base = + table.fileIO().exists(basePath) + ? IcebergMetadata.fromPath(table.fileIO(), basePath) + : null; + commitToExternalCatalog( + IcebergMetadata.fromPath(table.fileIO(), existingPath), + existingPath, + base, + base == null ? null : basePath); + } + // a failed earlier attempt skipped the normal post-publication cleanup + deleteApplicableMetadataFiles(snapshotId); + retireAbandonedSuffix(); + table.fileIO() + .deleteQuietly( + new Path( + pathFactory.metadataDirectory(), + RETIRE_PENDING_FILENAME)); + return; + } + // a reused snapshot id: only the current head may replace the abandoned + // metadata; a delayed replay must not move hint or pointer backwards + Long latestNow = table.snapshotManager().latestSnapshotId(); + if (latestNow == null || latestNow != snapshotId) { + return; + } + // read the identity now, delete only at the write site: readers and the + // catalog pointer keep a working file until the replacement is built + IcebergMetadata abandoned = tryReadMetadata(pathFactory.toMetadataPath(snapshotId)); + if (abandoned != null) { + abandonedUuid = abandoned.tableUuid(); + abandonedLastColumnId = abandoned.lastColumnId(); + } + } + // steady-state commits skip the listing; anything suspicious lists the actual + // files, because the hint alone can lag while readers still probe past it + Path retirePending = new Path(pathFactory.metadataDirectory(), RETIRE_PENDING_FILENAME); + boolean suspectRollback = + abandonedUuid != null + || readVersionHint() != snapshotId - 1 + || table.fileIO().exists(pathFactory.toMetadataPath(snapshotId + 1)) + || table.fileIO().exists(retirePending); + long newestExisting = suspectRollback ? newestExistingMetadataVersion() : -1; + boolean retireSuffix = abandonedUuid != null || newestExisting > snapshotId; + if (retireSuffix && newestExisting > snapshotId) { + // the newest abandoned version carries the authoritative high-water mark + IcebergMetadata surviving = + tryReadMetadata(pathFactory.toMetadataPath(newestExisting)); + if (surviving != null) { + if (abandonedUuid == null) { + abandonedUuid = surviving.tableUuid(); + } + abandonedLastColumnId = + Math.max(abandonedLastColumnId, surviving.lastColumnId()); + } } Path baseMetadataPath = pathFactory.toMetadataPath(snapshotId - 1); @@ -320,9 +415,19 @@ private void createMetadata( .equals(DELETION_VECTORS_INDEX)) .collect(Collectors.toList()), snapshot, - baseMetadataPath); + baseMetadataPath, + abandonedLastColumnId); } else { - createMetadataWithoutBase(snapshotId); + createMetadataWithoutBase(snapshotId, abandonedUuid, abandonedLastColumnId); + } + + if (retireSuffix) { + // only after the replacement is durable, so readers keep a working head + retireAbandonedSuffix(); + } + if (suspectRollback) { + // the listing ran and every leftover above the head is gone + table.fileIO().deleteQuietly(retirePending); } } catch (IOException e) { throw new UncheckedIOException(e); @@ -334,6 +439,17 @@ private void createMetadata( // ------------------------------------------------------------------------------------- private void createMetadataWithoutBase(long snapshotId) throws IOException { + createMetadataWithoutBase(snapshotId, null, 0); + } + + private void createMetadataWithoutBase(long snapshotId, @Nullable String inheritUuid) + throws IOException { + createMetadataWithoutBase(snapshotId, inheritUuid, 0); + } + + private void createMetadataWithoutBase( + long snapshotId, @Nullable String inheritUuid, int lastColumnIdFloor) + throws IOException { SnapshotReader snapshotReader = table.newSnapshotReader().withSnapshot(snapshotId); Snapshot paimonSnapshot = table.snapshotManager().snapshot(snapshotId); SchemaCache schemaCache = new SchemaCache(); @@ -396,7 +512,9 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException { String manifestListFileName = manifestList.writeWithoutRolling(allManifestFileMetas); + // current schema follows the latest; the snapshot entry records its own schema int schemaId = (int) schemaCache.getLatestSchemaId(); + int snapshotSchemaId = (int) paimonSnapshot.schemaId(); IcebergSchema icebergSchema = schemaCache.get(schemaId); List partitionFields = getPartitionFields(table.schema().partitionKeys(), icebergSchema); @@ -410,10 +528,11 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException { snapshotId, snapshotId, snapshotId == Snapshot.FIRST_SNAPSHOT_ID ? null : (Long) (snapshotId - 1), - System.currentTimeMillis(), + // the Paimon snapshot's own commit time, the as-of time readers see + paimonSnapshot.timeMillis(), snapshotSummary, pathFactory.toManifestListPath(manifestListFileName).toString(), - schemaId, + snapshotSchemaId, null, null); @@ -425,7 +544,9 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException { // After https://github.com/apache/paimon/issues/6107 we can add tags here. Map refs = new HashMap<>(); - String tableUuid = UUID.randomUUID().toString(); + // keep the identity of the metadata this rebuild replaces, so already loaded readers + // and external catalogs keep refreshing the same table + String tableUuid = inheritUuid != null ? inheritUuid : UUID.randomUUID().toString(); List allSchemas = IntStream.rangeClosed(0, schemaId) @@ -437,7 +558,14 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException { tableUuid, table.location().toString(), snapshotId, - icebergSchema.highestFieldId(), + // every emitted schema counts, and a rebuild must not regress + // below the replaced metadata's high-water mark + Math.max( + lastColumnIdFloor, + allSchemas.stream() + .mapToInt(IcebergSchema::highestFieldId) + .max() + .orElse(icebergSchema.highestFieldId())), allSchemas, schemaId, Collections.singletonList(new IcebergPartitionSpec(partitionFields)), @@ -452,26 +580,31 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException { refs); Path metadataPath = pathFactory.toMetadataPath(snapshotId); - table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); - table.fileIO() - .overwriteFileUtf8( - new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), - String.valueOf(snapshotId)); - - expireAllBefore(snapshotId); - - if (metadataCommitter != null) { - switch (metadataCommitter.identifier()) { - case "hive": - metadataCommitter.commitMetadata(metadataPath, null); - break; - case "rest": - metadataCommitter.commitMetadata(metadata, null); - break; - default: - throw new UnsupportedOperationException( - "Unsupported metadata committer: " + metadataCommitter.identifier()); - } + // 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 + boolean written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); + if (!written + && table.fileIO().exists(metadataPath) + && !metadataMatchesSnapshot(snapshotId, paimonSnapshot)) { + table.fileIO().deleteQuietly(metadataPath); + written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); + } + if (!written && !metadataMatchesSnapshot(snapshotId, paimonSnapshot)) { + // no twin published this snapshot's metadata; fail so the commit retries + 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 + Long latestAtPublish = table.snapshotManager().latestSnapshotId(); + if (latestAtPublish != null && latestAtPublish == snapshotId) { + table.fileIO() + .overwriteFileUtf8( + new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), + String.valueOf(snapshotId)); + commitToExternalCatalog(metadata, metadataPath, null, null); + // cleanup only after the catalog serves the new head: a skipped or failed + // publication must not delete files an external pointer still references + expireAllBefore(snapshotId); } } @@ -608,21 +741,232 @@ private static void collectVariantFields( // Create metadata based on old ones // ------------------------------------------------------------------------------------- + /** + * Whether the existing metadata for {@code snapshotId} was built from this very Paimon + * snapshot, judged by the commit identity in the snapshot summary. Unreadable counts as a + * mismatch; metadata without an identity (older releases) is trusted, so the protection only + * covers metadata written since. + * + *

A replacement reuses its metadata version (versions are keyed by Paimon snapshot id), so + * readers that already loaded the abandoned version converge only after reloading the table. + */ + /** + * Deletes every metadata version above the current Paimon snapshots, which would otherwise + * shadow the replaced timeline for readers probing past the hint. A failed deletion fails the + * commit so a retry finishes the job; referenced manifests are left to orphan cleanup (the + * shared prefix makes reference counting non-trivial). + */ + private void retireAbandonedSuffix() throws IOException { + for (FileStatus status : table.fileIO().listStatus(pathFactory.metadataDirectory())) { + String name = status.getPath().getName(); + if (!name.startsWith("v") || !name.endsWith(".metadata.json")) { + continue; + } + long version; + try { + version = Long.parseLong(name.substring(1, name.indexOf('.'))); + } catch (NumberFormatException ignored) { + continue; + } + Long latestNow = table.snapshotManager().latestSnapshotId(); + if (latestNow == null || version <= latestNow) { + continue; + } + table.fileIO().deleteQuietly(status.getPath()); + if (table.fileIO().exists(status.getPath())) { + throw new IllegalStateException( + "Failed to retire abandoned Iceberg metadata " + status.getPath()); + } + } + } + + static final String RETIRE_PENDING_FILENAME = "retire-pending"; + + /** + * Marks that a rollback may have left abandoned metadata behind; written before the rollback + * deletes anything, and cleared once a commit has listed and retired the leftovers. + */ + public static void markRetirePendingForRollback(FileStoreTable table) { + if (table.coreOptions().toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE) + == IcebergOptions.StorageType.DISABLED) { + return; + } + try { + Path dir = catalogTableMetadataPath(table); + if (table.fileIO().exists(dir)) { + table.fileIO().overwriteFileUtf8(new Path(dir, RETIRE_PENDING_FILENAME), ""); + } + } catch (Exception e) { + // best-effort: the commit-time suspicion gate still covers the common cases + } + } + + /** The version recorded in the hint file, or -1 when absent or unreadable. */ + private long readVersionHint() { + try { + return Long.parseLong( + table.fileIO() + .readFileUtf8( + new Path( + pathFactory.metadataDirectory(), VERSION_HINT_FILENAME)) + .trim()); + } catch (Exception e) { + return -1; + } + } + + private void commitToExternalCatalog( + IcebergMetadata metadata, + Path metadataPath, + @Nullable IcebergMetadata baseMetadata, + @Nullable Path baseMetadataPath) { + if (metadataCommitter == null) { + return; + } + switch (metadataCommitter.identifier()) { + case "hive": + metadataCommitter.commitMetadata(metadataPath, baseMetadataPath); + break; + case "rest": + metadataCommitter.commitMetadata(metadata, baseMetadata); + break; + default: + throw new UnsupportedOperationException( + "Unsupported metadata committer: " + metadataCommitter.identifier()); + } + } + + /** The newest existing metadata file version, or -1 when there is none. */ + private long newestExistingMetadataVersion() throws IOException { + FileStatus[] statuses; + try { + statuses = table.fileIO().listStatus(pathFactory.metadataDirectory()); + } catch (FileNotFoundException e) { + // only a missing directory counts as empty; a transient listing failure must + // fail the commit, or a stale suffix would silently survive + return -1; + } + long newest = -1; + for (FileStatus status : statuses) { + String name = status.getPath().getName(); + if (!name.startsWith("v") || !name.endsWith(".metadata.json")) { + continue; + } + try { + newest = Math.max(newest, Long.parseLong(name.substring(1, name.indexOf('.')))); + } catch (NumberFormatException ignored) { + } + } + return newest; + } + + /** The given metadata file, or null when unreadable. */ + @Nullable + private IcebergMetadata tryReadMetadata(Path metadataPath) { + try { + return IcebergMetadata.fromPath(table.fileIO(), metadataPath); + } catch (Exception e) { + return null; + } + } + + private boolean metadataMatchesSnapshot(long snapshotId, Snapshot snapshot) { + try { + IcebergMetadata existing = + IcebergMetadata.fromPath( + table.fileIO(), pathFactory.toMetadataPath(snapshotId)); + return metadataMatchesSnapshot(existing, snapshot); + } catch (Exception e) { + return false; + } + } + + private static boolean metadataMatchesSnapshot(IcebergMetadata metadata, Snapshot snapshot) { + if (metadata.currentSnapshot() == null) { + return false; + } + String identity = + metadata.currentSnapshot().summary().get(SNAPSHOT_SUMMARY_PAIMON_COMMIT_IDENTITY); + return identity == null || identity.equals(commitIdentity(snapshot)); + } + private void createMetadataWithBase( FileChangesCollector fileChangesCollector, List indexFiles, Snapshot snapshot, - Path baseMetadataPath) + Path baseMetadataPath, + int lastColumnIdFloor) throws IOException { long snapshotId = snapshot.id(); IcebergMetadata baseMetadata = IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath); + // a base left on the abandoned timeline must be rebuilt, not extended + if (table.snapshotManager().snapshotExists(snapshotId - 1) + && !metadataMatchesSnapshot( + baseMetadata, table.snapshotManager().snapshot(snapshotId - 1))) { + Long latestNow = table.snapshotManager().latestSnapshotId(); + if (latestNow == null || latestNow != snapshotId) { + // a delayed replay on the abandoned timeline: leave publication to the head + return; + } + // keep the stale base's identity so external catalogs do not recreate the table + createMetadataWithoutBase( + snapshotId, + baseMetadata.tableUuid(), + Math.max(lastColumnIdFloor, baseMetadata.lastColumnId())); + return; + } + if (!isSameFormatVersion(baseMetadata.formatVersion())) { // we need to recreate iceberg metadata if format version changed - createMetadataWithoutBase(snapshot.id()); + createMetadataWithoutBase( + snapshot.id(), null, Math.max(lastColumnIdFloor, baseMetadata.lastColumnId())); return; } + // decide the schema story before any manifest is written + SchemaCache schemaCache = new SchemaCache(); + int schemaId = (int) schemaCache.getLatestSchemaId(); + int snapshotSchemaId = (int) snapshot.schemaId(); + IcebergSchema icebergSchema = schemaCache.get(schemaId); + // re-verified each commit: a rollback re-evolution can redefine an already + // verified id while this callback only ever sees increasing snapshot ids + for (IcebergSchema known : baseMetadata.schemas()) { + if (known.schemaId() > schemaId) { + continue; + } + IcebergSchema current = + known.schemaId() == schemaId + ? icebergSchema + : schemaCache.get(known.schemaId()); + if (!known.equals(current)) { + // a re-evolution reused this id with different fields; rebuild from scratch + createMetadataWithoutBase( + snapshot.id(), + baseMetadata.tableUuid(), + Math.max(lastColumnIdFloor, baseMetadata.lastColumnId())); + return; + } + } + if (schemaId < baseMetadata.currentSchemaId()) { + // pointer-only schema rollback keeps the base; an abandoned-timeline base + // (snapshot entry mismatching the live snapshot) is rebuilt + IcebergSnapshot baseCurrent = baseMetadata.currentSnapshot(); + SnapshotManager snapshotManager = table.snapshotManager(); + boolean pointerRollbackOnly = + baseCurrent != null + && snapshotManager.snapshotExists(snapshotId - 1) + && baseCurrent.schemaId() + == (int) snapshotManager.snapshot(snapshotId - 1).schemaId(); + if (!pointerRollbackOnly) { + createMetadataWithoutBase( + snapshot.id(), + baseMetadata.tableUuid(), + Math.max(lastColumnIdFloor, baseMetadata.lastColumnId())); + return; + } + } + List baseManifestFileMetas = manifestList.read(baseMetadata.currentSnapshot().manifestList()); @@ -749,22 +1093,19 @@ private void createMetadataWithBase( computeSnapshotSummary(operation, snapshot, metrics); // add new schemas if needed - SchemaCache schemaCache = new SchemaCache(); - int schemaId = (int) schemaCache.getLatestSchemaId(); - IcebergSchema icebergSchema = schemaCache.get(schemaId); List schemas = baseMetadata.schemas(); - if (baseMetadata.currentSchemaId() != schemaId) { - Preconditions.checkArgument( - schemaId > baseMetadata.currentSchemaId(), - "currentSchemaId{%s} in paimon should be greater than currentSchemaId{%s} in base metadata.", - schemaId, - baseMetadata.currentSchemaId()); + if (schemaId > baseMetadata.currentSchemaId()) { + // append only ids the list does not already carry + Set knownSchemaIds = + schemas.stream().map(IcebergSchema::schemaId).collect(Collectors.toSet()); schemas = new ArrayList<>(schemas); schemas.addAll( IntStream.rangeClosed(baseMetadata.currentSchemaId() + 1, schemaId) + .filter(id -> !knownSchemaIds.contains(id)) .mapToObj(schemaCache::get) .collect(Collectors.toList())); } + // a schema-pointer rollback (validated above): only the current pointer moves List snapshots = new ArrayList<>(baseMetadata.snapshots()); snapshots.add( @@ -772,10 +1113,12 @@ private void createMetadataWithBase( snapshotId, snapshotId, snapshotId - 1, - System.currentTimeMillis(), + // the Paimon snapshot's own commit time, the as-of time readers see + snapshot.timeMillis(), snapshotSummary, pathFactory.toManifestListPath(manifestListFileName).toString(), - schemaId, + // the snapshot's own schema, for time travel + snapshotSchemaId, null, null)); @@ -809,7 +1152,13 @@ private void createMetadataWithBase( baseMetadata.tableUuid(), baseMetadata.location(), snapshotId, - icebergSchema.highestFieldId(), + // must not regress when the current schema is older than the base's + // never below what the replaced metadata already handed out + Math.max( + lastColumnIdFloor, + Math.max( + baseMetadata.lastColumnId(), + icebergSchema.highestFieldId())), schemas, schemaId, baseMetadata.partitionSpecs(), @@ -819,30 +1168,34 @@ private void createMetadataWithBase( refs); Path metadataPath = pathFactory.toMetadataPath(snapshotId); - table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); - table.fileIO() - .overwriteFileUtf8( - new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), - String.valueOf(snapshotId)); - - deleteApplicableMetadataFiles(snapshotId); - for (int i = 0; i + 1 < toExpireExceptLast.size(); i++) { - expireManifestList( - new Path(toExpireExceptLast.get(i).manifestList()).getName(), - new Path(toExpireExceptLast.get(i + 1).manifestList()).getName()); + // atomic-first: see the no-base path + boolean written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); + if (!written + && table.fileIO().exists(metadataPath) + && !metadataMatchesSnapshot(snapshotId, snapshot)) { + table.fileIO().deleteQuietly(metadataPath); + written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); } - - if (metadataCommitter != null) { - switch (metadataCommitter.identifier()) { - case "hive": - metadataCommitter.commitMetadata(metadataPath, baseMetadataPath); - break; - case "rest": - metadataCommitter.commitMetadata(metadata, baseMetadata); - break; - default: - throw new UnsupportedOperationException( - "Unsupported metadata committer: " + metadataCommitter.identifier()); + if (!written && !metadataMatchesSnapshot(snapshotId, snapshot)) { + // no twin published this snapshot's metadata; fail so the commit retries + 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 + Long latestAtPublish = table.snapshotManager().latestSnapshotId(); + if (latestAtPublish != null && latestAtPublish == snapshotId) { + table.fileIO() + .overwriteFileUtf8( + new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), + String.valueOf(snapshotId)); + commitToExternalCatalog(metadata, metadataPath, baseMetadata, baseMetadataPath); + // cleanup only after the catalog serves the new head: a skipped or failed + // publication must not delete files an external pointer still references + deleteApplicableMetadataFiles(snapshotId); + for (int i = 0; i + 1 < toExpireExceptLast.size(); i++) { + expireManifestList( + new Path(toExpireExceptLast.get(i).manifestList()).getName(), + new Path(toExpireExceptLast.get(i + 1).manifestList()).getName()); } } } @@ -1439,6 +1792,24 @@ private static class SummaryMetrics { long totalEqualityDeletes; } + /** + * Summary entry identifying the Paimon snapshot this metadata was built from; it tells live + * metadata from metadata a rollback abandoned. + */ + static final String SNAPSHOT_SUMMARY_PAIMON_COMMIT_IDENTITY = "paimon-commit-identity"; + + private static String commitIdentity(Snapshot snapshot) { + // snapshot uuid when present; legacy snapshots fall back to user/identifier/time + if (snapshot.uuid() != null) { + return snapshot.uuid(); + } + return snapshot.commitUser() + + ":" + + snapshot.commitIdentifier() + + ":" + + snapshot.timeMillis(); + } + private IcebergSnapshotSummary computeSnapshotSummary( String operation, Snapshot snapshot, SummaryMetrics metrics) { @@ -1481,6 +1852,8 @@ private IcebergSnapshotSummary computeSnapshotSummary( } }); } + // after the user-property copy, so a same-key property cannot overwrite it + summary.put(SNAPSHOT_SUMMARY_PAIMON_COMMIT_IDENTITY, commitIdentity(snapshot)); return summary; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index cf8388fc55ee..2f893d4e3b84 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -24,6 +24,7 @@ import org.apache.paimon.consumer.ConsumerManager; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; +import org.apache.paimon.iceberg.IcebergCommitCallback; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFileMeta; @@ -518,6 +519,7 @@ private Optional tryTimeTravel(Options options) { @Override public void rollbackTo(long snapshotId) { + IcebergCommitCallback.markRetirePendingForRollback(this); SnapshotManager snapshotManager = snapshotManager(); try { snapshotManager.rollback(Instant.snapshot(snapshotId)); @@ -543,6 +545,7 @@ public void rollbackTo(long snapshotId) { @Override public void rollbackTo(String tagName) { + IcebergCommitCallback.markRetirePendingForRollback(this); SnapshotManager snapshotManager = snapshotManager(); try { snapshotManager.rollback(Instant.tag(tagName)); diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index c8a1b3fa4c99..d8a22aaee494 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -40,6 +40,10 @@ import org.apache.paimon.iceberg.manifest.IcebergManifestList; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergRef; +import org.apache.paimon.iceberg.metadata.IcebergSchema; +import org.apache.paimon.iceberg.metadata.IcebergSnapshot; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.options.ExpireConfig; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.schema.Schema; @@ -56,6 +60,8 @@ import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine; + import org.apache.avro.Schema.Field; import org.apache.avro.Schema.Type; import org.apache.avro.file.DataFileReader; @@ -456,6 +462,420 @@ public void testRetryCreateMetadata() throws Exception { commit.close(); } + @Test + public void testCommitAfterRollbackDoesNotDuplicateSchemas() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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)); + table.createTag("before-evolution", 1); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + table = table.copyWithLatestSchema(); + write = table.newWrite(commitUser); + commit = table.newCommit(commitUser); + write.write(GenericRow.of(2, 20, 200)); + commit.commit(2, write.prepareCommit(false, 2)); + write.close(); + commit.close(); + + TableCommitImpl rollbackCommit = table.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(table.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + + write = table.newWrite(commitUser); + commit = table.newCommit(commitUser); + write.write(GenericRow.of(3, 30, 300)); + commit.commit(3, write.prepareCommit(false, 3)); + write.close(); + commit.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergMetadata metadata = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + List schemaIds = + metadata.schemas().stream() + .map(IcebergSchema::schemaId) + .collect(Collectors.toList()); + assertThat(schemaIds).doesNotHaveDuplicates(); + assertThat(metadata.currentSchemaId()) + .isEqualTo((int) table.snapshotManager().snapshot(latestId).schemaId()); + } + + @Test + public void testRollbackSnapshotRecordsItsOwnSchema() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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)); + table.createTag("before-evolution", 1); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + table = table.copyWithLatestSchema(); + write = table.newWrite(commitUser); + commit = table.newCommit(commitUser); + write.write(GenericRow.of(2, 20, 200)); + commit.commit(2, write.prepareCommit(false, 2)); + int evolvedSchemaId = (int) table.snapshotManager().snapshot(2).schemaId(); + int evolvedLastColumnId = + IcebergMetadata.fromPath( + table.fileIO(), + new IcebergPathFactory(new Path(table.location(), "metadata")) + .toMetadataPath(2)) + .lastColumnId(); + write.close(); + commit.close(); + + TableCommitImpl rollbackCommit = table.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(table.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + long rolledBackId = table.snapshotManager().latestSnapshotId(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergMetadata rebuilt = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(rolledBackId)); + assertThat(rebuilt.currentSchemaId()).isEqualTo(evolvedSchemaId); + assertThat(rebuilt.currentSnapshot().schemaId()) + .isEqualTo((int) table.snapshotManager().snapshot(rolledBackId).schemaId()); + assertThat(rebuilt.currentSnapshot().schemaId()).isLessThan(evolvedSchemaId); + assertThat(rebuilt.lastColumnId()).isGreaterThanOrEqualTo(evolvedLastColumnId); + } + + @Test + public void testBaseLessRebuildRecordsRollbackSnapshotSchema() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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)); + table.createTag("before-evolution", 1); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + FileStoreTable evolved = table.copyWithLatestSchema(); + TableWriteImpl write2 = evolved.newWrite(commitUser); + TableCommitImpl commit2 = evolved.newCommit(commitUser); + write2.write(GenericRow.of(2, 20, 200)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + for (org.apache.paimon.fs.FileStatus st : + table.fileIO().listStatus(pathFactory.metadataDirectory())) { + if (st.getPath().getName().endsWith(".metadata.json")) { + table.fileIO().deleteQuietly(st.getPath()); + } + } + + TableCommitImpl rollbackCommit = evolved.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(evolved.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergMetadata rebuilt = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + assertThat(rebuilt.currentSchemaId()).isEqualTo(1); + assertThat(rebuilt.currentSnapshot().schemaId()) + .isEqualTo((int) table.snapshotManager().snapshot(latestId).schemaId()); + assertThat(rebuilt.schemas()).anyMatch(s -> s.schemaId() == 1); + assertThat(rebuilt.lastColumnId()) + .isEqualTo( + rebuilt.schemas().stream() + .mapToInt(IcebergSchema::highestFieldId) + .max() + .getAsInt()); + } + + @Test + public void testBaseLessRebuildKeepsLastColumnIdAboveDroppedFields() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + schemaManager.commitChanges(SchemaChange.dropColumn("w")); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + for (org.apache.paimon.fs.FileStatus st : + table.fileIO().listStatus(pathFactory.metadataDirectory())) { + if (st.getPath().getName().endsWith(".metadata.json")) { + table.fileIO().deleteQuietly(st.getPath()); + } + } + + FileStoreTable latest = table.copyWithLatestSchema(); + TableWriteImpl write2 = latest.newWrite(commitUser); + TableCommitImpl commit2 = latest.newCommit(commitUser); + write2.write(GenericRow.of(2, 20)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergMetadata rebuilt = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + int maxFieldId = + rebuilt.schemas().stream().mapToInt(IcebergSchema::highestFieldId).max().getAsInt(); + assertThat(rebuilt.lastColumnId()).isGreaterThanOrEqualTo(maxFieldId); + } + + @Test + public void testSchemaPointerRollbackKeepsHistory() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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)); + table.createTag("before-evolution", 1); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + FileStoreTable evolved = table.copyWithLatestSchema(); + TableWriteImpl write2 = evolved.newWrite(commitUser); + TableCommitImpl commit2 = evolved.newCommit(commitUser); + write2.write(GenericRow.of(2, 20, 200)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + TableCommitImpl rollbackCommit = evolved.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(evolved.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + table.deleteTag("before-evolution"); + table.newExpireSnapshots() + .config(ExpireConfig.builder().snapshotRetainMax(1).snapshotRetainMin(1).build()) + .expire(); + schemaManager.rollbackTo( + 0, table.snapshotManager(), table.tagManager(), table.changelogManager()); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + long rollbackId = table.snapshotManager().latestSnapshotId(); + String uuidBefore = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(rollbackId)) + .tableUuid(); + + TableWriteImpl write3 = table.newWrite(commitUser); + TableCommitImpl commit3 = table.newCommit(commitUser); + write3.write(GenericRow.of(3, 30)); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergMetadata metadata = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + assertThat(metadata.tableUuid()).isEqualTo(uuidBefore); + assertThat(metadata.currentSchemaId()).isEqualTo(0); + assertThat(metadata.snapshots().stream().map(IcebergSnapshot::snapshotId)) + .contains(rollbackId, latestId); + } + + @Test + public void testSchemaRollbackWithAbandonedBaseRebuildsMetadata() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + FileStoreTable evolved = table.copyWithLatestSchema(); + TableWriteImpl write2 = evolved.newWrite(commitUser); + TableCommitImpl commit2 = evolved.newCommit(commitUser); + write2.write(GenericRow.of(2, 20, 200)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + table.rollbackTo(1); + schemaManager.rollbackTo( + 0, table.snapshotManager(), table.tagManager(), table.changelogManager()); + + TableWriteImpl write3 = table.newWrite(commitUser); + TableCommitImpl commit3 = table.newCommit(commitUser); + write3.write(GenericRow.of(3, 30)); + commit3.commit(2, write3.prepareCommit(false, 2)); + write3.write(GenericRow.of(4, 40)); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + + assertThat(getIcebergResult()) + .containsExactlyInAnyOrder("Record(1, 10)", "Record(3, 30)", "Record(4, 40)"); + } + + @Test + public void testReusedSchemaIdAfterSchemaRollbackRebuildsMetadata() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + FileStoreTable evolved = table.copyWithLatestSchema(); + TableWriteImpl write2 = evolved.newWrite(commitUser); + TableCommitImpl commit2 = evolved.newCommit(commitUser); + write2.write(GenericRow.of(2, 20, 200)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + table.rollbackTo(1); + schemaManager.rollbackTo( + 0, table.snapshotManager(), table.tagManager(), table.changelogManager()); + schemaManager.commitChanges(SchemaChange.addColumn("x", DataTypes.STRING())); + + FileStoreTable reEvolved = table.copyWithLatestSchema(); + TableWriteImpl write3 = reEvolved.newWrite(commitUser); + TableCommitImpl commit3 = reEvolved.newCommit(commitUser); + write3.write(GenericRow.of(3, 30, BinaryString.fromString("three"))); + commit3.commit(2, write3.prepareCommit(false, 2)); + write3.write(GenericRow.of(4, 40, BinaryString.fromString("four"))); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergMetadata metadata = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + assertThat( + metadata.schemas().stream() + .filter(sch -> sch.schemaId() == 1) + .flatMap(sch -> sch.fields().stream()) + .map(f -> f.name())) + .contains("x") + .doesNotContain("w"); + } + + @Test + public void testRebuildAfterSchemaRollbackKeepsLastColumnIdHighWaterMark() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + schemaManager.commitChanges(SchemaChange.addColumn("y", DataTypes.INT())); + FileStoreTable evolved = table.copyWithLatestSchema(); + TableWriteImpl write2 = evolved.newWrite(commitUser); + TableCommitImpl commit2 = evolved.newCommit(commitUser); + write2.write(GenericRow.of(2, 20, 200, 2000)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + int oldLastColumnId = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(2)) + .lastColumnId(); + + table.rollbackTo(1); + schemaManager.rollbackTo( + 0, table.snapshotManager(), table.tagManager(), table.changelogManager()); + schemaManager.commitChanges(SchemaChange.addColumn("x", DataTypes.STRING())); + + FileStoreTable reEvolved = table.copyWithLatestSchema(); + TableWriteImpl write3 = reEvolved.newWrite(commitUser); + TableCommitImpl commit3 = reEvolved.newCommit(commitUser); + write3.write(GenericRow.of(3, 30, BinaryString.fromString("three"))); + commit3.commit(2, write3.prepareCommit(false, 2)); + write3.write(GenericRow.of(4, 40, BinaryString.fromString("four"))); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + + long latestId = table.snapshotManager().latestSnapshotId(); + IcebergMetadata rebuilt = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(latestId)); + assertThat(rebuilt.lastColumnId()).isGreaterThanOrEqualTo(oldLastColumnId); + } + @Test public void testSchemaChange() throws Exception { RowType rowType = @@ -1685,6 +2105,437 @@ private TestRecord(BinaryRow partition, GenericRow record) { } } + @Test + public void testDeepRollbackRetiresAbandonedSuffixWhenReusedVersionExpired() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + for (int i = 1; i <= 4; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(2))).isFalse(); + String uuidBefore = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(4)).tableUuid(); + + table.rollbackTo(1); + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(9, 90)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(3))).isFalse(); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(4))).isFalse(); + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(9, 90)"); + long latestId = table.snapshotManager().latestSnapshotId(); + assertThat( + IcebergMetadata.fromPath( + table.fileIO(), pathFactory.toMetadataPath(latestId)) + .tableUuid()) + .isEqualTo(uuidBefore); + } + + @Test + public void testRetryRecommitsMetadataToExternalCatalog() throws Exception { + RecordingIcebergMetadataCommitter.COMMITS.clear(); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1) + .copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + IcebergOptions.StorageType.HADOOP_CATALOG.toString())); + + 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(); + assertThat(RecordingIcebergMetadataCommitter.COMMITS).isNotEmpty(); + + RecordingIcebergMetadataCommitter.COMMITS.clear(); + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new ManifestCommittable(2)); + callback.close(); + assertThat(RecordingIcebergMetadataCommitter.COMMITS).isNotEmpty(); + } + + @Test + public void testRetryOfOldSnapshotDoesNotMoveCatalogPointer() throws Exception { + RecordingIcebergMetadataCommitter.COMMITS.clear(); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1) + .copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + IcebergOptions.StorageType.HADOOP_CATALOG.toString())); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + for (int i = 1; i <= 3; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + assertThat(RecordingIcebergMetadataCommitter.COMMITS).isNotEmpty(); + + RecordingIcebergMetadataCommitter.COMMITS.clear(); + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new ManifestCommittable(2)); + callback.close(); + assertThat(RecordingIcebergMetadataCommitter.COMMITS).isEmpty(); + } + + @Test + public void testDeepRollbackRetiresSuffixWhenVersionHintLagsBehind() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + for (int i = 1; i <= 4; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(2))).isFalse(); + String uuidBefore = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(4)).tableUuid(); + + table.rollbackTo(1); + table.fileIO() + .overwriteFileUtf8( + new Path(pathFactory.metadataDirectory(), "version-hint.text"), "1"); + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(9, 90)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(3))).isFalse(); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(4))).isFalse(); + long latestId = table.snapshotManager().latestSnapshotId(); + assertThat( + IcebergMetadata.fromPath( + table.fileIO(), pathFactory.toMetadataPath(latestId)) + .tableUuid()) + .isEqualTo(uuidBefore); + } + + @Test + public void testDelayedReplayOfReusedSnapshotDoesNotMoveHeadBack() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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(); + + table.rollbackTo(1); + + FileStoreTable disabled = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "disabled")); + TableWriteImpl write2 = disabled.newWrite(commitUser); + TableCommitImpl commit2 = disabled.newCommit(commitUser); + write2.write(GenericRow.of(3, 30)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + TableWriteImpl write3 = table.newWrite(commitUser); + TableCommitImpl commit3 = table.newCommit(commitUser); + write3.write(GenericRow.of(4, 40)); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + String hintBefore = + table.fileIO() + .readFileUtf8( + new Path(pathFactory.metadataDirectory(), "version-hint.text")) + .trim(); + assertThat(hintBefore).isEqualTo("3"); + + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new ManifestCommittable(2)); + callback.close(); + + assertThat( + table.fileIO() + .readFileUtf8( + new Path( + pathFactory.metadataDirectory(), + "version-hint.text")) + .trim()) + .isEqualTo("3"); + } + + @Test + public void testRetryAfterCatalogFailureKeepsItsBase() throws Exception { + RecordingIcebergMetadataCommitter.COMMITS.clear(); + RecordingIcebergMetadataCommitter.BASES.clear(); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map overrides = new HashMap<>(); + overrides.put( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), + IcebergOptions.StorageType.HADOOP_CATALOG.toString()); + overrides.put(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX.key(), "0"); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1) + .copy(overrides); + + 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)); + + RecordingIcebergMetadataCommitter.failNextCommit = true; + write.write(GenericRow.of(2, 20)); + assertThatThrownBy(() -> commit.commit(2, write.prepareCommit(false, 2))) + .hasStackTraceContaining("injected catalog failure"); + write.close(); + try { + commit.close(); + } catch (Exception ignored) { + } + + RecordingIcebergMetadataCommitter.BASES.clear(); + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new ManifestCommittable(2)); + callback.close(); + assertThat(RecordingIcebergMetadataCommitter.BASES).hasSize(1); + assertThat(RecordingIcebergMetadataCommitter.BASES.get(0)).isNotNull(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(IcebergCommitCallback.catalogTableMetadataPath(table)); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(1))).isFalse(); + } + + @Test + public void testDeepRollbackRecoversUuidWhenVersionHintMissing() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + for (int i = 1; i <= 4; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(2))).isFalse(); + String uuidBefore = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(4)).tableUuid(); + + table.rollbackTo(1); + table.fileIO() + .deleteQuietly(new Path(pathFactory.metadataDirectory(), "version-hint.text")); + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(9, 90)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(3))).isFalse(); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(4))).isFalse(); + long latestId = table.snapshotManager().latestSnapshotId(); + assertThat( + IcebergMetadata.fromPath( + table.fileIO(), pathFactory.toMetadataPath(latestId)) + .tableUuid()) + .isEqualTo(uuidBefore); + } + + @Test + public void testRecommitAfterRollbackIgnoresStaleSnapshotCache() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable stale = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + stale.setSnapshotCache(Caffeine.newBuilder().maximumSize(128).build()); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = stale.newWrite(commitUser); + TableCommitImpl commit = stale.newCommit(commitUser); + for (int i = 1; i <= 4; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + stale.snapshotManager().snapshot(2); + + try (FileSystemCatalog freshCatalog = + new FileSystemCatalog(LocalFileIO.create(), new Path(tempDir.toString()))) { + FileStoreTable fresh = + (FileStoreTable) freshCatalog.getTable(Identifier.create("mydb", "t")); + fresh.rollbackTo(1); + } + + TableWriteImpl write2 = stale.newWrite(commitUser); + TableCommitImpl commit2 = stale.newCommit(commitUser); + write2.write(GenericRow.of(9, 90)); + commit2.commit(5, write2.prepareCommit(false, 5)); + write2.close(); + commit2.close(); + + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(9, 90)"); + } + + @Test + public void testRecommitAfterRollbackReplacesStaleMetadata() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1); + + 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(); + + TableWriteImpl writeExtra = table.newWrite(commitUser); + TableCommitImpl commitExtra = table.newCommit(commitUser); + writeExtra.write(GenericRow.of(9, 90)); + commitExtra.commit(3, writeExtra.prepareCommit(false, 3)); + writeExtra.close(); + commitExtra.close(); + + table.rollbackTo(1); + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(3, 30)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + IcebergPathFactory suffixPathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + assertThat(table.fileIO().exists(suffixPathFactory.toMetadataPath(3))).isFalse(); + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10)", "Record(3, 30)"); + + TableWriteImpl write3 = table.newWrite(commitUser); + TableCommitImpl commit3 = table.newCommit(commitUser); + write3.write(GenericRow.of(4, 40)); + commit3.commit(3, write3.prepareCommit(false, 3)); + write3.close(); + commit3.close(); + assertThat(getIcebergResult()) + .containsExactlyInAnyOrder("Record(1, 10)", "Record(3, 30)", "Record(4, 40)"); + } + + @Test + public void testRollbackMarksRetirementAcrossVersionGaps() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable(rowType, Collections.emptyList(), Collections.emptyList(), -1) + .copy( + Collections.singletonMap( + IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX.key(), "0")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + for (int i = 1; i <= 4; i++) { + write.write(GenericRow.of(i, i * 10)); + commit.commit(i, write.prepareCommit(false, i)); + } + write.close(); + commit.close(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(4))).isTrue(); + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(3))).isFalse(); + String uuidBefore = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(4)).tableUuid(); + + table.fileIO() + .overwriteFileUtf8( + new Path(pathFactory.metadataDirectory(), "version-hint.text"), "1"); + table.rollbackTo(1); + + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(9, 90)); + commit2.commit(5, write2.prepareCommit(false, 5)); + write2.close(); + commit2.close(); + + assertThat(table.fileIO().exists(pathFactory.toMetadataPath(4))).isFalse(); + assertThat( + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(2)) + .tableUuid()) + .isEqualTo(uuidBefore); + } + // ------------------------------------------------------------------------ // Utils // ------------------------------------------------------------------------ diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/RecordingIcebergMetadataCommitter.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/RecordingIcebergMetadataCommitter.java new file mode 100644 index 000000000000..f83de7aed588 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/RecordingIcebergMetadataCommitter.java @@ -0,0 +1,78 @@ +/* + * 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.fs.Path; +import org.apache.paimon.iceberg.metadata.IcebergMetadata; +import org.apache.paimon.table.FileStoreTable; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** An {@link IcebergMetadataCommitter} recording every commit, for tests. */ +public class RecordingIcebergMetadataCommitter implements IcebergMetadataCommitter { + + public static final List COMMITS = Collections.synchronizedList(new ArrayList<>()); + public static final List BASES = Collections.synchronizedList(new ArrayList<>()); + public static volatile boolean failNextCommit = false; + + private static void maybeFail() { + if (failNextCommit) { + failNextCommit = false; + throw new RuntimeException("injected catalog failure"); + } + } + + @Override + public String identifier() { + return "hive"; + } + + @Override + public void commitMetadata(Path newMetadataPath, @Nullable Path baseMetadataPath) { + maybeFail(); + COMMITS.add(newMetadataPath); + BASES.add(baseMetadataPath); + } + + @Override + public void commitMetadata( + IcebergMetadata newIcebergMetadata, @Nullable IcebergMetadata baseIcebergMetadata) { + maybeFail(); + COMMITS.add(newIcebergMetadata); + BASES.add(baseIcebergMetadata); + } + + /** Registered under hadoop-catalog: no real committer exists there, so no ambiguity. */ + public static class Factory implements IcebergMetadataCommitterFactory { + + @Override + public String identifier() { + return IcebergOptions.StorageType.HADOOP_CATALOG.toString(); + } + + @Override + public IcebergMetadataCommitter create(FileStoreTable table) { + return new RecordingIcebergMetadataCommitter(); + } + } +} diff --git a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory index 76ffd7a52f73..c48309213e6f 100644 --- a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory +++ b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory @@ -16,3 +16,4 @@ org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory org.apache.paimon.mergetree.compact.aggregate.TestMapOnlyAggFactory org.apache.paimon.rest.auth.CustomTestDLFTokenLoaderFactory +org.apache.paimon.iceberg.RecordingIcebergMetadataCommitter$Factory 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 9dbc79670187..2357a989c9ca 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 @@ -71,6 +71,8 @@ */ public class IcebergRestMetadataCommitter implements IcebergMetadataCommitter { + private static final String PAIMON_COMMIT_IDENTITY = "paimon-commit-identity"; + private static final Logger LOG = LoggerFactory.getLogger(IcebergRestMetadataCommitter.class); private static final String REST_CATALOG_NAME = "rest-catalog"; @@ -160,6 +162,23 @@ private void commitMetadataImpl( TableMetadata metadata = ((BaseTable) icebergTable).operations().current(); + org.apache.iceberg.Snapshot catalogCurrent = metadata.currentSnapshot(); + org.apache.iceberg.Snapshot newCurrent = newMetadata.currentSnapshot(); + if (catalogCurrent != null + && newCurrent != null + && catalogCurrent.snapshotId() == newCurrent.snapshotId() + && java.util.Objects.equals( + catalogCurrent.summary().get(PAIMON_COMMIT_IDENTITY), + newCurrent.summary().get(PAIMON_COMMIT_IDENTITY))) { + // an idempotent retry: the catalog is already at this snapshot; rebuilding + // through updatesForIncorrectBase would drop and recreate the table + LOG.info( + "Iceberg table {} is already at snapshot {}, nothing to commit.", + icebergTableIdentifier, + newCurrent.snapshotId()); + return; + } + if (metadata.currentSnapshot() == null) { // Table exists in the REST catalog but has no snapshots yet. This happens // when a previous createTable() or recreateTable() succeeded but the @@ -229,19 +248,16 @@ private TableMetadata.Builder updatesForCorrectBase( } else { // add new schema if needed - Preconditions.checkArgument( - newMetadata.currentSchemaId() >= schemaId, - "the new metadata has correct base, but the schemaId(%s) in iceberg table " - + "is greater than currentSchemaId(%s) in new metadata.", - schemaId, - newMetadata.currentSchemaId()); - if (newMetadata.currentSchemaId() != schemaId) { + if (newMetadata.currentSchemaId() > schemaId) { addAndSetCurrentSchema( newMetadata.schemas().stream() .filter(schema -> schema.schemaId() > schemaId) .collect(Collectors.toList()), newMetadata.currentSchemaId(), updateBuilder); + } else if (newMetadata.currentSchemaId() < schemaId) { + // a rollback moved the current schema back; only the pointer moves + updateBuilder.setCurrentSchema(newMetadata.currentSchemaId()); } // add snapshot @@ -468,6 +484,21 @@ private static boolean checkBase( return false; } + // the same numeric id can belong to a rolled-back timeline; extending from it would + // keep the abandoned history in the catalog + IcebergSnapshot baseCurrent = baseIcebergMetadata.currentSnapshot(); + if (baseCurrent != null + && currentMetadata.currentSnapshot().snapshotId() == baseCurrent.snapshotId()) { + String catalogIdentity = + currentMetadata.currentSnapshot().summary().get(PAIMON_COMMIT_IDENTITY); + String baseIdentity = baseCurrent.summary().get(PAIMON_COMMIT_IDENTITY); + if (catalogIdentity != null + && baseIdentity != null + && !catalogIdentity.equals(baseIdentity)) { + return false; + } + } + // if the iceberg table is existed, check whether the current metadata of the table is the // base of the new table metadata, we use current snapshot id to check. // Note: callers must ensure currentMetadata.currentSnapshot() is non-null before calling @@ -603,7 +634,15 @@ private IcebergMetadata adjustMetadataForRest(IcebergMetadata newIcebergMetadata snapshot.sequenceNumber(), snapshot.snapshotId(), snapshot.parentSnapshotId(), - snapshot.timestampMs(), + // a slow rebuild must not trip Iceberg's + // one-minute update-timestamp window + snapshot.snapshotId() + == newIcebergMetadata + .currentSnapshotId() + ? Math.max( + snapshot.timestampMs(), + System.currentTimeMillis() - 59_000L) + : snapshot.timestampMs(), snapshot.summary(), snapshot.manifestList(), remappedSchemaId, 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 0fbf130a7365..052f76bda7ca 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 @@ -415,6 +415,85 @@ public void testSchemaAndPropertiesChange() throws Exception { commit.close(); } + @Test + public void testCommitAfterSchemaRollback() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + randomFormat(), + 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)); + table.createTag("before-evolution", 1); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("v2", DataTypes.STRING())); + table = table.copy(table.schemaManager().latest().get()); + write.close(); + write = table.newWrite(commitUser); + commit.close(); + commit = table.newCommit(commitUser); + write.write(GenericRow.of(2, 20, BinaryString.fromString("two"))); + commit.commit(2, write.prepareCommit(false, 2)); + write.close(); + commit.close(); + + TableCommitImpl rollbackCommit = table.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(table.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(1, 10, null)"); + Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + System.out.println( + "PROBE catalog currentSchemaId=" + + ((org.apache.iceberg.BaseTable) icebergTable) + .operations() + .current() + .currentSchemaId() + + " schemas=" + + icebergTable.schemas().keySet() + + " snapSchemaId=" + + icebergTable.currentSnapshot().schemaId()); + long localLatest = table.snapshotManager().latestSnapshotId(); + org.apache.paimon.iceberg.metadata.IcebergMetadata localMeta = + org.apache.paimon.iceberg.metadata.IcebergMetadata.fromPath( + table.fileIO(), + new org.apache.paimon.fs.Path( + table.location().getParent().getParent(), + "iceberg/mydb/t/metadata/v" + localLatest + ".metadata.json")); + System.out.println( + "PROBE local current=" + + localMeta.currentSchemaId() + + " snapEntry=" + + localMeta.currentSnapshot().schemaId()); + System.out.println( + "PROBE schemaLatest=" + + new SchemaManager(table.fileIO(), table.location()).latest().get().id()); + for (org.apache.paimon.fs.FileStatus st : + table.fileIO() + .listStatus( + new org.apache.paimon.fs.Path( + table.location().getParent().getParent(), + "iceberg/mydb/t/metadata"))) { + if (st.getPath().getName().endsWith(".metadata.json")) { + System.out.println("PROBE json: " + st.getPath().getName()); + } + } + assertThat(icebergTable.schema().columns().stream().map(c -> c.name())) + .containsExactly("k", "v", "v2"); + assertThat(icebergTable.currentSnapshot().schemaId()).isEqualTo(1); + } + @Test public void testOptionOnlyAlterTableDoesNotCrashIcebergSync() throws Exception { // The fix deduplicates schemas in adjustMetadataForRest() and remaps @@ -751,6 +830,158 @@ public void testIcebergSnapshotExpire() throws Exception { .containsExactlyInAnyOrder("Record(1, 11)", "Record(2, 20)", "Record(3, 30)"); } + @Test + public void testRetryAdvancesLaggingCatalogWithoutRecreate() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + randomFormat(), + 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.close(); + commit.close(); + + FileStoreTable localOnly = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "hadoop-catalog")); + TableWriteImpl write2 = localOnly.newWrite(commitUser); + TableCommitImpl commit2 = localOnly.newCommit(commitUser); + write2.write(GenericRow.of(2, 20)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(1); + icebergTable.updateProperties().set("user-custom-prop", "kept").commit(); + String uuidBefore = ((BaseTable) icebergTable).operations().current().uuid(); + + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(2)); + callback.close(); + + Table reloaded = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + assertThat(((BaseTable) reloaded).operations().current().uuid()).isEqualTo(uuidBefore); + assertThat(reloaded.properties()).containsEntry("user-custom-prop", "kept"); + assertThat(reloaded.currentSnapshot().snapshotId()).isEqualTo(2); + } + + @Test + public void testAbandonedBaseWithSameIdIsNotExtendedInCatalog() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + randomFormat(), + 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(); + + table.rollbackTo(1); + FileStoreTable localOnly = + table.copy( + Collections.singletonMap( + IcebergOptions.METADATA_ICEBERG_STORAGE.key(), "hadoop-catalog")); + TableWriteImpl write2 = localOnly.newWrite(commitUser); + TableCommitImpl commit2 = localOnly.newCommit(commitUser); + write2.write(GenericRow.of(3, 30)); + commit2.commit(3, write2.prepareCommit(false, 3)); + write2.close(); + commit2.close(); + + TableWriteImpl write3 = table.newWrite(commitUser); + TableCommitImpl commit3 = table.newCommit(commitUser); + write3.write(GenericRow.of(4, 40)); + commit3.commit(4, write3.prepareCommit(false, 4)); + write3.close(); + commit3.close(); + + assertThat(getIcebergResult()) + .containsExactlyInAnyOrder("Record(1, 10)", "Record(3, 30)", "Record(4, 40)"); + + org.apache.paimon.iceberg.metadata.IcebergMetadata localMetadata = + org.apache.paimon.iceberg.metadata.IcebergMetadata.fromPath( + table.fileIO(), + new org.apache.paimon.fs.Path( + IcebergCommitCallback.catalogTableMetadataPath(table), + "v3.metadata.json")); + String localIdentity = + localMetadata.snapshots().stream() + .filter(snap -> snap.snapshotId() == 2) + .findFirst() + .get() + .summary() + .get("paimon-commit-identity"); + Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + org.apache.iceberg.Snapshot catalogSnapshot2 = icebergTable.snapshot(2); + if (catalogSnapshot2 != null) { + assertThat(catalogSnapshot2.summary().get("paimon-commit-identity")) + .isEqualTo(localIdentity); + } + } + + @Test + public void testIdempotentRetryDoesNotRecreateTable() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + randomFormat(), + 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(); + + Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + icebergTable.updateProperties().set("user-custom-prop", "kept").commit(); + String uuidBefore = ((BaseTable) icebergTable).operations().current().uuid(); + + IcebergCommitCallback callback = new IcebergCommitCallback(table, commitUser); + callback.retry(new org.apache.paimon.manifest.ManifestCommittable(2)); + callback.close(); + + Table reloaded = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); + assertThat(((BaseTable) reloaded).operations().current().uuid()).isEqualTo(uuidBefore); + assertThat(reloaded.properties()).containsEntry("user-custom-prop", "kept"); + } + @Test public void testWithIncorrectBase() throws Exception { RowType rowType =