JingsongLi commented on code in PR #9348:
URL: https://github.com/apache/paimon/pull/9348#discussion_r3888247392
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -459,6 +462,253 @@ 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>Each replay step persists its metadata file, so an interrupted
rebuild resumes from the
+ * newest already-written metadata on the next commit. Intermediate steps
skip the version hint
+ * and the external catalog commit; only the final step publishes, so an
external catalog sees a
+ * single transition. 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);
+
+ // 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 (isSameFormatVersion(metadata.formatVersion())
+ && (formatVersion <
IcebergMetadata.FORMAT_VERSION_V3
+ || metadata.nextRowId() != null)
+ && coversRetainedPrefix(metadata, id, startId)) {
Review Comment:
[P1] Reject unusable resume bases before selecting them
This candidate is not checked against the live snapshot identity or schema
definitions. If an abandoned-timeline/schema-incompatible base triggered the
rebuild, this loop selects the same base again; the final
`createMetadataWithBase` rejects it and recursively enters `rebuildFullHistory`
through `recreateFromUnusableBase`. I reproduced this with rollback, an
Iceberg-disabled commit that reused snapshot 2, and a full-history commit for
snapshot 3: it ends in `StackOverflowError`. Please apply all
extension-usability checks here, or explicitly exclude the rejected base and
restart a non-recursive fresh replay.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -1274,6 +1635,28 @@ private void createMetadataWithBase(
}
}
+ /**
+ * Recreate metadata when the base metadata of a commit turned out to be
unusable. At the head
+ * of the history this honors {@link IcebergOptions#SYNC_FULL_HISTORY}; in
the middle of a
+ * {@link #rebuildFullHistory(long)} replay (where an unusable base should
be impossible, since
+ * the replay itself validates or writes every base) it falls back to
single-snapshot metadata
+ * instead of recursing into another replay.
+ */
+ private void recreateFromUnusableBase(
+ long snapshotId,
+ @Nullable String inheritUuid,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ boolean intermediate)
+ throws IOException {
+ if (intermediate) {
+ createMetadataWithoutBase(
Review Comment:
[P1] Do not truncate history after an intermediate base-read failure
When `intermediate` is true, this fallback writes metadata containing only
`snapshotId`. The replay loop then extends that truncated base and can publish
the final file without revalidating `coversRetainedPrefix`, so one transient
I/O error, corruption, or concurrent replacement silently drops all earlier
retained snapshots and tags. Please propagate/retry the failure or restart from
`startId`, and validate the retained prefix again before final publication.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -937,8 +1255,45 @@ private void createMetadataWithBase(
int lastColumnIdFloor,
long nextRowIdFloor)
throws IOException {
+ createMetadataWithBase(
+ fileChangesCollector,
+ indexFiles,
+ snapshot,
+ baseMetadataPath,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ false);
+ }
+
+ /**
+ * @param intermediate whether this metadata is an intermediate step of a
{@link
+ * #rebuildFullHistory} replay; intermediate steps skip the version
hint and the external
+ * catalog commit, which only the final step publishes.
+ */
+ private void createMetadataWithBase(
+ FileChangesCollector fileChangesCollector,
+ List<IndexManifestEntry> indexFiles,
+ Snapshot snapshot,
+ Path baseMetadataPath,
+ int lastColumnIdFloor,
+ long nextRowIdFloor,
+ boolean intermediate)
+ throws IOException {
long snapshotId = snapshot.id();
- IcebergMetadata baseMetadata =
IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath);
+ IcebergMetadata baseMetadata;
+ try {
+ baseMetadata = IcebergMetadata.fromPath(table.fileIO(),
baseMetadataPath);
+ } catch (Exception e) {
+ // an unreadable base is an unusable base: recreate instead of
failing the commit,
+ // so a corrupted metadata file self-heals like a structurally
invalid one
+ LOG.warn(
+ "Unreadable base Iceberg metadata {}, recreating
metadata.",
+ baseMetadataPath,
+ e);
+ recreateFromUnusableBase(
Review Comment:
[P1] Do not reset v3 row lineage after an unreadable base
For a normal next commit, the caller floors are commonly zero. If the
published v3 base is corrupt, this catch therefore rebuilds with a null UUID
and `nextRowIdFloor == 0`, even though the unreadable metadata may already have
issued higher row IDs. A regression test with `v2.next-row-id == 3` produces
snapshot 3 with `first-row-id == 0`, reusing IDs 0-2; before this change the
commit failed instead of publishing reset lineage. For v3, recover the
UUID/high-water mark from trustworthy older or catalog metadata, and fail
safely when it cannot be recovered.
##########
paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java:
##########
@@ -459,6 +462,253 @@ 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>Each replay step persists its metadata file, so an interrupted
rebuild resumes from the
+ * newest already-written metadata on the next commit. Intermediate steps
skip the version hint
+ * and the external catalog commit; only the final step publishes, so an
external catalog sees a
+ * single transition. 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);
+
+ // 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 (isSameFormatVersion(metadata.formatVersion())
+ && (formatVersion <
IcebergMetadata.FORMAT_VERSION_V3
+ || metadata.nextRowId() != null)
+ && coversRetainedPrefix(metadata, id, startId)) {
+ baseId = id;
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read existing Iceberg metadata {},
rebuilding history from scratch",
+ metadataPath,
+ e);
+ }
+ break;
+ }
+ }
+
+ long firstWithBase;
+ boolean freshRebuild = baseId == -1;
+ StaleBuild staleBuild = null;
+ if (freshRebuild) {
+ // No usable base: the whole old build is stale. Nothing of it is
deleted yet, so
+ // an external catalog that still points at the old metadata keeps
a fully readable
+ // table during the entire replay; the old files are cleaned only
after the final
+ // step has published. Their references are collected up front,
tolerating
+ // unreadable files (that is what triggered some rebuilds in the
first place).
+ staleBuild = collectStaleBuild(currentSnapshotId);
+ // a leftover file in the replay range must not survive as a
replay step's output:
+ // it may match the step's commit identity (a regenerated build of
the same Paimon
+ // snapshot does) while carrying another format or content, so
each target is
+ // removed just before its replacement is written
+ table.fileIO().deleteQuietly(pathFactory.toMetadataPath(startId));
+ createMetadataWithoutBase(
+ startId,
+ inheritUuid,
+ lastColumnIdFloor,
+ nextRowIdFloor,
+ startId != currentSnapshotId);
+ firstWithBase = startId + 1;
+ } else {
+ firstWithBase = baseId + 1;
+ }
+
+ for (long id = firstWithBase; id <= currentSnapshotId; id++) {
+ long snapshotId = id;
+ Snapshot snapshot = snapshotManager.snapshot(snapshotId);
+ if (freshRebuild && snapshotId != currentSnapshotId) {
+ // see above; the final step replaces its twin through the
regular write path
+
table.fileIO().deleteQuietly(pathFactory.toMetadataPath(snapshotId));
Review Comment:
[P1] Keep the published metadata file intact until replay commits
At this point `v${snapshotId}.metadata.json` can still be the file
referenced by Hive's `metadata_location` or Hadoop's version hint, especially
when `snapshotId == currentSnapshotId - 1`. Deleting it before
`createMetadataWithBase` finishes creates a reader outage, and any later
manifest/DV/metadata write failure leaves the external pointer aimed at a
missing file until another callback succeeds. `intermediate` suppresses
hint/catalog updates, but it does not make mutations to the already-published
path invisible. Please build the replay under unique staging paths and switch
the published head only after the final metadata is durable.
--
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]
