This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 79a8b64473 [iceberg] Fix Iceberg metadata after a rollback (#9036)
79a8b64473 is described below
commit 79a8b64473f63d28f6fe9ea2c0c8b7b2c1e0900c
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Aug 10 16:40:07 2026 +0800
[iceberg] Fix Iceberg metadata after a rollback (#9036)
---
.../paimon/iceberg/IcebergCommitCallback.java | 501 ++++++++++--
.../paimon/table/AbstractFileStoreTable.java | 3 +
.../paimon/iceberg/IcebergCompatibilityTest.java | 851 +++++++++++++++++++++
.../iceberg/RecordingIcebergMetadataCommitter.java | 78 ++
.../services/org.apache.paimon.factories.Factory | 1 +
.../iceberg/IcebergRestMetadataCommitter.java | 55 +-
.../iceberg/IcebergRestMetadataCommitterTest.java | 231 ++++++
7 files changed, 1648 insertions(+), 72 deletions(-)
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 336b0fe865..4b6776d339 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.GenericArray;
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 org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
@@ -296,6 +298,23 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
List<IndexManifestEntry> 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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
.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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
//
-------------------------------------------------------------------------------------
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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<IcebergPartitionField> partitionFields =
getPartitionFields(table.schema().partitionKeys(),
icebergSchema);
@@ -410,10 +528,11 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
// After https://github.com/apache/paimon/issues/6107 we can add tags
here.
Map<String, IcebergRef> 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<IcebergSchema> allSchemas =
IntStream.rangeClosed(0, schemaId)
@@ -437,7 +558,14 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
// 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.
+ *
+ * <p>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<IndexManifestEntry> 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<IcebergManifestFileMeta> baseManifestFileMetas =
manifestList.read(baseMetadata.currentSnapshot().manifestList());
@@ -749,22 +1093,19 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
computeSnapshotSummary(operation, snapshot, metrics);
// add new schemas if needed
- SchemaCache schemaCache = new SchemaCache();
- int schemaId = (int) schemaCache.getLatestSchemaId();
- IcebergSchema icebergSchema = schemaCache.get(schemaId);
List<IcebergSchema> 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<Integer> 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<IcebergSnapshot> snapshots = new
ArrayList<>(baseMetadata.snapshots());
snapshots.add(
@@ -772,10 +1113,12 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
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 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
}
});
}
+ // 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 cf8388fc55..2f893d4e3b 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.catalog.Identifier;
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 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@Override
public void rollbackTo(long snapshotId) {
+ IcebergCommitCallback.markRetirePendingForRollback(this);
SnapshotManager snapshotManager = snapshotManager();
try {
snapshotManager.rollback(Instant.snapshot(snapshotId));
@@ -543,6 +545,7 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
@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 c8a1b3fa4c..d8a22aaee4 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.IcebergManifestFileMeta;
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.DataTypes;
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 class IcebergCompatibilityTest {
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<Integer> 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 @@ public class IcebergCompatibilityTest {
}
}
+ @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<String, String> 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 0000000000..f83de7aed5
--- /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<Object> COMMITS =
Collections.synchronizedList(new ArrayList<>());
+ public static final List<Object> 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 76ffd7a52f..c48309213e 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 9dbc796701..2357a989c9 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 @@ import static
org.apache.iceberg.TableProperties.METADATA_PREVIOUS_VERSIONS_MAX;
*/
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 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
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 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
} 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 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
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 @@ public class IcebergRestMetadataCommitter implements
IcebergMetadataCommitter {
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 0fbf130a73..052f76bda7 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 class IcebergRestMetadataCommitterTest {
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 class IcebergRestMetadataCommitterTest {
.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 =
