[iceberg] Enhance iceberg snapshot metadata. (#6354)

Author: plusplusjiajia

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -339,10 +339,13 @@ private void createMetadataWithoutBase(long snapshotId) throws IOException {
                 new IcebergSnapshot(
                         snapshotId,
                         snapshotId,
+                        null,
                         System.currentTimeMillis(),
                         IcebergSnapshotSummary.APPEND,
                         pathFactory.toManifestListPath(manifestListFileName).toString(),
-                        schemaId);
+                        schemaId,
+                        null,
+                        null);
 
         // Tags can only be included in Iceberg if they point to an Iceberg snapshot that
         // exists. Otherwise an Iceberg client fails to parse the metadata and all reads fail.
@@ -599,10 +602,13 @@ private void createMetadataWithBase(
                 new IcebergSnapshot(
                         snapshotId,
                         snapshotId,
+                        snapshotId - 1,
                         System.currentTimeMillis(),
                         snapshotSummary,
                         pathFactory.toManifestListPath(manifestListFileName).toString(),
-                        schemaId));
+                        schemaId,
+                        null,
+                        null));
 
         // all snapshots in this list, except the last one, need to expire
         List<IcebergSnapshot> toExpireExceptLast = new ArrayList<>();

paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergMetadata.java

@@ -319,6 +319,7 @@ public int hashCode() {
                 defaultSortOrderId,
                 snapshots,
                 currentSnapshotId,
+                properties,
                 refs);
     }
 
@@ -347,6 +348,7 @@ public boolean equals(Object o) {
                 && defaultSortOrderId == that.defaultSortOrderId
                 && Objects.equals(snapshots, that.snapshots)
                 && currentSnapshotId == that.currentSnapshotId
+                && Objects.equals(properties, that.properties)
                 && Objects.equals(refs, that.refs);
     }
 }

paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSnapshot.java

@@ -21,8 +21,11 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+import javax.annotation.Nullable;
+
 import java.util.Objects;
 
 /**
@@ -35,17 +38,25 @@ public class IcebergSnapshot {
 
     private static final String FIELD_SEQUENCE_NUMBER = "sequence-number";
     private static final String FIELD_SNAPSHOT_ID = "snapshot-id";
+    private static final String FIELD_PARENT_SNAPSHOT_ID = "parent-snapshot-id";
     private static final String FIELD_TIMESTAMP_MS = "timestamp-ms";
     private static final String FIELD_SUMMARY = "summary";
     private static final String FIELD_MANIFEST_LIST = "manifest-list";
     private static final String FIELD_SCHEMA_ID = "schema-id";
+    private static final String FIELD_FIRST_ROW_ID = "first-row-id";
+    private static final String FIELD_ADDED_ROWS = "added-rows";
 
     @JsonProperty(FIELD_SEQUENCE_NUMBER)
     private final long sequenceNumber;
 
     @JsonProperty(FIELD_SNAPSHOT_ID)
     private final long snapshotId;
 
+    @JsonProperty(FIELD_PARENT_SNAPSHOT_ID)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final Long parentSnapshotId;
+
     @JsonProperty(FIELD_TIMESTAMP_MS)
     private final long timestampMs;
 
@@ -58,20 +69,36 @@ public class IcebergSnapshot {
     @JsonProperty(FIELD_SCHEMA_ID)
     private final int schemaId;
 
+    @JsonProperty(FIELD_FIRST_ROW_ID)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final Long firstRowId;
+
+    @JsonProperty(FIELD_ADDED_ROWS)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final Long addedRows;
+
     @JsonCreator
     public IcebergSnapshot(
             @JsonProperty(FIELD_SEQUENCE_NUMBER) long sequenceNumber,
             @JsonProperty(FIELD_SNAPSHOT_ID) long snapshotId,
+            @JsonProperty(FIELD_PARENT_SNAPSHOT_ID) Long parentSnapshotId,
             @JsonProperty(FIELD_TIMESTAMP_MS) long timestampMs,
             @JsonProperty(FIELD_SUMMARY) IcebergSnapshotSummary summary,
             @JsonProperty(FIELD_MANIFEST_LIST) String manifestList,
-            @JsonProperty(FIELD_SCHEMA_ID) int schemaId) {
+            @JsonProperty(FIELD_SCHEMA_ID) int schemaId,
+            @JsonProperty(FIELD_FIRST_ROW_ID) Long firstRowId,
+            @JsonProperty(FIELD_ADDED_ROWS) Long addedRows) {
         this.sequenceNumber = sequenceNumber;
         this.snapshotId = snapshotId;
+        this.parentSnapshotId = parentSnapshotId;
         this.timestampMs = timestampMs;
         this.summary = summary;
         this.manifestList = manifestList;
         this.schemaId = schemaId;
+        this.firstRowId = firstRowId;
+        this.addedRows = addedRows;
     }
 
     @JsonGetter(FIELD_SEQUENCE_NUMBER)
@@ -104,10 +131,33 @@ public int schemaId() {
         return schemaId;
     }
 
+    @JsonGetter(FIELD_PARENT_SNAPSHOT_ID)
+    public Long parentSnapshotId() {
+        return parentSnapshotId;
+    }
+
+    @JsonGetter(FIELD_ADDED_ROWS)
+    public Long addedRows() {
+        return addedRows;
+    }
+
+    @JsonGetter(FIELD_FIRST_ROW_ID)
+    public Long firstRowId() {
+        return firstRowId;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(
-                sequenceNumber, snapshotId, timestampMs, summary, manifestList, schemaId);
+                sequenceNumber,
+                snapshotId,
+                parentSnapshotId,
+                timestampMs,
+                summary,
+                manifestList,
+                schemaId,
+                addedRows,
+                firstRowId);
     }
 
     @Override
@@ -122,9 +172,12 @@ public boolean equals(Object o) {
         IcebergSnapshot that = (IcebergSnapshot) o;
         return sequenceNumber == that.sequenceNumber
                 && snapshotId == that.snapshotId
+                && Objects.equals(parentSnapshotId, that.parentSnapshotId)
                 && timestampMs == that.timestampMs
                 && Objects.equals(summary, that.summary)
                 && Objects.equals(manifestList, that.manifestList)
-                && schemaId == that.schemaId;
+                && schemaId == that.schemaId
+                && Objects.equals(addedRows, that.addedRows)
+                && Objects.equals(firstRowId, that.firstRowId);
     }
 }

paimon-iceberg/src/main/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitter.java

@@ -374,10 +374,13 @@ private IcebergMetadata adjustMetadataForRest(IcebergMetadata newIcebergMetadata
                                         new IcebergSnapshot(
                                                 snapshot.sequenceNumber(),
                                                 snapshot.snapshotId(),
+                                                snapshot.parentSnapshotId(),
                                                 snapshot.timestampMs(),
                                                 snapshot.summary(),
                                                 snapshot.manifestList(),
-                                                snapshot.schemaId() + 1))
+                                                snapshot.schemaId() + 1,
+                                                snapshot.firstRowId(),
+                                                snapshot.addedRows()))
                         .collect(Collectors.toList());
         return new IcebergMetadata(
                 newIcebergMetadata.formatVersion(),