Add support for row lineage in v3

Signed-off-by: dttung2905 <ttdao.2015@accountancy.smu.edu.sg>

Author: dttung2905

manifest.go

@@ -303,6 +303,8 @@ func (m *manifestFileV1) Partitions() []FieldSummary {
 	return *m.PartitionList
 }
 
+func (*manifestFileV1) FirstRowId() *int64 { return nil }
+
 func (m *manifestFileV1) FetchEntries(fs iceio.IO, discardDeleted bool) ([]ManifestEntry, error) {
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
@@ -323,7 +325,7 @@ type manifestFile struct {
 	DeletedRowsCount   int64           `avro:"deleted_rows_count"`
 	PartitionList      *[]FieldSummary `avro:"partitions"`
 	Key                []byte          `avro:"key_metadata"`
-	FirstRowId         *int64          `avro:"first_row_id"`
+	FirstRowID         *int64          `avro:"first_row_id"`
 
 	version int `avro:"-"`
 }
@@ -400,6 +402,8 @@ func (m *manifestFile) Partitions() []FieldSummary {
 	return *m.PartitionList
 }
 
+func (m *manifestFile) FirstRowId() *int64 { return m.FirstRowID }
+
 func (m *manifestFile) HasAddedFiles() bool    { return m.AddedFilesCount != 0 }
 func (m *manifestFile) HasExistingFiles() bool { return m.ExistingFilesCount != 0 }
 func (m *manifestFile) FetchEntries(fs iceio.IO, discardDeleted bool) ([]ManifestEntry, error) {
@@ -520,6 +524,9 @@ type ManifestFile interface {
 	// field in the spec. Each field in the list corresponds to a field in
 	// the manifest file's partition spec.
 	Partitions() []FieldSummary
+	// FirstRowId returns the first _row_id assigned to rows in this manifest (v3+ data manifests only).
+	// Returns nil for v1/v2 or for delete manifests.
+	FirstRowId() *int64
 
 	// HasAddedFiles returns true if AddedDataFiles > 0 or if it was null.
 	HasAddedFiles() bool
@@ -595,6 +602,11 @@ type ManifestReader struct {
 	schemaLoaded        bool
 	partitionSpec       PartitionSpec
 	partitionSpecLoaded bool
+
+	// nextFirstRowID tracks the next first_row_id to assign when reading v3 data
+	// manifests; used for First Row ID inheritance (null data file first_row_id
+	// gets manifest's first_row_id + sum of preceding null-first_row_id files' record_count).
+	nextFirstRowID *int64
 }
 
 // NewManifestReader returns a value that can read the contents of an avro manifest
@@ -759,6 +771,19 @@ func (c *ManifestReader) ReadEntry() (ManifestEntry, error) {
 		tmp = tmp.(*fallbackManifestEntry).toEntry()
 	}
 	tmp.inherit(c.file)
+	// Apply first_row_id inheritance for v3 data manifests (spec: First Row ID Inheritance).
+	if c.content == ManifestContentData && c.file.FirstRowId() != nil {
+		if df, ok := tmp.DataFile().(*dataFile); ok && df.FirstRowIDField == nil {
+			if c.nextFirstRowID == nil {
+				c.nextFirstRowID = new(int64)
+				*c.nextFirstRowID = *c.file.FirstRowId()
+			}
+			id := new(int64)
+			*id = *c.nextFirstRowID
+			df.FirstRowIDField = id
+			*c.nextFirstRowID += df.Count()
+		}
+	}
 	if fieldToIDMap, ok := tmp.DataFile().(hasFieldToIDMap); ok {
 		fieldToIDMap.setFieldNameToIDMap(c.fieldNameToID)
 		fieldToIDMap.setFieldIDToLogicalTypeMap(c.fieldIDToType)
@@ -1404,10 +1429,10 @@ func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
 			wrapped := *(file.(*manifestFile))
 			if m.version == 3 {
 				// Ref: https://github.com/apache/iceberg/blob/ea2071568dc66148b483a82eefedcd2992b435f7/core/src/main/java/org/apache/iceberg/ManifestListWriter.java#L157-L168
-				if wrapped.Content == ManifestContentData && wrapped.FirstRowId == nil {
+				if wrapped.Content == ManifestContentData && wrapped.FirstRowID == nil {
 					if m.nextRowID != nil {
 						firstRowID := *m.nextRowID
-						wrapped.FirstRowId = &firstRowID
+						wrapped.FirstRowID = &firstRowID
 						*m.nextRowID += wrapped.ExistingRowsCount + wrapped.AddedRowsCount
 					}
 				}
@@ -1445,32 +1470,40 @@ func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
 }
 
 // WriteManifestList writes a list of manifest files to an avro file.
-func WriteManifestList(version int, out io.Writer, snapshotID int64, parentSnapshotID, sequenceNumber *int64, firstRowId int64, files []ManifestFile) (err error) {
+// For v3 tables, it returns the next row ID after assigning first_row_id to data manifests; otherwise 0.
+func WriteManifestList(version int, out io.Writer, snapshotID int64, parentSnapshotID, sequenceNumber *int64, firstRowId int64, files []ManifestFile) (nextRowIDAfter int64, err error) {
 	var writer *ManifestListWriter
 
 	switch version {
 	case 1:
 		writer, err = NewManifestListWriterV1(out, snapshotID, parentSnapshotID)
 	case 2:
 		if sequenceNumber == nil {
-			return errors.New("sequence number is required for V2 tables")
+			return 0, errors.New("sequence number is required for V2 tables")
 		}
 		writer, err = NewManifestListWriterV2(out, snapshotID, *sequenceNumber, parentSnapshotID)
 	case 3:
 		if sequenceNumber == nil {
-			return errors.New("sequence number is required for V3 tables")
+			return 0, errors.New("sequence number is required for V3 tables")
 		}
 		writer, err = NewManifestListWriterV3(out, snapshotID, *sequenceNumber, firstRowId, parentSnapshotID)
 	default:
-		return fmt.Errorf("unsupported manifest version: %d", version)
+		return 0, fmt.Errorf("unsupported manifest version: %d", version)
 	}
 
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer internal.CheckedClose(writer, &err)
 
-	return writer.AddManifests(files)
+	if err = writer.AddManifests(files); err != nil {
+		return 0, err
+	}
+	if version == 3 && writer.NextRowID() != nil {
+		nextRowIDAfter = *writer.NextRowID()
+	}
+
+	return nextRowIDAfter, nil
 }
 
 func WriteManifest(

manifest_test.go

@@ -481,12 +481,15 @@ type ManifestTestSuite struct {
 }
 
 func (m *ManifestTestSuite) writeManifestList() {
-	m.Require().NoError(WriteManifestList(1, &m.v1ManifestList, snapshotID, nil, nil, 0, manifestFileRecordsV1))
+	_, err := WriteManifestList(1, &m.v1ManifestList, snapshotID, nil, nil, 0, manifestFileRecordsV1)
+	m.Require().NoError(err)
 	unassignedSequenceNum := int64(-1)
-	m.Require().NoError(WriteManifestList(2, &m.v2ManifestList, snapshotID, nil, &unassignedSequenceNum, 0, manifestFileRecordsV2))
+	_, err = WriteManifestList(2, &m.v2ManifestList, snapshotID, nil, &unassignedSequenceNum, 0, manifestFileRecordsV2)
+	m.Require().NoError(err)
 	v3SequenceNum := int64(5)
 	firstRowID := int64(1000)
-	m.Require().NoError(WriteManifestList(3, &m.v3ManifestList, snapshotID, nil, &v3SequenceNum, firstRowID, manifestFileRecordsV3))
+	_, err = WriteManifestList(3, &m.v3ManifestList, snapshotID, nil, &v3SequenceNum, firstRowID, manifestFileRecordsV3)
+	m.Require().NoError(err)
 }
 
 func (m *ManifestTestSuite) writeManifestEntries() {
@@ -760,13 +763,75 @@ func (m *ManifestTestSuite) TestReadManifestListV3() {
 	m.Nil(list[0].KeyMetadata())
 	m.Zero(list[0].PartitionSpecID())
 
+	// V3 manifest list assigns first_row_id to data manifests
+	m.Require().NotNil(list[0].FirstRowId(), "v3 data manifest should have first_row_id")
+	m.EqualValues(1000, *list[0].FirstRowId())
+
 	part := list[0].Partitions()[0]
 	m.True(part.ContainsNull)
 	m.False(*part.ContainsNaN)
 	m.Equal([]byte{0x01, 0x00, 0x00, 0x00}, *part.LowerBound)
 	m.Equal([]byte{0x02, 0x00, 0x00, 0x00}, *part.UpperBound)
 }
 
+func (m *ManifestTestSuite) TestV3DataManifestFirstRowIDInheritance() {
+	// Build a v3 data manifest with two entries that have null first_row_id.
+	partitionSpec := NewPartitionSpecID(1,
+		PartitionField{FieldID: 1000, SourceID: 1, Name: "x", Transform: IdentityTransform{}})
+	firstCount, secondCount := int64(10), int64(20)
+	entriesWithNullFirstRowID := []ManifestEntry{
+		&manifestEntry{
+			EntryStatus: EntryStatusADDED,
+			Snapshot:    &entrySnapshotID,
+			Data: &dataFile{
+				Content:           EntryContentData,
+				Path:              "/data/file1.parquet",
+				Format:            ParquetFile,
+				PartitionData:     map[string]any{"x": int(1)},
+				RecordCount:        firstCount,
+				FileSize:          1000,
+				BlockSizeInBytes:  64 * 1024,
+				FirstRowIDField:   nil, // null so reader will inherit
+			},
+		},
+		&manifestEntry{
+			EntryStatus: EntryStatusADDED,
+			Snapshot:    &entrySnapshotID,
+			Data: &dataFile{
+				Content:           EntryContentData,
+				Path:              "/data/file2.parquet",
+				Format:            ParquetFile,
+				PartitionData:     map[string]any{"x": int(2)},
+				RecordCount:       secondCount,
+				FileSize:          2000,
+				BlockSizeInBytes:  64 * 1024,
+				FirstRowIDField:   nil,
+			},
+		},
+	}
+	var manifestBuf bytes.Buffer
+	_, err := WriteManifest("/manifest.avro", &manifestBuf, 3, partitionSpec, testSchema, entrySnapshotID, entriesWithNullFirstRowID)
+	m.Require().NoError(err)
+
+	manifestFirstRowID := int64(1000)
+	file := &manifestFile{
+		version:    3,
+		Path:       "/manifest.avro",
+		Content:    ManifestContentData,
+		FirstRowID: &manifestFirstRowID,
+	}
+	entries, err := ReadManifest(file, bytes.NewReader(manifestBuf.Bytes()), false)
+	m.Require().NoError(err)
+	m.Require().Len(entries, 2)
+
+	// First entry gets manifest's first_row_id
+	m.Require().NotNil(entries[0].DataFile().FirstRowID())
+	m.EqualValues(1000, *entries[0].DataFile().FirstRowID())
+	// Second entry gets previous + previous file's record_count
+	m.Require().NotNil(entries[1].DataFile().FirstRowID())
+	m.EqualValues(1000+firstCount, *entries[1].DataFile().FirstRowID())
+}
+
 func (m *ManifestTestSuite) TestReadManifestListIncompleteSchema() {
 	// This prevents a regression that could be caused by using a schema cache
 	// across multiple read/write operations of an avro file. While it may sound
@@ -779,7 +844,7 @@ func (m *ManifestTestSuite) TestReadManifestListIncompleteSchema() {
 	// any cache. (Note: if working correctly, this will have no such side effect.)
 	var buf bytes.Buffer
 	seqNum := int64(9876)
-	err := WriteManifestList(2, &buf, 1234, nil, &seqNum, 0, []ManifestFile{
+	_, err := WriteManifestList(2, &buf, 1234, nil, &seqNum, 0, []ManifestFile{
 		NewManifestFile(2, "s3://bucket/namespace/table/metadata/abcd-0123.avro", 99, 0, 1234).Build(),
 	})
 	m.NoError(err)
@@ -1550,7 +1615,7 @@ func (m *ManifestTestSuite) TestWriteManifestListClosesWriterOnError() {
 	m.Require().NoError(writer.Close())
 
 	out := &limitedWriter{limit: header.Len(), err: errLimitedWrite}
-	err = WriteManifestList(2, out, snapshotID, nil, &seqNum, 0, []ManifestFile{
+	_, err = WriteManifestList(2, out, snapshotID, nil, &seqNum, 0, []ManifestFile{
 		manifestFileRecordsV2[0],
 		manifestFileRecordsV1[0],
 	})

metadata_columns.go

@@ -0,0 +1,61 @@
+// 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 it 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 iceberg
+
+// Row lineage metadata column field IDs (v3+). Reserved IDs are Integer.MAX_VALUE - 107 and 108
+// per the Iceberg spec (Metadata Columns / Row Lineage).
+const (
+	// RowIDFieldID is the field ID for _row_id (optional long). A unique long identifier for every row.
+	RowIDFieldID = 2147483540
+	// LastUpdatedSequenceNumberFieldID is the field ID for _last_updated_sequence_number (optional long).
+	// The sequence number of the commit that last updated the row.
+	LastUpdatedSequenceNumberFieldID = 2147483539
+)
+
+// Row lineage metadata column names (v3+).
+const (
+	RowIDColumnName                         = "_row_id"
+	LastUpdatedSequenceNumberColumnName     = "_last_updated_sequence_number"
+)
+
+// RowID returns a NestedField for _row_id (optional long) for use in schemas that include row lineage.
+func RowID() NestedField {
+	return NestedField{
+		ID:        RowIDFieldID,
+		Name:      RowIDColumnName,
+		Required:  false,
+		Doc:       "Implicit row ID that is automatically assigned",
+		Type:      Int64Type{},
+	}
+}
+
+// LastUpdatedSequenceNumber returns a NestedField for _last_updated_sequence_number (optional long).
+func LastUpdatedSequenceNumber() NestedField {
+	return NestedField{
+		ID:       LastUpdatedSequenceNumberFieldID,
+		Name:     LastUpdatedSequenceNumberColumnName,
+		Required: false,
+		Doc:      "Sequence number when the row was last updated",
+		Type:     Int64Type{},
+	}
+}
+
+// IsMetadataColumn returns true if the field ID is a reserved metadata column (e.g. row lineage).
+func IsMetadataColumn(fieldID int) bool {
+	return fieldID == RowIDFieldID || fieldID == LastUpdatedSequenceNumberFieldID
+}