vbabenkoru commented on code in PR #9348:
URL: https://github.com/apache/paimon/pull/9348#discussion_r4435759312
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -459,6 +462,397 @@ private void createMetadata(
}
}
+ /**
+ * Create Iceberg metadata when no usable base metadata exists: either the
very first Iceberg
+ * commit for this table, or a recovery after the previous metadata became
unusable (format
+ * version change, missing row lineage, Iceberg-layer commit failure).
+ *
+ * <p>By default only the current snapshot is exposed to Iceberg. With
{@link
+ * IcebergOptions#SYNC_FULL_HISTORY} the whole retained Paimon history is
replayed instead, so
+ * Iceberg readers keep time travel and tags (see <a
+ *
href="https://github.com/apache/paimon/issues/6107";>apache/paimon#6107</a>).
+ */
+ private void recreateMetadata(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ if (syncFullHistory) {
+ rebuildFullHistory(snapshotId, inheritUuid, lastColumnIdFloor,
nextRowIdFloor);
+ } else {
+ createMetadataWithoutBase(snapshotId, inheritUuid,
lastColumnIdFloor, nextRowIdFloor);
+ }
+ }
+
+ /**
+ * Rebuild Iceberg metadata from every Paimon snapshot that is still
retained, ending at {@code
+ * currentSnapshotId}: create metadata afresh for the earliest retained
snapshot, then replay
+ * each following snapshot on top of its predecessor, exactly like live
commits would have.
+ * Schemas, tags and (for format version 3) the row-id space therefore
accumulate consistently
+ * across the whole replayed history.
+ *
+ * <p>Every replay step writes to a uniquely named staging path; the
published chain, the
+ * version hint and the external catalog stay untouched until the final
staged metadata is
+ * durable, and only then is the staged chain promoted into place. A
rebuild that fails at any
+ * point therefore leaves the previously published metadata fully
readable, and its staged
+ * leftovers are removed by the next rebuild. Replayed snapshots keep
their original Paimon
+ * commit timestamps and are subject to the same retention policy ({@link
+ * CoreOptions#SNAPSHOT_NUM_RETAINED_MIN}, {@link
CoreOptions#SNAPSHOT_TIME_RETAINED}, ...) that
+ * live commits apply.
+ */
+ private void rebuildFullHistory(
+ long currentSnapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Long earliest = snapshotManager.earliestSnapshotId();
+ long startId = earliest == null ? currentSnapshotId :
Math.min(earliest, currentSnapshotId);
+
+ deleteStagedLeftovers();
+
+ // Resume from the newest existing metadata below the current
snapshot, if it is usable.
+ // Anything older than the newest existing file is stale by
definition: live commits only
+ // ever read the immediately preceding metadata.
+ long baseId = -1;
+ for (long id = currentSnapshotId - 1; id >= startId; id--) {
+ Path metadataPath = pathFactory.toMetadataPath(id);
+ if (table.fileIO().exists(metadataPath)) {
+ try {
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(table.fileIO(),
metadataPath);
+ if (isUsableReplayBase(metadata, id, startId)) {
+ baseId = id;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read existing Iceberg metadata {},
rebuilding history from scratch",
+ metadataPath,
+ e);
+ }
+ break;
+ }
+ }
+
+ String rebuildUuid = UUID.randomUUID().toString();
+ long firstStagedId;
+ boolean freshRebuild = baseId == -1;
+ StaleBuild staleBuild = null;
+ if (freshRebuild) {
+ // No usable base: the whole old build is stale. Nothing of it is
touched during the
+ // replay, so an external catalog that still points at the old
metadata keeps a fully
+ // readable table; the old files are replaced and cleaned only by
the promotion. Their
+ // references are collected up front, tolerating unreadable files
(that is what
+ // triggered some rebuilds in the first place).
+ staleBuild = collectStaleBuild(currentSnapshotId);
+ firstStagedId = startId;
+ createMetadataWithoutBase(
+ startId,
+ inheritUuid,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, startId));
+ } else {
+ firstStagedId = baseId + 1;
+ }
+
+ for (long id = firstStagedId == startId ? startId + 1 : firstStagedId;
+ id <= currentSnapshotId;
+ id++) {
+ long snapshotId = id;
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ Path basePath =
+ snapshotId - 1 < firstStagedId
+ ? pathFactory.toMetadataPath(snapshotId - 1)
+ : stagedMetadataPath(rebuildUuid, snapshotId - 1);
+ createMetadataWithBase(
+ (removedFiles, addedFiles) ->
+ collectFileChanges(snapshotId, removedFiles,
addedFiles),
+ indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX),
+ snapshot,
+ basePath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, snapshotId));
+ }
+
+ boolean promoted =
+ promoteStagedReplay(rebuildUuid, firstStagedId,
currentSnapshotId, startId);
+ if (promoted && staleBuild != null) {
+ deleteStaleBuild(staleBuild, currentSnapshotId, startId);
+ }
+ }
+
+ /**
+ * Whether an existing metadata file can serve as the resume base of a
full-history replay. On
+ * top of the structural checks, the base must describe the live timeline:
an abandoned base (a
+ * rolled-back Paimon snapshot reused the id) or one carrying re-evolved
schema definitions
+ * would be rejected again by the first replay step, and a rebuild must
never select a base its
+ * own replay refuses to extend.
+ */
+ private boolean isUsableReplayBase(IcebergMetadata metadata, long baseId,
long startId) {
+ if (!isSameFormatVersion(metadata.formatVersion())) {
+ return false;
+ }
+ if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3 &&
metadata.nextRowId() == null) {
+ return false;
+ }
+ if (!coversRetainedPrefix(metadata, baseId, startId)) {
+ return false;
+ }
+ SnapshotManager snapshotManager = table.snapshotManager();
+ if (!snapshotManager.snapshotExists(baseId)
+ || !metadataMatchesSnapshot(metadata,
snapshotManager.snapshot(baseId))) {
+ return false;
+ }
+ SchemaCache schemaCache = new SchemaCache();
+ long latestSchemaId = schemaCache.getLatestSchemaId();
+ for (IcebergSchema known : metadata.schemas()) {
+ if (known.schemaId() > latestSchemaId
+ || !known.equals(schemaCache.get(known.schemaId()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private Path stagedMetadataPath(String rebuildUuid, long snapshotId) {
+ return new Path(
+ pathFactory.metadataDirectory(),
+ String.format("rebuild-%s-v%d.metadata.json", rebuildUuid,
snapshotId));
+ }
+
+ /** Staged files of crashed or superseded rebuilds; only ever garbage. */
+ private void deleteStagedLeftovers() throws IOException {
+ FileStatus[] statuses;
+ try {
+ statuses =
table.fileIO().listStatus(pathFactory.metadataDirectory());
+ } catch (FileNotFoundException e) {
+ return;
+ }
+ for (FileStatus status : statuses) {
+ String name = status.getPath().getName();
+ if (name.startsWith("rebuild-") &&
name.endsWith(".metadata.json")) {
+ table.fileIO().deleteQuietly(status.getPath());
+ }
+ }
+ }
+
+ /**
+ * Switch the published chain to the staged one, then the version hint and
the external catalog.
+ * The staged head is durable and validated before any published path is
touched; the remaining
+ * window is the per-file replacement of each chain position, ending one
below the head, and
+ * every path serves complete metadata again as soon as its replacement
lands.
+ *
+ * @return false if a newer commit superseded this rebuild; the staged
files are discarded and
+ * the published chain is left for that commit's own rebuild
+ */
+ private boolean promoteStagedReplay(
+ String rebuildUuid, long firstStagedId, long currentSnapshotId,
long startId)
+ throws IOException {
+ IcebergMetadata head =
+ IcebergMetadata.fromPath(
+ table.fileIO(), stagedMetadataPath(rebuildUuid,
currentSnapshotId));
+ Preconditions.checkState(
+ head.currentSnapshotId() == currentSnapshotId,
+ "Staged replay head is at snapshot %s instead of %s",
+ head.currentSnapshotId(),
+ currentSnapshotId);
+ // a truncated chain must never be promoted, no matter what produced it
+ Preconditions.checkState(
+ coversRetainedPrefix(head, currentSnapshotId, startId),
+ "Staged replay head for snapshot %s does not cover the
retained history",
+ currentSnapshotId);
+
+ Long latest = table.snapshotManager().latestSnapshotId();
+ if (latest == null || latest != currentSnapshotId) {
+ for (long id = firstStagedId; id <= currentSnapshotId; id++) {
+ table.fileIO().deleteQuietly(stagedMetadataPath(rebuildUuid,
id));
+ }
+ return false;
+ }
+
+ for (long id = firstStagedId; id <= currentSnapshotId; id++) {
+ Path staged = stagedMetadataPath(rebuildUuid, id);
+ Path published = pathFactory.toMetadataPath(id);
+ if (table.fileIO().exists(published)) {
+ table.fileIO().deleteQuietly(published);
Review Comment:
Fixed in the latest commit. Promotion no longer deletes or renames published
files. Each one is replaced with `FileIO#overwriteFileUtf8`, the existing
atomic replace (overwrite rename on HDFS, single PUT on object stores), so a
position is never missing. The file the hint points to is replaced last, after
the hint moves to the new head, and the old manifests are deleted only after
promotion succeeds.
`testHeadPromotionFailureKeepsPublishedHead` makes the head overwrite fail
and checks that the hint and the old head are unchanged and still readable.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -459,6 +462,397 @@ private void createMetadata(
}
}
+ /**
+ * Create Iceberg metadata when no usable base metadata exists: either the
very first Iceberg
+ * commit for this table, or a recovery after the previous metadata became
unusable (format
+ * version change, missing row lineage, Iceberg-layer commit failure).
+ *
+ * <p>By default only the current snapshot is exposed to Iceberg. With
{@link
+ * IcebergOptions#SYNC_FULL_HISTORY} the whole retained Paimon history is
replayed instead, so
+ * Iceberg readers keep time travel and tags (see <a
+ *
href="https://github.com/apache/paimon/issues/6107";>apache/paimon#6107</a>).
+ */
+ private void recreateMetadata(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ if (syncFullHistory) {
+ rebuildFullHistory(snapshotId, inheritUuid, lastColumnIdFloor,
nextRowIdFloor);
+ } else {
+ createMetadataWithoutBase(snapshotId, inheritUuid,
lastColumnIdFloor, nextRowIdFloor);
+ }
+ }
+
+ /**
+ * Rebuild Iceberg metadata from every Paimon snapshot that is still
retained, ending at {@code
+ * currentSnapshotId}: create metadata afresh for the earliest retained
snapshot, then replay
+ * each following snapshot on top of its predecessor, exactly like live
commits would have.
+ * Schemas, tags and (for format version 3) the row-id space therefore
accumulate consistently
+ * across the whole replayed history.
+ *
+ * <p>Every replay step writes to a uniquely named staging path; the
published chain, the
+ * version hint and the external catalog stay untouched until the final
staged metadata is
+ * durable, and only then is the staged chain promoted into place. A
rebuild that fails at any
+ * point therefore leaves the previously published metadata fully
readable, and its staged
+ * leftovers are removed by the next rebuild. Replayed snapshots keep
their original Paimon
+ * commit timestamps and are subject to the same retention policy ({@link
+ * CoreOptions#SNAPSHOT_NUM_RETAINED_MIN}, {@link
CoreOptions#SNAPSHOT_TIME_RETAINED}, ...) that
+ * live commits apply.
+ */
+ private void rebuildFullHistory(
+ long currentSnapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor)
+ throws IOException {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Long earliest = snapshotManager.earliestSnapshotId();
+ long startId = earliest == null ? currentSnapshotId :
Math.min(earliest, currentSnapshotId);
+
+ deleteStagedLeftovers();
+
+ // Resume from the newest existing metadata below the current
snapshot, if it is usable.
+ // Anything older than the newest existing file is stale by
definition: live commits only
+ // ever read the immediately preceding metadata.
+ long baseId = -1;
+ for (long id = currentSnapshotId - 1; id >= startId; id--) {
+ Path metadataPath = pathFactory.toMetadataPath(id);
+ if (table.fileIO().exists(metadataPath)) {
+ try {
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(table.fileIO(),
metadataPath);
+ if (isUsableReplayBase(metadata, id, startId)) {
+ baseId = id;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read existing Iceberg metadata {},
rebuilding history from scratch",
+ metadataPath,
+ e);
+ }
+ break;
+ }
+ }
+
+ String rebuildUuid = UUID.randomUUID().toString();
+ long firstStagedId;
+ boolean freshRebuild = baseId == -1;
+ StaleBuild staleBuild = null;
+ if (freshRebuild) {
+ // No usable base: the whole old build is stale. Nothing of it is
touched during the
+ // replay, so an external catalog that still points at the old
metadata keeps a fully
+ // readable table; the old files are replaced and cleaned only by
the promotion. Their
+ // references are collected up front, tolerating unreadable files
(that is what
+ // triggered some rebuilds in the first place).
+ staleBuild = collectStaleBuild(currentSnapshotId);
+ firstStagedId = startId;
+ createMetadataWithoutBase(
+ startId,
+ inheritUuid,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, startId));
+ } else {
+ firstStagedId = baseId + 1;
+ }
+
+ for (long id = firstStagedId == startId ? startId + 1 : firstStagedId;
+ id <= currentSnapshotId;
+ id++) {
+ long snapshotId = id;
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ Path basePath =
+ snapshotId - 1 < firstStagedId
+ ? pathFactory.toMetadataPath(snapshotId - 1)
+ : stagedMetadataPath(rebuildUuid, snapshotId - 1);
+ createMetadataWithBase(
+ (removedFiles, addedFiles) ->
+ collectFileChanges(snapshotId, removedFiles,
addedFiles),
+ indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX),
+ snapshot,
+ basePath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ stagedMetadataPath(rebuildUuid, snapshotId));
+ }
+
+ boolean promoted =
+ promoteStagedReplay(rebuildUuid, firstStagedId,
currentSnapshotId, startId);
+ if (promoted && staleBuild != null) {
+ deleteStaleBuild(staleBuild, currentSnapshotId, startId);
+ }
+ }
+
+ /**
+ * Whether an existing metadata file can serve as the resume base of a
full-history replay. On
+ * top of the structural checks, the base must describe the live timeline:
an abandoned base (a
+ * rolled-back Paimon snapshot reused the id) or one carrying re-evolved
schema definitions
+ * would be rejected again by the first replay step, and a rebuild must
never select a base its
+ * own replay refuses to extend.
+ */
+ private boolean isUsableReplayBase(IcebergMetadata metadata, long baseId,
long startId) {
+ if (!isSameFormatVersion(metadata.formatVersion())) {
+ return false;
+ }
+ if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3 &&
metadata.nextRowId() == null) {
+ return false;
+ }
+ if (!coversRetainedPrefix(metadata, baseId, startId)) {
+ return false;
+ }
+ SnapshotManager snapshotManager = table.snapshotManager();
+ if (!snapshotManager.snapshotExists(baseId)
+ || !metadataMatchesSnapshot(metadata,
snapshotManager.snapshot(baseId))) {
+ return false;
+ }
+ SchemaCache schemaCache = new SchemaCache();
+ long latestSchemaId = schemaCache.getLatestSchemaId();
+ for (IcebergSchema known : metadata.schemas()) {
+ if (known.schemaId() > latestSchemaId
+ || !known.equals(schemaCache.get(known.schemaId()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private Path stagedMetadataPath(String rebuildUuid, long snapshotId) {
+ return new Path(
+ pathFactory.metadataDirectory(),
+ String.format("rebuild-%s-v%d.metadata.json", rebuildUuid,
snapshotId));
+ }
+
+ /** Staged files of crashed or superseded rebuilds; only ever garbage. */
+ private void deleteStagedLeftovers() throws IOException {
+ FileStatus[] statuses;
+ try {
+ statuses =
table.fileIO().listStatus(pathFactory.metadataDirectory());
+ } catch (FileNotFoundException e) {
+ return;
+ }
+ for (FileStatus status : statuses) {
+ String name = status.getPath().getName();
+ if (name.startsWith("rebuild-") &&
name.endsWith(".metadata.json")) {
+ table.fileIO().deleteQuietly(status.getPath());
+ }
+ }
+ }
+
+ /**
+ * Switch the published chain to the staged one, then the version hint and
the external catalog.
+ * The staged head is durable and validated before any published path is
touched; the remaining
+ * window is the per-file replacement of each chain position, ending one
below the head, and
+ * every path serves complete metadata again as soon as its replacement
lands.
+ *
+ * @return false if a newer commit superseded this rebuild; the staged
files are discarded and
+ * the published chain is left for that commit's own rebuild
+ */
+ private boolean promoteStagedReplay(
+ String rebuildUuid, long firstStagedId, long currentSnapshotId,
long startId)
+ throws IOException {
+ IcebergMetadata head =
+ IcebergMetadata.fromPath(
+ table.fileIO(), stagedMetadataPath(rebuildUuid,
currentSnapshotId));
+ Preconditions.checkState(
+ head.currentSnapshotId() == currentSnapshotId,
+ "Staged replay head is at snapshot %s instead of %s",
+ head.currentSnapshotId(),
+ currentSnapshotId);
+ // a truncated chain must never be promoted, no matter what produced it
+ Preconditions.checkState(
+ coversRetainedPrefix(head, currentSnapshotId, startId),
+ "Staged replay head for snapshot %s does not cover the
retained history",
+ currentSnapshotId);
+
+ Long latest = table.snapshotManager().latestSnapshotId();
Review Comment:
Fixed in the same commit. Every publication, live or replayed, now runs the
latest-snapshot check, the hint switch and the catalog commit under the table
lock from `CatalogEnvironment#snapshotCommit`, the same lock snapshot commits
take. Without a configured lock, the pointer moves forward after each switch to
the newest snapshot with complete metadata.
`testNewerCommitDuringPromotionIsNotRolledBack` commits snapshot 4 during
the snapshot-3 switch and checks that the hint ends at 4. Concurrent rebuilds
also no longer delete each other's staging files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
