[WIP][iceberg] Support the v3 variant type with row lineage and REST catalog

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -311,6 +311,7 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
         if (tableRollback != null) {
             rollback = new CommitRollback(tableRollback);
         }
+        List<CommitCallback> commitCallbacks = createCommitCallbacks(commitUser, table);
         return new FileStoreCommitImpl(
                 snapshotCommit,
                 fileIO,
@@ -327,8 +328,8 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                 this::newScan,
                 newStatsFileHandler(),
                 bucketMode(),
-                createCommitPreCallbacks(table),
-                createCommitCallbacks(commitUser, table),
+                createCommitPreCallbacks(table, commitCallbacks),
+                commitCallbacks,
                 conflictDetectFactory,
                 rollback);
     }
@@ -389,11 +390,17 @@ public InternalRowPartitionComputer partitionComputer() {
                 options.legacyPartitionName());
     }
 
-    private List<CommitPreCallback> createCommitPreCallbacks(FileStoreTable table) {
+    private List<CommitPreCallback> createCommitPreCallbacks(
+            FileStoreTable table, List<CommitCallback> commitCallbacks) {
         List<CommitPreCallback> callbacks = new ArrayList<>();
         if (options.isChainTable()) {
             callbacks.add(new ChainTableCommitPreCallback(table));
         }
+        // reuse the same Iceberg callback instance: it validates before the commit too
+        commitCallbacks.stream()
+                .filter(callback -> callback instanceof IcebergCommitCallback)
+                .map(callback -> (IcebergCommitCallback) callback)
+                .forEach(callbacks::add);
         return callbacks;
     }
 
@@ -591,7 +598,7 @@ public List<TagCallback> createTagCallbacks(FileStoreTable table) {
         }
         if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
                 != IcebergOptions.StorageType.DISABLED) {
-            callbacks.add(new IcebergCommitCallback(table, ""));
+            callbacks.add(IcebergCommitCallback.forTagCallbacks(table));
         }
         return callbacks;
     }

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

@@ -0,0 +1,94 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+
+import java.io.IOException;
+
+/**
+ * Serializes base-less v3 row-id allocation across concurrent Iceberg metadata callbacks.
+ *
+ * <p>A regeneration with no metadata base ({@code createMetadataWithoutBase}) picks its start by
+ * scanning surviving metadata; concurrent regenerations scan the same watermark and would assign
+ * overlapping {@code first_row_id} ranges. This reserves the range through an atomic
+ * write-if-not-exists on a monotonically versioned file: two reservers that read the same state
+ * cannot both advance the same version, so the loser retries and reserves a disjoint range. The
+ * start is never below {@code floor}, the highest id already published in surviving metadata.
+ *
+ * <p>Reservation files live in a dot-directory that Iceberg metadata scans and retention ignore.
+ * The atomic guarantee is only as strong as {@link FileIO#tryToWriteAtomic} on the underlying store
+ * (durable on HDFS/local rename; object storage needs a conditional put or an external lock), which
+ * matches Iceberg's own {@code HadoopTableOperations} requirement.
+ */
+class IcebergRowIdReservation {
+
+    private final FileIO fileIO;
+    private final Path reservationDir;
+
+    /** Invoked between reading the watermark and the CAS write; a seam for deterministic tests. */
+    @VisibleForTesting Runnable beforeCommitHook = () -> {};
+
+    IcebergRowIdReservation(FileIO fileIO, Path metadataDirectory) {
+        this.fileIO = fileIO;
+        this.reservationDir = new Path(metadataDirectory, ".rowid-reservation");
+    }
+
+    /**
+     * Reserves {@code amount} contiguous row ids at or above {@code floor} and returns the start.
+     * Retries until it wins the compare-and-set, so concurrent reservers receive disjoint ranges.
+     */
+    long reserve(long amount, long floor) throws IOException {
+        while (true) {
+            long version = latestVersion();
+            long reserved = version == 0 ? floor : readNextRowId(version);
+            long start = Math.max(floor, reserved);
+            beforeCommitHook.run();
+            if (fileIO.tryToWriteAtomic(versionFile(version + 1), Long.toString(start + amount))) {
+                return start;
+            }
+            // lost the CAS: another reserver advanced this version, so re-read and retry
+        }
+    }
+
+    private long latestVersion() throws IOException {
+        if (!fileIO.exists(reservationDir)) {
+            return 0;
+        }
+        long max = 0;
+        for (FileStatus status : fileIO.listStatus(reservationDir)) {
+            String name = status.getPath().getName();
+            if (name.startsWith("v") && name.endsWith(".rowid")) {
+                max = Math.max(max, Long.parseLong(name.substring(1, name.length() - 6)));
+            }
+        }
+        return max;
+    }
+
+    private long readNextRowId(long version) throws IOException {
+        return Long.parseLong(fileIO.readFileUtf8(versionFile(version)).trim());
+    }
+
+    private Path versionFile(long version) {
+        return new Path(reservationDir, "v" + version + ".rowid");
+    }
+}

paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java

