JingsongLi commented on code in PR #9348:
URL: https://github.com/apache/paimon/pull/9348#discussion_r3932140672
##########
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:
[P1] Keep the published head crash-safe during promotion. This still deletes
an existing before the staged rename. If the process dies in that window, or
both attempts return false/throw (which explicitly permits), Hive/REST and
the version hint can remain pointed at a missing head indefinitely—the same
availability failure staging was meant to eliminate. Please use an atomic
overwrite/replace primitive for published positions, or publish a
generation-specific immutable head and switch only the external pointer; add a
failure-injection test at the head promotion.
##########
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:
[P1] Prevent an older replay from publishing after a newer commit. This
latest-snapshot check is separated from both the per-file promotion and the
hint/catalog update. Snapshot N+1 can commit after this check, extend/publish
metadata, and then this slow N replay overwrites the hint and external catalog
back to N; depending on timing, N+1 can also extend the old partial chain and
permanently lose the rebuilt history. Recheck is still TOCTOU unless
publication is serialized/claimed conditionally. Please make replay publication
mutually exclusive with later callbacks (or use a CAS against the exact
catalog/base head), and test an N+1 callback interleaved after this check.
--
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]
