iceberg maintenance: v3 row lineage so v3 tables compact and rewrite manifests (#11653)

* iceberg maintenance: commit v3 row lineage on the three commit paths

iceberg-go 0.7.0 refuses every snapshot on a format-version 3 table
that lacks first-row-id and added-rows ("invalid row lineage"), so no
v3 table was ever compacted or had its manifests rewritten; worse,
passing 0 to the manifest list would have handed new files row IDs
existing rows already hold.

Each commit now plans first-row-id from the table's next-row-id, lets
the v3 manifest-list writer assign manifest first_row_id from there,
and reports the assigned range as the snapshot's added-rows so
next-row-id advances by exactly that. The commit guard also treats a
moved next-row-id or a changed format version as a stale plan, like a
moved head: a commit to another branch advances next-row-id too, and
manifests written for the old version must not be committed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: keep row identity through v3 compaction

A rewrite that moves rows into a new data file must not renumber them.
On v3 tables the merge now appends _row_id and
_last_updated_sequence_number (spec field ids 2147483540/2147483539)
to the output and fills each row with the value its input
materializes, else the inherited one: the input file's first_row_id
plus the row's position before any delete, and the manifest entry's
data sequence number. Inputs are matched with their lineage columns
set aside, so a file an earlier compaction wrote merges with one an
engine wrote; a duplicate lineage column or a reserved name under
another field id fails the bin instead of writing a doubled column.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: skip rewrite_position_delete_files on v3 tables

The spec forbids adding position-delete files to a format-version 3
table (v3 writes deletion vectors), and this rewrite adds new ones.
Detection no longer proposes the operation and the operation returns a
skip result; compaction still applies and removes existing position
deletes.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: keep Puffin deletion vectors out of position-delete reads

A Puffin file in a delete manifest is a deletion vector, not a
position-delete file: reading it as Parquet fails, and merging the data
file it covers without applying the vector resurrects deleted rows.

Skip Puffin entries in the position-delete collection and the delete
rewrite group scan, and exclude the files a vector covers from
compaction candidates - in compactDataFiles and in eligibility
detection alike. A vector whose referenced_data_file is missing or
unresolvable fails the run rather than silently compacting whatever it
points at.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: name a merged file's own first row ID

Rewritten rows keep their IDs, so a merged data file's first_row_id is
the _row_id of its first written row - not the next free range the
manifest list would otherwise assign to the entry at commit. Read the
first materialized value back off the merged output and stamp it on the
manifest entry.

Add an end-to-end test that populates a v3 table whose entries carry no
first_row_id, so the compact path exercises manifest first_row_id
inheritance through the stored manifest list, then commits and reads
the result back through the real manifest readers.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: locate the _row_id column by index, not row position

Repeated columns shift parquet.Row positions, so leaf.ColumnIndex is not
the value's slot — a first row containing a list or map saved an unrelated
value as the merged file's first_row_id.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: read the returned row before treating EOF as empty

parquet-go can return a row alongside io.EOF; a one-row merged file then
lost its first_row_id and the manifest list would allocate a fresh range.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: avoid nil dereference in first-row-id failure message

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/worker/tasks/iceberg/compact.go

@@ -113,6 +113,7 @@ func (h *Handler) compactDataFiles(
 	var positionDeletes map[string][]int64
 	var eqDeleteGroups []equalityDeleteGroup
 	var allDeleteEntries []iceberg.ManifestEntry
+	var dvPaths map[string]bool
 	if config.ApplyDeletes && len(deleteManifests) > 0 {
 		for _, mf := range deleteManifests {
 			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
@@ -126,10 +127,19 @@ func (h *Handler) compactDataFiles(
 			allDeleteEntries = append(allDeleteEntries, entries...)
 		}
 
+		var err error
+		dvPaths, err = deletionVectorPaths(allDeleteEntries, bucketName, dataPath)
+		if err != nil {
+			return "", nil, err
+		}
+
 		// Separate position and equality deletes, filtering by partition
 		// predicate so out-of-scope deletes don't affect the merge.
 		var posDeleteEntries, eqDeleteEntries []iceberg.ManifestEntry
 		for _, entry := range allDeleteEntries {
+			if entry.DataFile().FileFormat() == iceberg.PuffinFile {
+				continue
+			}
 			if predicate != nil {
 				spec, ok := specsByID[int(entry.DataFile().SpecID())]
 				if !ok {
@@ -166,10 +176,13 @@ func (h *Handler) compactDataFiles(
 		}
 	}
 
-	candidateEntries := allEntries
+	candidateEntries, err := excludeDeletionVectorFiles(allEntries, dvPaths, bucketName, dataPath)
+	if err != nil {
+		return "", nil, err
+	}
 	if predicate != nil {
-		candidateEntries = make([]iceberg.ManifestEntry, 0, len(allEntries))
-		for _, entry := range allEntries {
+		filtered := make([]iceberg.ManifestEntry, 0, len(candidateEntries))
+		for _, entry := range candidateEntries {
 			spec, ok := specsByID[int(entry.DataFile().SpecID())]
 			if !ok {
 				continue
@@ -179,9 +192,10 @@ func (h *Handler) compactDataFiles(
 				return "", nil, err
 			}
 			if match {
-				candidateEntries = append(candidateEntries, entry)
+				filtered = append(filtered, entry)
 			}
 		}
+		candidateEntries = filtered
 	}
 
 	minInputFiles, err := compactionMinInputFiles(config.MinInputFiles)
@@ -205,6 +219,11 @@ func (h *Handler) compactDataFiles(
 
 	version := meta.Version()
 	snapshotID := currentSnap.SnapshotID
+	// v3 row lineage: the snapshot starts at the table's next-row-id, and the
+	// merged files carry every row's _row_id and _last_updated_sequence_number
+	// (row_lineage.go).
+	lineage := hasRowLineage(version)
+	firstRowID := rowLineageFirstRowID(meta)
 
 	// Compute the snapshot ID for the commit up front so all manifest entries
 	// reference the same snapshot that will actually be committed.
@@ -273,9 +292,9 @@ func (h *Handler) compactDataFiles(
 		var recordCount int64
 		rowGroupRows := rowsPerRowGroup(bin, config)
 		if rewritePlan != nil && rewritePlan.strategy == "sort" {
-			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows)
+			mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows, lineage)
 		} else {
-			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows)
+			mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows, lineage)
 		}
 		if err != nil {
 			glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err)
@@ -325,6 +344,20 @@ func (h *Handler) compactDataFiles(
 				stats.applyTo(dfBuilder)
 			}
 
+			// A merged file keeps its rows' row IDs, so its entry names the first
+			// materialized one instead of inheriting a fresh range at commit.
+			if lineage {
+				firstID, err := mergedFileFirstRowID(mergedData)
+				if err != nil {
+					glog.Warningf("iceberg compact: failed to read merged file lineage for bin %d: %v", binIdx, err)
+					_ = deleteFilerFile(ctx, filerClient, dataDir, mergedFileName)
+					goto binDone
+				}
+				if firstID != nil {
+					dfBuilder.FirstRowID(*firstID)
+				}
+			}
+
 			mergedDataFile := dfBuilder.Build()
 			summary.addFile(mergedDataFile)
 			newEntry := iceberg.NewManifestEntry(
@@ -471,7 +504,7 @@ func (h *Handler) compactDataFiles(
 	// Write new manifest list
 	var manifestListBuf bytes.Buffer
 	seqNum := currentSnap.SequenceNumber + 1
-	err = iceberg.WriteManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, 0, allManifests)
+	addedRows, err := writeManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, firstRowID, allManifests)
 	if err != nil {
 		return "", nil, fmt.Errorf("write compact manifest list: %w", err)
 	}
@@ -486,9 +519,9 @@ func (h *Handler) compactDataFiles(
 	manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListFileName)
 	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
 		// Guard: verify table head hasn't advanced since we planned.
-		cs := currentMeta.CurrentSnapshot()
-		if cs == nil || cs.SnapshotID != snapshotID {
-			return errStalePlan
+		cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID)
+		if err != nil {
+			return err
 		}
 
 		newSnapshot := &table.Snapshot{
@@ -512,6 +545,7 @@ func (h *Handler) compactDataFiles(
 		if rewritePlan != nil && rewritePlan.strategy == "sort" {
 			newSnapshot.Summary.Properties["sort-fields"] = rewritePlan.summaryLabel()
 		}
+		setRowLineage(newSnapshot, version, firstRowID, addedRows)
 		if err := builder.AddSnapshot(newSnapshot); err != nil {
 			return err
 		}
@@ -722,6 +756,48 @@ func partitionKey(partition map[int]any) string {
 	return strings.Join(parts, "\x00")
 }
 
+// deletionVectorPaths names the data files a Puffin deletion-vector entry
+// covers, normalized to merge-time keys. Merging a covered file needs a
+// Puffin reader to apply the vector; without one the merge would resurrect
+// deleted rows, so the file must stay out of compaction bins.
+func deletionVectorPaths(entries []iceberg.ManifestEntry, bucketName, dataPath string) (map[string]bool, error) {
+	paths := make(map[string]bool)
+	for _, entry := range entries {
+		df := entry.DataFile()
+		if df.ContentType() != iceberg.EntryContentPosDeletes || df.FileFormat() != iceberg.PuffinFile {
+			continue
+		}
+		ref := df.ReferencedDataFile()
+		if ref == nil || *ref == "" {
+			return nil, fmt.Errorf("deletion vector %s has no referenced data file", df.FilePath())
+		}
+		normalized, err := normalizeIcebergPath(*ref, bucketName, dataPath)
+		if err != nil {
+			return nil, fmt.Errorf("resolve deletion vector source %s: %w", *ref, err)
+		}
+		paths[normalized] = true
+	}
+	return paths, nil
+}
+
+// excludeDeletionVectorFiles drops entries a deletion vector covers.
+func excludeDeletionVectorFiles(entries []iceberg.ManifestEntry, dvPaths map[string]bool, bucketName, dataPath string) ([]iceberg.ManifestEntry, error) {
+	if len(dvPaths) == 0 {
+		return entries, nil
+	}
+	kept := make([]iceberg.ManifestEntry, 0, len(entries))
+	for _, entry := range entries {
+		normalized, err := normalizeIcebergPath(entry.DataFile().FilePath(), bucketName, dataPath)
+		if err != nil {
+			return nil, fmt.Errorf("resolve data file %s: %w", entry.DataFile().FilePath(), err)
+		}
+		if !dvPaths[normalized] {
+			kept = append(kept, entry)
+		}
+	}
+	return kept, nil
+}
+
 // collectPositionDeletes reads position delete Parquet files and returns a map
 // from normalized data file path to sorted row positions that should be deleted.
 // Paths are normalized so that absolute S3 URLs and relative paths match.
@@ -733,8 +809,9 @@ func collectPositionDeletes(
 ) (map[string][]int64, error) {
 	result := make(map[string][]int64)
 	for _, entry := range deleteEntries {
-		if entry.DataFile().ContentType() != iceberg.EntryContentPosDeletes {
-			continue
+		df := entry.DataFile()
+		if df.ContentType() != iceberg.EntryContentPosDeletes || df.FileFormat() == iceberg.PuffinFile {
+			continue // Puffin is a deletion vector, not a row-level delete file
 		}
 		fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 		if err != nil {
@@ -1011,6 +1088,7 @@ func mergeParquetFiles(
 	eqDeleteGroups []equalityDeleteGroup,
 	icebergSchema *iceberg.Schema,
 	rowGroupRows int64,
+	lineage bool,
 ) ([]byte, int64, error) {
 	if len(entries) == 0 {
 		return nil, 0, fmt.Errorf("no entries to merge")
@@ -1034,11 +1112,24 @@ func mergeParquetFiles(
 		return nil, 0, fmt.Errorf("resolve equality columns: %w", err)
 	}
 
-	var outputBuf bytes.Buffer
-	writer := parquet.NewWriter(&outputBuf, parquetSchema, parquet.MaxRowsPerRowGroup(rowGroupRows))
+	// On a v3 table the output is the first file's columns plus the lineage
+	// columns (row_lineage.go).
+	outputSchema := parquetSchema
+	var layout *lineageLayout
+	if lineage {
+		if layout, err = newLineageLayout(parquetSchema); err != nil {
+			firstReader.Close()
+			return nil, 0, fmt.Errorf("schema of %s: %w", entries[0].DataFile().FilePath(), err)
+		}
+		outputSchema = layout.schema
+	}
 
-	drainReader := func(reader *parquet.Reader, source string) (int64, error) {
-		return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
+	var outputBuf bytes.Buffer
+	writer := parquet.NewWriter(&outputBuf, outputSchema, parquet.MaxRowsPerRowGroup(rowGroupRows))
+
+	drainReader := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) {
+		source := entry.DataFile().FilePath()
+		return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error {
 			if _, err := writer.WriteRows(filtered); err != nil {
 				return fmt.Errorf("write rows from %s: %w", source, err)
 			}
@@ -1047,8 +1138,7 @@ func mergeParquetFiles(
 	}
 
 	// Drain the first file.
-	firstSource := entries[0].DataFile().FilePath()
-	totalRows, err := drainReader(firstReader, firstSource)
+	totalRows, err := drainReader(firstReader, entries[0])
 	if err != nil {
 		writer.Close()
 		return nil, 0, err
@@ -1071,13 +1161,14 @@ func mergeParquetFiles(
 		}
 
 		reader := parquet.NewReader(bytes.NewReader(data))
-		if !schemasEqual(parquetSchema, reader.Schema()) {
+		// A v3 merge matches files with their lineage columns set aside.
+		if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) {
 			reader.Close()
 			writer.Close()
 			return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath())
 		}
 
-		rowsWritten, err := drainReader(reader, entry.DataFile().FilePath())
+		rowsWritten, err := drainReader(reader, entry)
 		if err != nil {
 			writer.Close()
 			return nil, 0, err
@@ -1124,12 +1215,15 @@ func visitFilteredParquetRows(
 	source, bucketName, dataPath string,
 	positionDeletes map[string][]int64,
 	resolvedEqGroups []resolvedEqDeleteGroup,
-	onRows func([]parquet.Row) error,
+	onRows func(rows []parquet.Row, positions []int64) error,
 ) (int64, error) {
 	defer reader.Close()
 
 	rows := make([]parquet.Row, 256)
 	filteredRows := make([]parquet.Row, 0, len(rows))
+	// each kept row's position in the file before any delete: what a v3
+	// _row_id is inherited from
+	filteredPositions := make([]int64, 0, len(rows))
 	normalizedSource, err := normalizeIcebergPath(source, bucketName, dataPath)
 	if err != nil {
 		return 0, err
@@ -1149,6 +1243,7 @@ func visitFilteredParquetRows(
 		n, readErr := reader.ReadRows(rows)
 		if n > 0 {
 			filteredRows = filteredRows[:0]
+			filteredPositions = filteredPositions[:0]
 			for i := 0; i < n; i++ {
 				rowPos := absolutePos + int64(i)
 				for posDeleteIdx < len(posDeletes) && posDeletes[posDeleteIdx] < rowPos {
@@ -1172,10 +1267,11 @@ func visitFilteredParquetRows(
 				}
 
 				filteredRows = append(filteredRows, rows[i])
+				filteredPositions = append(filteredPositions, rowPos)
 			}
 			absolutePos += int64(n)
 			if len(filteredRows) > 0 {
-				if err := onRows(filteredRows); err != nil {
+				if err := onRows(filteredRows, filteredPositions); err != nil {
 					return totalRows, err
 				}
 				totalRows += int64(len(filteredRows))
@@ -1201,6 +1297,7 @@ func mergeParquetFilesSorted(
 	icebergSchema *iceberg.Schema,
 	rewritePlan *compactionRewritePlan,
 	rowGroupRows int64,
+	lineage bool,
 ) ([]byte, int64, error) {
 	if len(entries) == 0 {
 		return nil, 0, fmt.Errorf("no entries to merge")
@@ -1220,7 +1317,17 @@ func mergeParquetFilesSorted(
 		return nil, 0, fmt.Errorf("no parquet schema found in %s", entries[0].DataFile().FilePath())
 	}
 
-	sortingColumns, err := rewritePlan.parquetSortingColumns(parquetSchema)
+	outputSchema := parquetSchema
+	var layout *lineageLayout
+	if lineage {
+		if layout, err = newLineageLayout(parquetSchema); err != nil {
+			firstReader.Close()
+			return nil, 0, fmt.Errorf("schema of %s: %w", entries[0].DataFile().FilePath(), err)
+		}
+		outputSchema = layout.schema
+	}
+
+	sortingColumns, err := rewritePlan.parquetSortingColumns(outputSchema)
 	if err != nil {
 		firstReader.Close()
 		return nil, 0, fmt.Errorf("resolve parquet sorting columns: %w", err)
@@ -1249,7 +1356,7 @@ func mergeParquetFilesSorted(
 	}
 
 	var outputBuf bytes.Buffer
-	writer := parquet.NewSortingWriter[any](&outputBuf, sortBufferRows, parquetSchema,
+	writer := parquet.NewSortingWriter[any](&outputBuf, sortBufferRows, outputSchema,
 		parquet.MaxRowsPerRowGroup(rowGroupRows),
 		parquet.SortingWriterConfig(
 			parquet.SortingColumns(sortingColumns...),
@@ -1265,8 +1372,8 @@ func mergeParquetFilesSorted(
 		}
 	}()
 
-	writeRows := func(reader *parquet.Reader, source string) (int64, error) {
-		return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error {
+	writeRows := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) {
+		return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error {
 			// The writer's row buffer copies the values it is handed, so these
 			// rows need no cloning out of the reader's own buffer first.
 			_, writeErr := writer.WriteRows(filtered)
@@ -1274,7 +1381,7 @@ func mergeParquetFilesSorted(
 		})
 	}
 
-	totalRows, err := writeRows(firstReader, entries[0].DataFile().FilePath())
+	totalRows, err := writeRows(firstReader, entries[0])
 	if err != nil {
 		return nil, 0, err
 	}
@@ -1292,12 +1399,12 @@ func mergeParquetFilesSorted(
 		}
 
 		reader := parquet.NewReader(bytes.NewReader(data))
-		if !schemasEqual(parquetSchema, reader.Schema()) {
+		if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) {
 			reader.Close()
 			return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath())
 		}
 
-		rowsWritten, err := writeRows(reader, entry.DataFile().FilePath())
+		rowsWritten, err := writeRows(reader, entry)
 		if err != nil {
 			return nil, 0, err
 		}

weed/worker/tasks/iceberg/delete_rewrite.go

@@ -107,6 +107,9 @@ func collectDeleteRewriteGroups(
 			}
 
 			allPositionEntries = append(allPositionEntries, entry)
+			if entry.DataFile().FileFormat() == iceberg.PuffinFile {
+				continue
+			}
 
 			fileDeletes, err := readPositionDeleteFile(ctx, filerClient, bucketName, dataPath, entry.DataFile().FilePath())
 			if err != nil {
@@ -258,6 +261,8 @@ func writePositionDeleteFile(rows []positionDeleteRow) ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
+const skipDeleteRewriteV3 = "rewrite_position_delete_files skipped: position-delete files must not be added to a format-version 3 table"
+
 func (h *Handler) rewritePositionDeleteFiles(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
@@ -275,6 +280,13 @@ func (h *Handler) rewritePositionDeleteFiles(
 	if currentSnap == nil || currentSnap.ManifestList == "" {
 		return "no current snapshot", nil, nil
 	}
+	// The spec forbids adding position-delete files to a v3 table ("Position
+	// delete files must not be added to v3 tables"; v3 writes deletion
+	// vectors), and this rewrite adds new ones. Compaction still applies and
+	// removes existing position deletes.
+	if hasRowLineage(meta.Version()) {
+		return skipDeleteRewriteV3, nil, nil
+	}
 
 	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
 	if err != nil {
@@ -541,8 +553,10 @@ func (h *Handler) rewritePositionDeleteFiles(
 		allManifests = append(allManifests, mf)
 	}
 
+	firstRowID := rowLineageFirstRowID(meta)
 	var manifestListBuf bytes.Buffer
-	if err := iceberg.WriteManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, 0, allManifests); err != nil {
+	addedRows, err := writeManifestList(version, &manifestListBuf, newSnapID, &snapshotID, &seqNum, firstRowID, allManifests)
+	if err != nil {
 		return "", nil, fmt.Errorf("write delete manifest list: %w", err)
 	}
 	manifestListName := fmt.Sprintf("snap-%d-%s.avro", newSnapID, artifactSuffix)
@@ -553,9 +567,9 @@ func (h *Handler) rewritePositionDeleteFiles(
 
 	manifestListLocation := absoluteIcebergPath(bucketName, dataPath, "metadata", manifestListName)
 	err = h.commitWithRetry(ctx, filerClient, bucketName, tablePath, state.MetadataFileName, config, func(currentMeta table.Metadata, builder *table.MetadataBuilder) error {
-		cs := currentMeta.CurrentSnapshot()
-		if cs == nil || cs.SnapshotID != snapshotID {
-			return errStalePlan
+		cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID)
+		if err != nil {
+			return err
 		}
 		newSnapshot := &table.Snapshot{
 			SnapshotID:       newSnapID,
@@ -574,6 +588,7 @@ func (h *Handler) rewritePositionDeleteFiles(
 				return &id
 			}(),
 		}
+		setRowLineage(newSnapshot, version, firstRowID, addedRows)
 		if err := builder.AddSnapshot(newSnapshot); err != nil {
 			return err
 		}

weed/worker/tasks/iceberg/deletion_vector_test.go

@@ -0,0 +1,375 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+
+	filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// A deletion-vector entry names its data file in referenced_data_file; the
+// match against candidates is on the normalized path either side takes.
+func TestDeletionVectorPathMatching(t *testing.T) {
+	spec := *iceberg.UnpartitionedSpec
+	snapID := int64(1)
+
+	newEntry := func(path, ref string) iceberg.ManifestEntry {
+		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes, path,
+			iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 8)
+		if err != nil {
+			t.Fatalf("build dv file: %v", err)
+		}
+		if ref != "" {
+			dfb.ReferencedDataFile(ref)
+		}
+		return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
+	}
+
+	paths, err := deletionVectorPaths([]iceberg.ManifestEntry{
+		newEntry("metadata/dv.puffin", "s3://test-bucket/ns/tbl/data/f1.parquet"),
+	}, "test-bucket", "ns/tbl")
+	if err != nil {
+		t.Fatalf("deletionVectorPaths: %v", err)
+	}
+	if !paths["ns/tbl/data/f1.parquet"] {
+		t.Fatalf("dv paths = %v, want the normalized f1 path", paths)
+	}
+
+	dataEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+		mustDataFile(t, spec, "data/f1.parquet", 3))
+	otherEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+		mustDataFile(t, spec, "data/f2.parquet", 2))
+	kept, err := excludeDeletionVectorFiles([]iceberg.ManifestEntry{dataEntry, otherEntry}, paths, "test-bucket", "ns/tbl")
+	if err != nil {
+		t.Fatalf("excludeDeletionVectorFiles: %v", err)
+	}
+	if len(kept) != 1 || kept[0].DataFile().FilePath() != "data/f2.parquet" {
+		t.Fatalf("kept %d entries, want only f2", len(kept))
+	}
+}
+
+// A Puffin deletion vector is not a Parquet position-delete file: it must not
+// reach readPositionDeleteFile, and the data file it covers must stay
+// uncompacted until something applies the vector. Other files still merge and
+// ordinary Parquet position deletes still apply.
+func TestCompactDataFilesDeletionVector(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	ctx := context.Background()
+
+	setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"}
+	spec := *iceberg.UnpartitionedSpec
+	snapID := int64(1)
+
+	putDataFileRows(t, fs, "f1.parquet", []lineageDataRow{{1, "a"}, {2, "b"}})
+	putDataFileRows(t, fs, "f2.parquet", []lineageDataRow{{3, "c"}, {4, "d"}})
+	putDataFileRows(t, fs, "f3.parquet", []lineageDataRow{{5, "e"}})
+
+	dataEntries := []iceberg.ManifestEntry{
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, "data/f1.parquet", 2)),
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, "data/f2.parquet", 2)),
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, "data/f3.parquet", 1)),
+	}
+
+	// An ordinary Parquet position delete against f2's first row.
+	type posRow struct {
+		FilePath string `parquet:"file_path"`
+		Pos      int64  `parquet:"pos"`
+	}
+	var pdBuf bytes.Buffer
+	w := parquet.NewWriter(&pdBuf, parquet.SchemaOf(new(posRow)))
+	if err := w.Write(&posRow{"data/f2.parquet", 0}); err != nil {
+		t.Fatalf("write pos delete: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close pos delete: %v", err)
+	}
+	dataDir := "/buckets/test-bucket/ns/tbl/data"
+	fs.putEntry(dataDir, "pd1.parquet", &filer_pb.Entry{Name: "pd1.parquet", Content: pdBuf.Bytes()})
+	// The vector itself is never opened, but the file should exist like a real
+	// engine's write would leave it.
+	fs.putEntry(dataDir, "dv.puffin", &filer_pb.Entry{Name: "dv.puffin", Content: []byte("puffin-bytes")})
+
+	posDF, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes,
+		setup.fileRef("data", "pd1.parquet"), iceberg.ParquetFile, map[int]any{}, nil, nil, 1, int64(pdBuf.Len()))
+	if err != nil {
+		t.Fatalf("build pos delete file: %v", err)
+	}
+	dvDF, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes,
+		setup.fileRef("data", "dv.puffin"), iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 12)
+	if err != nil {
+		t.Fatalf("build dv file: %v", err)
+	}
+	dvDF.ReferencedDataFile("s3://test-bucket/ns/tbl/data/f1.parquet")
+	deleteEntries := []iceberg.ManifestEntry{
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, posDF.Build()),
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dvDF.Build()),
+	}
+
+	writeV3Table(t, fs, setup, dataEntries, deleteEntries, 0, 5)
+
+	handler := NewHandler(nil)
+	config := Config{
+		TargetFileSizeBytes: 4096,
+		MinInputFiles:       2,
+		ApplyDeletes:        true,
+	}
+	result, _, err := handler.compactDataFiles(ctx, client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	if !strings.Contains(result, "compacted 2 files into 1") {
+		t.Fatalf("unexpected result: %q", result)
+	}
+
+	state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	manifests, err := loadCurrentManifests(ctx, client, setup.BucketName, state.DataPath, state.Metadata)
+	if err != nil {
+		t.Fatalf("loadCurrentManifests: %v", err)
+	}
+
+	var dataMfs, deleteMfs []iceberg.ManifestFile
+	for _, mf := range manifests {
+		if mf.ManifestContent() == iceberg.ManifestContentData {
+			dataMfs = append(dataMfs, mf)
+		} else {
+			deleteMfs = append(deleteMfs, mf)
+		}
+	}
+	if len(deleteMfs) != 1 {
+		t.Fatalf("delete manifests carried = %d, want 1 (the vector still applies to f1)", len(deleteMfs))
+	}
+
+	var merged iceberg.DataFile
+	var f1Live bool
+	for _, mf := range dataMfs {
+		manifestData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, mf.FilePath())
+		if err != nil {
+			t.Fatalf("read manifest: %v", err)
+		}
+		entries, err := s3tables.ReadManifest(mf, manifestData, false, specByID(state.Metadata), state.Metadata.CurrentSchema())
+		if err != nil {
+			t.Fatalf("parse manifest: %v", err)
+		}
+		for _, e := range entries {
+			df := e.DataFile()
+			switch {
+			case e.Status() == iceberg.EntryStatusADDED:
+				merged = df
+			case e.Status() == iceberg.EntryStatusEXISTING && df.FilePath() == "data/f1.parquet":
+				f1Live = true
+			}
+		}
+	}
+	if !f1Live {
+		t.Fatal("f1 is covered by a deletion vector and must stay live")
+	}
+	if merged == nil {
+		t.Fatal("no merged file added")
+	}
+
+	mergedData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, merged.FilePath())
+	if err != nil {
+		t.Fatalf("read merged file: %v", err)
+	}
+	ids, _, rowIDs, _ := mergedColumns(t, mergedData)
+	if len(ids) != 2 || ids[0] != int64(4) || ids[1] != int64(5) {
+		t.Fatalf("merged ids = %v, want [4 5] (f2 minus its deleted row, f3)", ids)
+	}
+	if rowIDs[0] != int64(3) || rowIDs[1] != int64(4) {
+		t.Fatalf("merged _row_id = %v, want [3 4]", rowIDs)
+	}
+}
+
+// Detection applies the same exclusion compaction does: a table whose only
+// binable files are all covered by deletion vectors is not a candidate, while
+// unaffected files still qualify.
+func TestDetectSchedulesDeletionVector(t *testing.T) {
+	spec := *iceberg.UnpartitionedSpec
+	snapID := int64(1)
+
+	dataEntry := func(name string, count int64) iceberg.ManifestEntry {
+		return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, name, count))
+	}
+	dvEntry := func(t *testing.T, ref string) iceberg.ManifestEntry {
+		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentPosDeletes,
+			"data/dv.puffin", iceberg.PuffinFile, map[int]any{}, nil, nil, 0, 12)
+		if err != nil {
+			t.Fatalf("build dv file: %v", err)
+		}
+		dfb.ReferencedDataFile(ref)
+		return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
+	}
+
+	config := Config{
+		SnapshotRetentionMs: hoursToMs(24 * 365),
+		MaxSnapshotsToKeep:  10,
+		TargetFileSizeBytes: 4096,
+		MinInputFiles:       2,
+		Operations:          "compact",
+		ApplyDeletes:        true,
+	}
+
+	t.Run("unaffected files stay eligible", func(t *testing.T) {
+		fs, client := startFakeFiler(t)
+		setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"}
+		writeV3Table(t, fs, setup,
+			[]iceberg.ManifestEntry{dataEntry("data/f1.parquet", 1), dataEntry("data/f2.parquet", 1), dataEntry("data/f3.parquet", 1)},
+			[]iceberg.ManifestEntry{dvEntry(t, "data/f1.parquet")},
+			0, 3)
+
+		tables, err := NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
+		if err != nil {
+			t.Fatalf("scanTablesForMaintenance: %v", err)
+		}
+		if len(tables) != 1 {
+			t.Fatalf("expected 1 compaction candidate (f2+f3 still binable), got %d", len(tables))
+		}
+	})
+
+	t.Run("all binable files covered", func(t *testing.T) {
+		fs, client := startFakeFiler(t)
+		setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"}
+		writeV3Table(t, fs, setup,
+			[]iceberg.ManifestEntry{dataEntry("data/f1.parquet", 1), dataEntry("data/f2.parquet", 1)},
+			[]iceberg.ManifestEntry{dvEntry(t, "data/f1.parquet")},
+			0, 2)
+
+		tables, err := NewHandler(nil).scanTablesForMaintenance(context.Background(), client, config, "", "", "", 0)
+		if err != nil {
+			t.Fatalf("scanTablesForMaintenance: %v", err)
+		}
+		if len(tables) != 0 {
+			t.Fatalf("expected no candidates (only the covered f1 and the lone f2 remain), got %d", len(tables))
+		}
+	})
+}
+
+// writeV3Table stores a format-version-3 table in the fake filer: a data
+// manifest whose entries carry no first_row_id (so readers exercise manifest
+// first_row_id inheritance), an optional delete manifest, and the manifest
+// list + snapshot that commit them at sequence number 1.
+func writeV3Table(t *testing.T, fs *fakeFilerServer, setup tableSetup,
+	dataManifestEntries, deleteManifestEntries []iceberg.ManifestEntry,
+	listFirstRowID, addedRows int64) table.Metadata {
+	t.Helper()
+
+	schema := newTestSchema()
+	spec := *iceberg.UnpartitionedSpec
+	location := "s3://" + setup.BucketName + "/" + setup.dataPath()
+
+	base, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder, location, nil)
+	if err != nil {
+		t.Fatalf("new metadata: %v", err)
+	}
+	builder, err := table.MetadataBuilderFromBase(base, location)
+	if err != nil {
+		t.Fatalf("metadata builder: %v", err)
+	}
+	if err := builder.SetFormatVersion(3); err != nil {
+		t.Fatalf("set format version: %v", err)
+	}
+	if _, err := builder.Build(); err != nil {
+		t.Fatalf("build v3 metadata: %v", err)
+	}
+
+	snapID := int64(1)
+	var manifestBuf bytes.Buffer
+	dataMf, err := iceberg.WriteManifest(setup.fileRef("metadata", "m0.avro"), &manifestBuf, 3, spec, schema, snapID, dataManifestEntries)
+	if err != nil {
+		t.Fatalf("write manifest: %v", err)
+	}
+	metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.dataPath(), "metadata")
+	fs.putEntry(metaDir, "m0.avro", &filer_pb.Entry{
+		Name:    "m0.avro",
+		Content: manifestBuf.Bytes(),
+	})
+
+	manifests := []iceberg.ManifestFile{dataMf}
+	if len(deleteManifestEntries) > 0 {
+		delMf, delBytes := writeDeleteManifestForTest(
+			t, setup.fileRef("metadata", "del.avro"), 3, spec, schema, snapID, deleteManifestEntries)
+		fs.putEntry(metaDir, "del.avro", &filer_pb.Entry{
+			Name:    "del.avro",
+			Content: delBytes,
+		})
+		manifests = append(manifests, delMf)
+	}
+
+	var listBuf bytes.Buffer
+	seqNum := int64(1)
+	if err := iceberg.WriteManifestList(3, &listBuf, snapID, nil, &seqNum, listFirstRowID, manifests); err != nil {
+		t.Fatalf("write manifest list: %v", err)
+	}
+	fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{
+		Name:    "snap-1.avro",
+		Content: listBuf.Bytes(),
+	})
+
+	snap := table.Snapshot{
+		SnapshotID:     snapID,
+		SequenceNumber: seqNum,
+		TimestampMs:    time.Now().UnixMilli(),
+		ManifestList:   setup.fileRef("metadata", "snap-1.avro"),
+		FirstRowID:     ptrInt64(listFirstRowID),
+		AddedRows:      &addedRows,
+	}
+	if err := builder.AddSnapshot(&snap); err != nil {
+		t.Fatalf("add snapshot: %v", err)
+	}
+	if err := builder.SetSnapshotRef(table.MainBranch, snapID, table.BranchRef); err != nil {
+		t.Fatalf("set snapshot ref: %v", err)
+	}
+	meta, err := builder.Build()
+	if err != nil {
+		t.Fatalf("build metadata with snapshot: %v", err)
+	}
+
+	fullMetadataJSON, err := json.Marshal(meta)
+	if err != nil {
+		t.Fatalf("marshal metadata: %v", err)
+	}
+	internalMeta := map[string]interface{}{
+		"metadataVersion":  1,
+		"metadataLocation": setup.fileRef("metadata", "v1.metadata.json"),
+		"metadata":         map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)},
+	}
+	xattr, err := json.Marshal(internalMeta)
+	if err != nil {
+		t.Fatalf("marshal xattr: %v", err)
+	}
+	bucketPath := path.Join(s3tables.TablesPath, setup.BucketName)
+	fs.putEntry(s3tables.TablesPath, setup.BucketName, &filer_pb.Entry{
+		Name:        setup.BucketName,
+		IsDirectory: true,
+		Extended:    map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")},
+	})
+	fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{Name: setup.Namespace, IsDirectory: true})
+	fs.putEntry(path.Join(bucketPath, setup.Namespace), setup.TableName, &filer_pb.Entry{
+		Name:        setup.TableName,
+		IsDirectory: true,
+		Extended: map[string][]byte{
+			s3tables.ExtendedKeyMetadata:        xattr,
+			s3tables.ExtendedKeyMetadataVersion: metadataVersionXattr(1),
+		},
+	})
+	return meta
+}
+
+func ptrInt64(v int64) *int64 { return &v }

weed/worker/tasks/iceberg/detection.go

@@ -308,6 +308,9 @@ func (h *Handler) tableNeedsMaintenance(
 				return true, nil
 			}
 		case "rewrite_position_delete_files":
+			if hasRowLineage(meta.Version()) {
+				continue // the rewrite would add position-delete files, which a v3 table must not get (skipDeleteRewriteV3)
+			}
 			manifests, err := getCurrentManifests()
 			if err != nil {
 				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
@@ -441,6 +444,7 @@ func hasEligibleCompaction(
 	schema := meta.CurrentSchema()
 
 	var allEntries []iceberg.ManifestEntry
+	var dvPaths map[string]bool
 	for _, mf := range dataManifests {
 		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
 		if err != nil {
@@ -453,10 +457,35 @@ func hasEligibleCompaction(
 		allEntries = append(allEntries, entries...)
 	}
 
-	candidateEntries := allEntries
+	// Deletion vectors exist only from format version 3.
+	if hasRowLineage(meta.Version()) {
+		var deleteEntries []iceberg.ManifestEntry
+		for _, mf := range manifests {
+			if mf.ManifestContent() != iceberg.ManifestContentDeletes {
+				continue
+			}
+			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
+			if err != nil {
+				return false, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err)
+			}
+			entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema)
+			if err != nil {
+				return false, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), err)
+			}
+			deleteEntries = append(deleteEntries, entries...)
+		}
+		if dvPaths, err = deletionVectorPaths(deleteEntries, bucketName, dataPath); err != nil {
+			return false, err
+		}
+	}
+
+	candidateEntries, err := excludeDeletionVectorFiles(allEntries, dvPaths, bucketName, dataPath)
+	if err != nil {
+		return false, err
+	}
 	if predicate != nil {
-		candidateEntries = make([]iceberg.ManifestEntry, 0, len(allEntries))
-		for _, entry := range allEntries {
+		filtered := make([]iceberg.ManifestEntry, 0, len(candidateEntries))
+		for _, entry := range candidateEntries {
 			spec, ok := specsByID[int(entry.DataFile().SpecID())]
 			if !ok {
 				continue
@@ -466,9 +495,10 @@ func hasEligibleCompaction(
 				return false, err
 			}
 			if match {
-				candidateEntries = append(candidateEntries, entry)
+				filtered = append(filtered, entry)
 			}
 		}
+		candidateEntries = filtered
 	}
 
 	rewritePlan, err := resolveCompactionRewritePlan(config, meta)

weed/worker/tasks/iceberg/handler_test.go

@@ -1303,7 +1303,7 @@ func TestMergeParquetFilesWithPositionDeletes(t *testing.T) {
 
 	merged, count, err := mergeParquetFiles(
 		context.Background(), client, "test-bucket", "ns/tbl",
-		entries, posDeletes, nil, nil, 0,
+		entries, posDeletes, nil, nil, 0, false,
 	)
 	if err != nil {
 		t.Fatalf("mergeParquetFiles: %v", err)
@@ -1419,7 +1419,7 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) {
 
 	merged, count, err := mergeParquetFiles(
 		context.Background(), client, "test-bucket", "ns/tbl",
-		entries, nil, eqGroups, schema, 0,
+		entries, nil, eqGroups, schema, 0, false,
 	)
 	if err != nil {
 		t.Fatalf("mergeParquetFiles: %v", err)
@@ -1479,7 +1479,7 @@ func TestMergeParquetFilesDictionaryEncodedInput(t *testing.T) {
 
 	merged, count, err := mergeParquetFiles(
 		context.Background(), client, "test-bucket", "ns/tbl",
-		entries, nil, nil, nil, 0,
+		entries, nil, nil, nil, 0, false,
 	)
 	if err != nil {
 		t.Fatalf("mergeParquetFiles: %v", err)

weed/worker/tasks/iceberg/operations.go

@@ -566,8 +566,11 @@ func (h *Handler) rewriteManifests(
 		}
 	}
 
+	// v3 row lineage (row_lineage.go): the merged manifests are assigned first_row_id from the table's next-row-id; the
+	// data files keep the first_row_id they carry, so no row is renumbered
+	firstRowID := rowLineageFirstRowID(meta)
 	var manifestListBuf bytes.Buffer
-	err = iceberg.WriteManifestList(version, &manifestListBuf, newSnapshotID, &snapshotID, &newSeqNum, 0, newManifests)
+	addedRows, err := writeManifestList(version, &manifestListBuf, newSnapshotID, &snapshotID, &newSeqNum, firstRowID, newManifests)
 	if err != nil {
 		return "", nil, fmt.Errorf("write manifest list: %w", err)
 	}
@@ -586,9 +589,9 @@ func (h *Handler) rewriteManifests(
 		// Guard: verify table head hasn't advanced since we planned.
 		// The merged manifest and manifest list were built against snapshotID;
 		// if the head moved, they reference stale state.
-		cs := currentMeta.CurrentSnapshot()
-		if cs == nil || cs.SnapshotID != snapshotID {
-			return errStalePlan
+		cs, err := checkCommitPlan(currentMeta, snapshotID, version, firstRowID)
+		if err != nil {
+			return err
 		}
 
 		newSnapshot := &table.Snapshot{
@@ -604,6 +607,7 @@ func (h *Handler) rewriteManifests(
 				return &id
 			}(),
 		}
+		setRowLineage(newSnapshot, version, firstRowID, addedRows)
 		if err := builder.AddSnapshot(newSnapshot); err != nil {
 			return err
 		}

weed/worker/tasks/iceberg/row_lineage.go

@@ -0,0 +1,393 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"reflect"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+)
+
+// Row lineage (Iceberg v3, spec "Row Lineage" and "Snapshot Row IDs").
+//
+// Every snapshot committed to a v3 table carries first-row-id, the table's next-row-id when the commit is attempted,
+// and added-rows, the upper bound of the row IDs its manifest list assigns; committing advances next-row-id by
+// added-rows. iceberg-go refuses a v3 snapshot without them ("invalid row lineage: first-row-id is required for v3
+// snapshots"), which is what every maintenance commit on a v3 table hit. The manifest list assigns first_row_id to the
+// data manifests that have none, starting at the snapshot's first-row-id; passing 0 there (as the maintenance code did)
+// would hand rows IDs other rows already hold.
+//
+// A rewrite that moves rows into a new data file (compaction) copies each row's _row_id and
+// _last_updated_sequence_number into it: the value the file materializes, else the one inheritance gives it
+// (the input file's first_row_id plus the row's position; the input entry's data sequence number). A compacted row keeps
+// its identity, and a rewrite that modifies no row keeps its last-updated sequence number.
+
+const minFormatVersionRowLineage = 3
+
+func hasRowLineage(version int) bool { return version >= minFormatVersionRowLineage }
+
+// rowLineageFirstRowID is the first-row-id a snapshot planned on meta starts at (0 below v3, where it is unused).
+func rowLineageFirstRowID(meta table.Metadata) int64 {
+	if !hasRowLineage(meta.Version()) {
+		return 0
+	}
+	return meta.NextRowID()
+}
+
+// checkRowLineagePlan: a plan that assigned row IDs from next-row-id `planned` is stale once the table's next-row-id
+// moved, even if the head did not (a commit to another branch advances it too); the spec has first-row-id reassigned on
+// every attempt, which for a manifest list already written means planning again.
+//
+// A format-version change since planning (an upgrade to v3) is stale too: the manifests and the snapshot were built for
+// the old version.
+func checkRowLineagePlan(plannedVersion, currentVersion int, planned, current int64) error {
+	if currentVersion != plannedVersion {
+		return fmt.Errorf("%w: format version changed from %d to %d", errStalePlan, plannedVersion, currentVersion)
+	}
+	if hasRowLineage(plannedVersion) && planned != current {
+		return fmt.Errorf("%w: next-row-id moved from %d to %d", errStalePlan, planned, current)
+	}
+	return nil
+}
+
+// checkCommitPlan is what every maintenance commit re-checks against the metadata it commits onto: the head it planned
+// against, and the row-lineage state (format version, next-row-id) its manifest list was written for.
+func checkCommitPlan(currentMeta table.Metadata, snapshotID int64, version int, firstRowID int64) (*table.Snapshot, error) {
+	cs := currentMeta.CurrentSnapshot()
+	if cs == nil || cs.SnapshotID != snapshotID {
+		return nil, errStalePlan
+	}
+	if err := checkRowLineagePlan(version, currentMeta.Version(), firstRowID, rowLineageFirstRowID(currentMeta)); err != nil {
+		return nil, err
+	}
+	return cs, nil
+}
+
+// writeManifestList writes the snapshot's manifest list and returns its added-rows for a v3 table (nil below v3): the row
+// IDs the writer assigned to manifests without a first_row_id, starting at firstRowID.
+func writeManifestList(version int, out io.Writer, snapshotID int64, parentSnapshotID, sequenceNumber *int64, firstRowID int64, manifests []iceberg.ManifestFile) (*int64, error) {
+	if !hasRowLineage(version) {
+		return nil, iceberg.WriteManifestList(version, out, snapshotID, parentSnapshotID, sequenceNumber, firstRowID, manifests)
+	}
+	if sequenceNumber == nil {
+		return nil, fmt.Errorf("sequence number is required for v%d tables", version)
+	}
+	writer, err := iceberg.NewManifestListWriterV3(out, snapshotID, *sequenceNumber, firstRowID, parentSnapshotID)
+	if err != nil {
+		return nil, err
+	}
+	if err := writer.AddManifests(manifests); err != nil {
+		writer.Close()
+		return nil, err
+	}
+	if err := writer.Close(); err != nil {
+		return nil, err
+	}
+	next := writer.NextRowID()
+	if next == nil || *next < firstRowID {
+		return nil, fmt.Errorf("manifest list writer reported next-row-id %v below first-row-id %d", next, firstRowID)
+	}
+	addedRows := *next - firstRowID
+	return &addedRows, nil
+}
+
+// setRowLineage puts first-row-id and added-rows on a snapshot of a v3 table.
+func setRowLineage(snapshot *table.Snapshot, version int, firstRowID int64, addedRows *int64) {
+	if !hasRowLineage(version) || addedRows == nil {
+		return
+	}
+	first, added := firstRowID, *addedRows
+	snapshot.FirstRowID, snapshot.AddedRows = &first, &added
+}
+
+// ---------------------------------------------------------------------------------------------- compaction rows
+
+// lineageSource is what inheritance gives the rows of one input data file.
+type lineageSource struct {
+	firstRowID *int64 // the file's first_row_id; nil: its rows have no row ID to carry (a null is written)
+	dataSeq    *int64 // the data sequence number of the file's manifest entry
+}
+
+func lineageSourceOf(entry iceberg.ManifestEntry) lineageSource {
+	src := lineageSource{dataSeq: manifestEntrySeqNum(entry)}
+	if id := entry.DataFile().FirstRowID(); id != nil {
+		v := *id
+		src.firstRowID = &v
+	}
+	return src
+}
+
+// lineageKind: 1 for _row_id, 2 for _last_updated_sequence_number, 0 for a data column. A column is a lineage column by
+// its reserved field id, or by its reserved name when it has no field id; a reserved name under another field id is an
+// error (kept as data, the output would have two columns of that name).
+func lineageKind(f parquet.Field) (int, error) {
+	switch f.ID() {
+	case iceberg.RowIDFieldID:
+		return 1, nil
+	case iceberg.LastUpdatedSequenceNumberFieldID:
+		return 2, nil
+	}
+	named := 0
+	switch f.Name() {
+	case iceberg.RowIDColumnName:
+		named = 1
+	case iceberg.LastUpdatedSequenceNumberColumnName:
+		named = 2
+	}
+	if named != 0 && f.ID() != 0 {
+		return 0, fmt.Errorf("column %s has field id %d, not the reserved id of that metadata column", f.Name(), f.ID())
+	}
+	return named, nil
+}
+
+// lineageLayout is the output of a v3 compaction: the reference file's own columns in their order (any lineage columns
+// it has set aside), then _row_id and _last_updated_sequence_number, optional longs with the spec's field ids.
+type lineageLayout struct {
+	schema      *parquet.Schema
+	reference   *parquet.Schema // the schema the layout was built from (the bin's first file)
+	baseFields  []parquet.Field
+	baseOffsets []int // output leaf index of each base field's first leaf
+	baseLeaves  int
+}
+
+func newLineageLayout(reference *parquet.Schema) (*lineageLayout, error) {
+	var base []parquet.Field
+	var offsets []int
+	leaves := 0
+	for _, f := range reference.Fields() {
+		kind, err := lineageKind(f)
+		if err != nil {
+			return nil, err
+		}
+		if kind != 0 {
+			continue
+		}
+		base = append(base, f)
+		offsets = append(offsets, leaves)
+		leaves += leafCount(f)
+	}
+	fields := append(append([]parquet.Field{}, base...),
+		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.RowIDFieldID), name: iceberg.RowIDColumnName},
+		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName},
+	)
+	return &lineageLayout{
+		schema:      parquet.NewSchema(reference.Name(), &orderedGroup{Node: reference, fields: fields}),
+		reference:   reference,
+		baseFields:  base,
+		baseOffsets: offsets,
+		baseLeaves:  leaves,
+	}, nil
+}
+
+// mapperFor matches an input file to the layout: its non-lineage columns must be the reference's, in the same order
+// (what the merge required of whole schemas before); its lineage columns, if any, are read wherever they are.
+func (l *lineageLayout) mapperFor(input *parquet.Schema, src lineageSource) (*lineageMapper, error) {
+	m := &lineageMapper{rowIDLeaf: -1, seqLeaf: -1, outRowID: l.baseLeaves, outSeq: l.baseLeaves + 1, src: src}
+	next, base := 0, 0
+	for _, f := range input.Fields() {
+		n := leafCount(f)
+		kind, err := lineageKind(f)
+		if err != nil {
+			return nil, err
+		}
+		if kind != 0 {
+			if n != 1 || !f.Leaf() {
+				return nil, fmt.Errorf("lineage column %s is not a primitive", f.Name())
+			}
+			target := &m.rowIDLeaf
+			if kind == 2 {
+				target = &m.seqLeaf
+			}
+			if *target >= 0 {
+				return nil, fmt.Errorf("more than one %s column", f.Name())
+			}
+			*target = next
+			m.leafMap = append(m.leafMap, -1)
+			next++
+			continue
+		}
+		if base >= len(l.baseFields) || f.Name() != l.baseFields[base].Name() || !parquet.EqualNodes(f, l.baseFields[base]) {
+			return nil, fmt.Errorf("cannot merge files with different schemas")
+		}
+		for i := 0; i < n; i++ {
+			m.leafMap = append(m.leafMap, l.baseOffsets[base]+i)
+		}
+		next += n
+		base++
+	}
+	if base != len(l.baseFields) {
+		return nil, fmt.Errorf("cannot merge files with different schemas")
+	}
+	return m, nil
+}
+
+// lineageMapper turns one input file's rows into the layout's rows.
+type lineageMapper struct {
+	leafMap          []int // input leaf index -> output leaf index; -1 for an input lineage column
+	rowIDLeaf        int   // input leaf index of _row_id, -1 when the file has none
+	seqLeaf          int   // input leaf index of _last_updated_sequence_number, -1 when the file has none
+	outRowID, outSeq int
+	src              lineageSource
+	rows             []parquet.Row   // reused per batch: both writers copy the values they are handed
+	values           []parquet.Value // backing store of rows
+}
+
+// mapRows maps rows read at the given positions of the input file (their position before any delete applied).
+func (m *lineageMapper) mapRows(rows []parquet.Row, positions []int64) []parquet.Row {
+	need := 0
+	for _, row := range rows {
+		need += len(row) + 2
+	}
+	if cap(m.values) < need {
+		m.values = make([]parquet.Value, 0, need)
+	}
+	m.values = m.values[:0]
+	m.rows = m.rows[:0]
+	for i, row := range rows {
+		start := len(m.values)
+		mapped := m.values[start:start:cap(m.values)]
+		var rowID, seq *parquet.Value
+		for j := range row {
+			v := row[j]
+			switch c := v.Column(); {
+			case c == m.rowIDLeaf:
+				rowID = &row[j]
+			case c == m.seqLeaf:
+				seq = &row[j]
+			default:
+				mapped = append(mapped, v.Level(v.RepetitionLevel(), v.DefinitionLevel(), m.leafMap[c]))
+			}
+		}
+		var inheritedRowID *int64
+		if m.src.firstRowID != nil {
+			id := *m.src.firstRowID + positions[i]
+			inheritedRowID = &id
+		}
+		mapped = append(mapped, lineageValue(rowID, inheritedRowID, m.outRowID), lineageValue(seq, m.src.dataSeq, m.outSeq))
+		m.values = m.values[:start+len(mapped)]
+		m.rows = append(m.rows, parquet.Row(mapped))
+	}
+	return m.rows
+}
+
+// lineageValue: the value the file materializes when it is not null, else the inherited one, else a null.
+func lineageValue(materialized *parquet.Value, inherited *int64, column int) parquet.Value {
+	if materialized != nil && !materialized.IsNull() {
+		return parquet.Int64Value(materialized.Int64()).Level(0, 1, column)
+	}
+	if inherited != nil {
+		return parquet.Int64Value(*inherited).Level(0, 1, column)
+	}
+	return parquet.NullValue().Level(0, 0, column)
+}
+
+func leafCount(n parquet.Node) int {
+	if n.Leaf() {
+		return 1
+	}
+	total := 0
+	for _, f := range n.Fields() {
+		total += leafCount(f)
+	}
+	return total
+}
+
+// orderedGroup is a group node with the fields given, in that order (parquet.Group sorts its fields by name).
+type orderedGroup struct {
+	parquet.Node
+	fields []parquet.Field
+}
+
+func (g *orderedGroup) Fields() []parquet.Field { return g.fields }
+func (g *orderedGroup) Leaf() bool              { return false }
+
+// GoType: rows are written as parquet.Row, never from Go values, so the base's type is never read for these columns.
+func (g *orderedGroup) GoType() reflect.Type { return g.Node.GoType() }
+
+// lineageField names a lineage column in the layout.
+type lineageField struct {
+	parquet.Node
+	name string
+}
+
+func (f lineageField) Name() string                           { return f.name }
+func (f lineageField) Value(base reflect.Value) reflect.Value { return reflect.Value{} }
+
+// prepareLineageInput: below v3 (layout nil) the file is merged as it is, with the equality-delete columns resolved on
+// the first file. On v3 the file is mapped onto the layout, and its equality-delete columns are resolved on its own
+// schema, since a file's lineage columns may sit anywhere among its columns.
+func prepareLineageInput(layout *lineageLayout, reader *parquet.Reader, entry iceberg.ManifestEntry, firstFileEqGroups []resolvedEqDeleteGroup,
+	eqDeleteGroups []equalityDeleteGroup, icebergSchema *iceberg.Schema) (*lineageMapper, []resolvedEqDeleteGroup, error) {
+	if layout == nil {
+		return nil, firstFileEqGroups, nil
+	}
+	source := entry.DataFile().FilePath()
+	mapper, err := layout.mapperFor(reader.Schema(), lineageSourceOf(entry))
+	if err != nil {
+		return nil, nil, fmt.Errorf("schema mismatch in %s: %w", source, err)
+	}
+	if reader.Schema() == layout.reference {
+		return mapper, firstFileEqGroups, nil // resolved on this very schema already
+	}
+	eqGroups, err := resolveEqualityDeleteGroupsForSchema(reader.Schema(), eqDeleteGroups, icebergSchema)
+	if err != nil {
+		return nil, nil, fmt.Errorf("resolve equality columns in %s: %w", source, err)
+	}
+	return mapper, eqGroups, nil
+}
+
+// copyFileRows drains one input file into write: deletes filtered and, on a v3 table (layout set), mapped onto the
+// layout with each row's lineage. Both merges, bin-packed and sorted, use it.
+func copyFileRows(ctx context.Context, reader *parquet.Reader, entry iceberg.ManifestEntry, layout *lineageLayout, bucketName, dataPath string,
+	positionDeletes map[string][]int64, firstFileEqGroups []resolvedEqDeleteGroup, eqDeleteGroups []equalityDeleteGroup,
+	icebergSchema *iceberg.Schema, write func([]parquet.Row) error) (int64, error) {
+	mapper, eqGroups, err := prepareLineageInput(layout, reader, entry, firstFileEqGroups, eqDeleteGroups, icebergSchema)
+	if err != nil {
+		reader.Close()
+		return 0, err
+	}
+	return visitFilteredParquetRows(ctx, reader, entry.DataFile().FilePath(), bucketName, dataPath, positionDeletes, eqGroups, func(filtered []parquet.Row, positions []int64) error {
+		if mapper != nil {
+			filtered = mapper.mapRows(filtered, positions)
+		}
+		return write(filtered)
+	})
+}
+
+// mergedFileFirstRowID is the _row_id of the merged output's first row, or nil when it carries none: the value the
+// manifest entry's first_row_id must name, since rewritten rows keep their IDs rather than taking fresh ones.
+func mergedFileFirstRowID(data []byte) (*int64, error) {
+	reader := parquet.NewReader(bytes.NewReader(data))
+	defer reader.Close()
+	schema := reader.Schema()
+	if schema == nil {
+		return nil, nil
+	}
+	leaf, ok := schema.Lookup(iceberg.RowIDColumnName)
+	if !ok {
+		return nil, nil
+	}
+	rows := make([]parquet.Row, 1)
+	n, err := reader.ReadRows(rows)
+	if err != nil && err != io.EOF {
+		return nil, err
+	}
+	if n == 0 {
+		return nil, nil
+	}
+	for _, v := range rows[0] {
+		if v.Column() != leaf.ColumnIndex {
+			continue
+		}
+		if v.IsNull() {
+			return nil, nil
+		}
+		id := v.Int64()
+		return &id, nil
+	}
+	return nil, nil
+}

weed/worker/tasks/iceberg/row_lineage_test.go

@@ -0,0 +1,384 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/parquet-go/parquet-go"
+
+	filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+type lineageDataRow struct {
+	ID   int64  `parquet:"id"`
+	Name string `parquet:"name"`
+}
+
+func putDataFileRows[T any](t *testing.T, fs *fakeFilerServer, name string, rows []T) {
+	t.Helper()
+	var buf bytes.Buffer
+	w := parquet.NewWriter(&buf, parquet.SchemaOf(new(T)))
+	for _, r := range rows {
+		if err := w.Write(&r); err != nil {
+			t.Fatalf("write row: %v", err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	fs.putEntry("/buckets/test-bucket/ns/tbl/data", name, &filer_pb.Entry{
+		Name:    name,
+		Content: buf.Bytes(),
+	})
+}
+
+func lineageEntry(t *testing.T, path string, firstRowID, dataSeq, recordCount int64) iceberg.ManifestEntry {
+	t.Helper()
+	dfb, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+		path, iceberg.ParquetFile, map[int]any{}, nil, nil, recordCount, 1024)
+	if err != nil {
+		t.Fatalf("build data file: %v", err)
+	}
+	snapID := int64(1)
+	return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &dataSeq, nil,
+		dfb.FirstRowID(firstRowID).Build())
+}
+
+// mergedColumns returns each merged row's (id, name, _row_id,
+// _last_updated_sequence_number): the two lineage columns the v3 merge
+// appends after the data columns.
+func mergedColumns(t *testing.T, data []byte) (ids, names, rowIDs, seqs []any) {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open merged: %v", err)
+	}
+	fields := f.Schema().Fields()
+	if len(fields) != 4 {
+		t.Fatalf("merged schema has %d fields, want 4 (id, name, _row_id, _last_updated_sequence_number)", len(fields))
+	}
+	if fields[2].Name() != iceberg.RowIDColumnName || fields[2].ID() != iceberg.RowIDFieldID {
+		t.Fatalf("field 2 = %s id %d, want %s id %d", fields[2].Name(), fields[2].ID(), iceberg.RowIDColumnName, iceberg.RowIDFieldID)
+	}
+	if fields[3].Name() != iceberg.LastUpdatedSequenceNumberColumnName || fields[3].ID() != iceberg.LastUpdatedSequenceNumberFieldID {
+		t.Fatalf("field 3 = %s id %d, want %s id %d", fields[3].Name(), fields[3].ID(), iceberg.LastUpdatedSequenceNumberColumnName, iceberg.LastUpdatedSequenceNumberFieldID)
+	}
+	for _, rg := range f.RowGroups() {
+		rows := rg.Rows()
+		buf := make([]parquet.Row, 8)
+		for {
+			n, err := rows.ReadRows(buf)
+			for _, row := range buf[:n] {
+				ids = append(ids, row[0].Int64())
+				names = append(names, row[1].String())
+				if row[2].IsNull() {
+					rowIDs = append(rowIDs, nil)
+				} else {
+					rowIDs = append(rowIDs, row[2].Int64())
+				}
+				if row[3].IsNull() {
+					seqs = append(seqs, nil)
+				} else {
+					seqs = append(seqs, row[3].Int64())
+				}
+			}
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				t.Fatalf("read merged rows: %v", err)
+			}
+		}
+		rows.Close()
+	}
+	return
+}
+
+// On a v3 table the merged file carries every row's identity: the
+// materialized value when the input has it, else the one inheritance
+// gives it (the input file's first_row_id plus the row's position, the
+// entry's data sequence number).
+func TestMergeParquetFilesRowLineage(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	putDataFileRows(t, fs, "file1.parquet", []lineageDataRow{{1, "alice"}, {2, "bob"}, {3, "carl"}})
+	putDataFileRows(t, fs, "file2.parquet", []lineageDataRow{{4, "dave"}, {5, "eve"}})
+
+	entries := []iceberg.ManifestEntry{
+		lineageEntry(t, "data/file1.parquet", 100, 7, 3),
+		lineageEntry(t, "data/file2.parquet", 200, 9, 2),
+	}
+
+	merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl",
+		entries, nil, nil, nil, 0, true)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if count != 5 {
+		t.Fatalf("expected 5 merged rows, got %d", count)
+	}
+
+	_, _, rowIDs, seqs := mergedColumns(t, merged)
+	wantIDs := []any{int64(100), int64(101), int64(102), int64(200), int64(201)}
+	wantSeqs := []any{int64(7), int64(7), int64(7), int64(9), int64(9)}
+	for i := range wantIDs {
+		if rowIDs[i] != wantIDs[i] {
+			t.Errorf("row %d _row_id = %v, want %v", i, rowIDs[i], wantIDs[i])
+		}
+		if seqs[i] != wantSeqs[i] {
+			t.Errorf("row %d _last_updated_sequence_number = %v, want %v", i, seqs[i], wantSeqs[i])
+		}
+	}
+}
+
+// A position delete removes a row without renumbering the survivors:
+// each kept row's _row_id is the file's first_row_id plus its position
+// before the delete.
+func TestMergeParquetFilesRowLineagePositionDelete(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	putDataFileRows(t, fs, "file1.parquet", []lineageDataRow{{1, "alice"}, {2, "bob"}, {3, "carl"}})
+
+	entries := []iceberg.ManifestEntry{lineageEntry(t, "data/file1.parquet", 100, 7, 3)}
+	posDeletes := map[string][]int64{"ns/tbl/data/file1.parquet": {1}}
+
+	merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl",
+		entries, posDeletes, nil, nil, 0, true)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if count != 2 {
+		t.Fatalf("expected 2 merged rows, got %d", count)
+	}
+	ids, _, rowIDs, _ := mergedColumns(t, merged)
+	if ids[0] != int64(1) || ids[1] != int64(3) {
+		t.Fatalf("merged ids = %v, want [1 3]", ids)
+	}
+	if rowIDs[0] != int64(100) || rowIDs[1] != int64(102) {
+		t.Fatalf("merged _row_id = %v, want [100 102]", rowIDs)
+	}
+}
+
+// A file an engine already wrote with lineage columns materializes its
+// rows' values; merged with a plain file, each input keeps its own
+// provenance.
+func TestMergeParquetFilesRowLineageMaterialized(t *testing.T) {
+	fs, client := startFakeFiler(t)
+
+	type materializedRow struct {
+		ID     int64  `parquet:"id"`
+		Name   string `parquet:"name"`
+		RowID  int64  `parquet:"_row_id,id(2147483540)"`
+		LastUp int64  `parquet:"_last_updated_sequence_number,id(2147483539)"`
+	}
+	putDataFileRows(t, fs, "plain.parquet", []lineageDataRow{{1, "alice"}})
+	putDataFileRows(t, fs, "mats.parquet", []materializedRow{{2, "bob", 5000, 42}})
+
+	entries := []iceberg.ManifestEntry{
+		lineageEntry(t, "data/plain.parquet", 100, 7, 1),
+		lineageEntry(t, "data/mats.parquet", 200, 9, 1),
+	}
+	merged, count, err := mergeParquetFiles(context.Background(), client, "test-bucket", "ns/tbl",
+		entries, nil, nil, nil, 0, true)
+	if err != nil {
+		t.Fatalf("mergeParquetFiles: %v", err)
+	}
+	if count != 2 {
+		t.Fatalf("expected 2 merged rows, got %d", count)
+	}
+	_, _, rowIDs, seqs := mergedColumns(t, merged)
+	if rowIDs[0] != int64(100) || rowIDs[1] != int64(5000) {
+		t.Errorf("_row_id = %v, want [100 5000]", rowIDs)
+	}
+	if seqs[0] != int64(7) || seqs[1] != int64(42) {
+		t.Errorf("_last_updated_sequence_number = %v, want [7 42]", seqs)
+	}
+}
+
+// The v3 manifest list assigns first_row_id to manifests that have
+// none, starting at the snapshot's first-row-id, and reports the range
+// it assigned as the snapshot's added-rows.
+func TestWriteManifestListV3RowIDs(t *testing.T) {
+	schema := newTestSchema()
+	spec := *iceberg.UnpartitionedSpec
+	snapID := int64(12)
+	dataSeq := int64(3)
+
+	entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, &dataSeq, nil,
+		mustDataFile(t, spec, "data/a.parquet", 4))
+	var manifestBuf bytes.Buffer
+	mf, err := iceberg.WriteManifest("metadata/m0.avro", &manifestBuf, 3, spec, schema, snapID,
+		[]iceberg.ManifestEntry{entry})
+	if err != nil {
+		t.Fatalf("write manifest: %v", err)
+	}
+
+	var listBuf bytes.Buffer
+	parent := int64(10)
+	seqNum := int64(3)
+	addedRows, err := writeManifestList(3, &listBuf, snapID, &parent, &seqNum, 500, []iceberg.ManifestFile{mf})
+	if err != nil {
+		t.Fatalf("writeManifestList: %v", err)
+	}
+	if addedRows == nil || *addedRows != 4 {
+		t.Fatalf("addedRows = %v, want 4", addedRows)
+	}
+
+	readBack, err := iceberg.ReadManifestList(&listBuf)
+	if err != nil {
+		t.Fatalf("read manifest list: %v", err)
+	}
+	if len(readBack) != 1 || readBack[0].FirstRowID() == nil || *readBack[0].FirstRowID() != 500 {
+		t.Fatalf("manifest first_row_id = %v, want 500", readBack)
+	}
+}
+
+func mustDataFile(t *testing.T, spec iceberg.PartitionSpec, path string, recordCount int64) iceberg.DataFile {
+	t.Helper()
+	dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, path, iceberg.ParquetFile,
+		map[int]any{}, nil, nil, recordCount, 1024)
+	if err != nil {
+		t.Fatalf("build data file: %v", err)
+	}
+	return dfb.Build()
+}
+
+// A plan that assigned row IDs from a next-row-id the table no longer
+// has is stale, like a moved head; so is a format version that changed
+// since planning.
+func TestCheckRowLineagePlan(t *testing.T) {
+	if err := checkRowLineagePlan(3, 2, 0, 0); err == nil {
+		t.Error("version change must be stale")
+	}
+	if err := checkRowLineagePlan(3, 3, 100, 200); err == nil {
+		t.Error("moved next-row-id must be stale")
+	}
+	if err := checkRowLineagePlan(2, 2, 0, 0); err != nil {
+		t.Errorf("v2 plan is never row-lineage stale: %v", err)
+	}
+	if err := checkRowLineagePlan(3, 3, 100, 100); err != nil {
+		t.Errorf("unchanged next-row-id must not be stale: %v", err)
+	}
+}
+
+// A data column named like a lineage column under a different field id
+// refuses the bin rather than writing a doubled column.
+func TestLineageLayoutRejectsReservedNameUnderOtherID(t *testing.T) {
+	type badRow struct {
+		ID    int64 `parquet:"id"`
+		RowID int64 `parquet:"_row_id,id(7)"`
+	}
+	schema := parquet.SchemaOf(new(badRow))
+	if _, err := newLineageLayout(schema); err == nil {
+		t.Fatal("reserved name under a non-reserved field id must fail the layout")
+	}
+}
+
+// Compaction on a v3 table reads row lineage the way a reader does: the
+// entries in the stored manifest carry no first_row_id, so the ones
+// compactDataFiles merges are the inherited values (the manifest-list
+// first_row_id plus the preceding files' record counts). The merged file
+// materializes those IDs and its new entry names its own first row's.
+func TestCompactDataFilesV3RowLineageManifestRead(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	ctx := context.Background()
+
+	setup := tableSetup{BucketName: "test-bucket", Namespace: "ns", TableName: "tbl"}
+	spec := *iceberg.UnpartitionedSpec
+	snapID := int64(1)
+
+	putDataFileRows(t, fs, "f1.parquet", []lineageDataRow{{1, "a"}, {2, "b"}, {3, "c"}})
+	putDataFileRows(t, fs, "f2.parquet", []lineageDataRow{{4, "d"}, {5, "e"}})
+
+	entries := []iceberg.ManifestEntry{
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, "data/f1.parquet", 3)),
+		iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil,
+			mustDataFile(t, spec, "data/f2.parquet", 2)),
+	}
+	writeV3Table(t, fs, setup, entries, nil, 0, 5)
+
+	handler := NewHandler(nil)
+	config := Config{
+		TargetFileSizeBytes: 4096,
+		MinInputFiles:       2,
+		ApplyDeletes:        true,
+	}
+	result, _, err := handler.compactDataFiles(ctx, client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	if !strings.Contains(result, "compacted 2 files into 1") {
+		t.Fatalf("unexpected result: %q", result)
+	}
+
+	state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	newSnap := state.Metadata.CurrentSnapshot()
+	if newSnap == nil || newSnap.SnapshotID == snapID {
+		t.Fatalf("no new snapshot committed: %+v", newSnap)
+	}
+	if newSnap.FirstRowID == nil || *newSnap.FirstRowID != 5 {
+		t.Fatalf("new snapshot first-row-id = %v, want 5", newSnap.FirstRowID)
+	}
+
+	manifestListData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, newSnap.ManifestList)
+	if err != nil {
+		t.Fatalf("read new manifest list: %v", err)
+	}
+	manifests, err := s3tables.ReadManifestList(manifestListData)
+	if err != nil {
+		t.Fatalf("parse new manifest list: %v", err)
+	}
+	if len(manifests) != 1 {
+		t.Fatalf("new snapshot has %d manifests, want 1", len(manifests))
+	}
+	manifestData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, manifests[0].FilePath())
+	if err != nil {
+		t.Fatalf("read new manifest: %v", err)
+	}
+	newEntries, err := s3tables.ReadManifest(manifests[0], manifestData, false, specByID(state.Metadata), state.Metadata.CurrentSchema())
+	if err != nil {
+		t.Fatalf("parse new manifest: %v", err)
+	}
+
+	var merged iceberg.DataFile
+	var deletedPaths int
+	for _, e := range newEntries {
+		switch e.Status() {
+		case iceberg.EntryStatusADDED:
+			merged = e.DataFile()
+		case iceberg.EntryStatusDELETED:
+			deletedPaths++
+		}
+	}
+	if deletedPaths != 2 {
+		t.Fatalf("new manifest has %d deleted entries, want 2", deletedPaths)
+	}
+	if merged == nil {
+		t.Fatal("new manifest has no added entry")
+	}
+	if merged.FirstRowID() == nil || *merged.FirstRowID() != 0 {
+		t.Fatalf("merged entry first_row_id = %v, want 0", merged.FirstRowID())
+	}
+
+	mergedData, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, merged.FilePath())
+	if err != nil {
+		t.Fatalf("read merged file: %v", err)
+	}
+	_, _, rowIDs, seqs := mergedColumns(t, mergedData)
+	wantRowIDs := []any{int64(0), int64(1), int64(2), int64(3), int64(4)}
+	for i, want := range wantRowIDs {
+		if rowIDs[i] != want {
+			t.Fatalf("row %d _row_id = %v, want %v", i, rowIDs[i], want)
+		}
+		if seqs[i] != int64(1) {
+			t.Fatalf("row %d _last_updated_sequence_number = %v, want 1", i, seqs[i])
+		}
+	}
+}