@@ -87,6 +87,9 @@ public static Content fromId(int id) {
     private final String referencedDataFile;
     private final Long contentOffset;
     private final Long contentSizeInBytes;
+    // v3 row lineage: explicit first row id of this file; null means the id is inherited
+    // from the manifest's first_row_id at read time
+    private final Long firstRowId;
 
     IcebergDataFileMeta(
             Content content,
@@ -126,6 +129,36 @@ public static Content fromId(int id) {
             String referencedDataFile,
             Long contentOffset,
             Long contentSizeInBytes) {
+        this(
+                content,
+                filePath,
+                fileFormat,
+                partition,
+                recordCount,
+                fileSizeInBytes,
+                nullValueCounts,
+                lowerBounds,
+                upperBounds,
+                referencedDataFile,
+                contentOffset,
+                contentSizeInBytes,
+                null);
+    }
+
+    IcebergDataFileMeta(
+            Content content,
+            String filePath,
+            String fileFormat,
+            BinaryRow partition,
+            long recordCount,
+            long fileSizeInBytes,
+            InternalMap nullValueCounts,
+            InternalMap lowerBounds,
+            InternalMap upperBounds,
+            String referencedDataFile,
+            Long contentOffset,
+            Long contentSizeInBytes,
+            Long firstRowId) {
         this.content = content;
         this.filePath = filePath;
         this.fileFormat = fileFormat;
@@ -139,6 +172,24 @@ public static Content fromId(int id) {
         this.referencedDataFile = referencedDataFile;
         this.contentOffset = contentOffset;
         this.contentSizeInBytes = contentSizeInBytes;
+        this.firstRowId = firstRowId;
+    }
+
+    public IcebergDataFileMeta withFirstRowId(Long firstRowId) {
+        return new IcebergDataFileMeta(
+                content,
+                filePath,
+                fileFormat,
+                partition,
+                recordCount,
+                fileSizeInBytes,
+                nullValueCounts,
+                lowerBounds,
+                upperBounds,
+                referencedDataFile,
+                contentOffset,
+                contentSizeInBytes,
+                firstRowId);
     }
 
     public static IcebergDataFileMeta create(
@@ -294,6 +345,10 @@ public Long contentSizeInBytes() {
         return contentSizeInBytes;
     }
 
+    public Long firstRowId() {
+        return firstRowId;
+    }
+
     public static RowType schema(RowType partitionType) {
         List<DataField> fields = new ArrayList<>();
         fields.add(new DataField(134, "content", DataTypes.INT().notNull()));
@@ -320,6 +375,8 @@ public static RowType schema(RowType partitionType) {
         fields.add(new DataField(143, "referenced_data_file", DataTypes.STRING()));
         fields.add(new DataField(144, "content_offset", DataTypes.BIGINT()));
         fields.add(new DataField(145, "content_size_in_bytes", DataTypes.BIGINT()));
+        // v3 row lineage (optional per spec, ignored by v2 readers)
+        fields.add(new DataField(142, "first_row_id", DataTypes.BIGINT()));
         return new RowType(false, fields);
     }
 
@@ -343,7 +400,8 @@ public boolean equals(Object o) {
                 && Objects.equals(upperBounds, that.upperBounds)
                 && Objects.equals(referencedDataFile, that.referencedDataFile)
                 && Objects.equals(contentOffset, that.contentOffset)
-                && Objects.equals(contentSizeInBytes, that.contentSizeInBytes);
+                && Objects.equals(contentSizeInBytes, that.contentSizeInBytes)
+                && Objects.equals(firstRowId, that.firstRowId);
     }
 
     @Override
@@ -360,6 +418,7 @@ public int hashCode() {
                 upperBounds,
                 referencedDataFile,
                 contentOffset,
-                contentSizeInBytes);
+                contentSizeInBytes,
+                firstRowId);
     }
 }

paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java

@@ -37,13 +37,17 @@ public class IcebergDataFileMetaSerializer extends ObjectSerializer<IcebergDataF
     private final InternalMapSerializer lowerBoundsSerializer;
     private final InternalMapSerializer upperBoundsSerializer;
 
+    private final int firstRowIdPos;
+
     public IcebergDataFileMetaSerializer(RowType partitionType) {
         super(IcebergDataFileMeta.schema(partitionType));
         this.partSerializer = new InternalRowSerializer(partitionType);
         this.nullValueCountsSerializer =
                 new InternalMapSerializer(DataTypes.INT(), DataTypes.BIGINT());
         this.lowerBoundsSerializer = new InternalMapSerializer(DataTypes.INT(), DataTypes.BYTES());
         this.upperBoundsSerializer = new InternalMapSerializer(DataTypes.INT(), DataTypes.BYTES());
+        this.firstRowIdPos =
+                IcebergDataFileMeta.schema(partitionType).getFieldIndex("first_row_id");
     }
 
     @Override
@@ -60,7 +64,8 @@ public InternalRow toRow(IcebergDataFileMeta file) {
                 file.upperBounds(),
                 BinaryString.fromString(file.referencedDataFile()),
                 file.contentOffset(),
-                file.contentSizeInBytes());
+                file.contentSizeInBytes(),
+                file.firstRowId());
     }
 
     @Override
@@ -77,6 +82,9 @@ public IcebergDataFileMeta fromRow(InternalRow row) {
                 upperBoundsSerializer.copy(row.getMap(8)),
                 row.isNullAt(9) ? null : row.getString(9).toString(),
                 row.isNullAt(10) ? null : row.getLong(10),
-                row.isNullAt(11) ? null : row.getLong(11));
+                row.isNullAt(11) ? null : row.getLong(11),
+                firstRowIdPos >= 0 && !row.isNullAt(firstRowIdPos)
+                        ? row.getLong(firstRowIdPos)
+                        : null);
     }
 }