fix(manifest): reject row ID allocation overflow

manifest.go

@@ -1573,6 +1573,10 @@ func NewManifestListWriterV2(out io.Writer, snapshotID, sequenceNumber int64, pa
 }
 
 func NewManifestListWriterV3(out io.Writer, snapshotId, sequenceNumber, firstRowID int64, parentSnapshot *int64) (*ManifestListWriter, error) {
+	if firstRowID < 0 {
+		return nil, fmt.Errorf("%w: first row ID must be non-negative: %d", ErrInvalidArgument, firstRowID)
+	}
+
 	m := &ManifestListWriter{
 		version:          3,
 		out:              out,
@@ -1594,6 +1598,19 @@ func NewManifestListWriterV3(out io.Writer, snapshotId, sequenceNumber, firstRow
 	})
 }
 
+func advanceRowID(firstRowID, existingRows, addedRows int64) (int64, error) {
+	if existingRows < 0 || addedRows < 0 {
+		return 0, fmt.Errorf("%w: row counts must be non-negative: existing=%d added=%d",
+			ErrInvalidArgument, existingRows, addedRows)
+	}
+	if existingRows > math.MaxInt64-firstRowID || addedRows > math.MaxInt64-firstRowID-existingRows {
+		return 0, fmt.Errorf("%w: assigning %d existing and %d added rows from first row ID %d overflows int64",
+			ErrInvalidArgument, existingRows, addedRows, firstRowID)
+	}
+
+	return firstRowID + existingRows + addedRows, nil
+}
+
 func (m *ManifestListWriter) init(meta map[string][]byte) error {
 	fileSchema, err := internal.NewManifestFileSchema(m.version)
 	if err != nil {
@@ -1649,6 +1666,8 @@ func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
 
 	case 2, 3:
 		for _, file := range files {
+			var assignedNextRowID *int64
+
 			// Per the Iceberg spec a v2 manifest list may reference v1 manifest
 			// files (and a v3 list may reference v1 or v2 manifests) so that a
 			// table can be upgraded without rewriting historical manifests. The
@@ -1671,8 +1690,12 @@ func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
 					if wrapped.FirstRowIDValue == nil {
 						if m.nextRowID != nil {
 							firstRowID := *m.nextRowID
+							nextRowID, err := advanceRowID(firstRowID, wrapped.ExistingRowsCount, wrapped.AddedRowsCount)
+							if err != nil {
+								return fmt.Errorf("manifest %q: %w", wrapped.Path, err)
+							}
 							wrapped.FirstRowIDValue = &firstRowID
-							*m.nextRowID += wrapped.ExistingRowsCount + wrapped.AddedRowsCount
+							assignedNextRowID = &nextRowID
 						}
 					}
 				}
@@ -1701,6 +1724,9 @@ func (m *ManifestListWriter) AddManifests(files []ManifestFile) error {
 			if err := m.writer.Encode(wrapped); err != nil {
 				return err
 			}
+			if assignedNextRowID != nil {
+				*m.nextRowID = *assignedNextRowID
+			}
 		}
 	default:
 		return fmt.Errorf("unsupported manifest version: %d", m.version)
@@ -1784,6 +1810,10 @@ func WriteManifestV3(
 	snapshotID int64,
 	entries []ManifestEntry,
 ) (mf ManifestFile, nextFirstRowID int64, err error) {
+	if firstRowID < 0 {
+		return nil, 0, fmt.Errorf("%w: first row ID must be non-negative: %d", ErrInvalidArgument, firstRowID)
+	}
+
 	cnt := &internal.CountingWriter{W: out}
 
 	w, err := NewManifestWriter(3, cnt, spec, schema, snapshotID)
@@ -1810,7 +1840,10 @@ func WriteManifestV3(
 	raw := mf.(*manifestFile)
 	v := firstRowID
 	raw.FirstRowIDValue = &v
-	nextFirstRowID = firstRowID + raw.AddedRowsCount + raw.ExistingRowsCount
+	nextFirstRowID, err = advanceRowID(firstRowID, raw.ExistingRowsCount, raw.AddedRowsCount)
+	if err != nil {
+		return nil, 0, err
+	}
 
 	return raw, nextFirstRowID, nil
 }

manifest_test.go

@@ -1235,6 +1235,22 @@ func (m *ManifestTestSuite) TestWriteManifestV3() {
 		m.EqualValues(520, nextID) // 500 + 10 + 10
 	})
 
+	m.Run("rejects negative first row ID", func() {
+		var buf bytes.Buffer
+		_, _, err := WriteManifestV3("/manifest.avro", &buf, -1, partitionSpec, testSchema, entrySnapshotID, entries)
+		m.Require().ErrorIs(err, ErrInvalidArgument)
+		m.Require().ErrorContains(err, "first row ID must be non-negative")
+	})
+
+	m.Run("rejects next row ID overflow", func() {
+		var buf bytes.Buffer
+		_, _, err := WriteManifestV3(
+			"/manifest.avro", &buf, math.MaxInt64-count, partitionSpec, testSchema, entrySnapshotID, entries,
+		)
+		m.Require().ErrorIs(err, ErrInvalidArgument)
+		m.Require().ErrorContains(err, "overflows int64")
+	})
+
 	m.Run("read inheritance from manifest", func() {
 		var buf bytes.Buffer
 		firstRowID := int64(500)
@@ -2430,6 +2446,47 @@ func (m *ManifestTestSuite) TestV3ManifestListWriterRowIDTracking() {
 	m.Require().NoError(err)
 }
 
+func (m *ManifestTestSuite) TestV3ManifestListWriterRejectsInvalidRowIDRanges() {
+	m.Run("negative first row ID", func() {
+		var buf bytes.Buffer
+		writer, err := NewManifestListWriterV3(&buf, snapshotID, 1, -1, nil)
+		m.Nil(writer)
+		m.Require().ErrorIs(err, ErrInvalidArgument)
+		m.Require().ErrorContains(err, "first row ID must be non-negative")
+	})
+
+	m.Run("negative row count", func() {
+		var buf bytes.Buffer
+		writer, err := NewManifestListWriterV3(&buf, snapshotID, 1, 0, nil)
+		m.Require().NoError(err)
+		manifest := NewManifestFile(3, "negative.avro", 100, 1, snapshotID).AddedRows(-1).Build()
+		err = writer.AddManifests([]ManifestFile{manifest})
+		m.Require().ErrorIs(err, ErrInvalidArgument)
+		m.Require().ErrorContains(err, "row counts must be non-negative")
+	})
+
+	m.Run("overflow", func() {
+		var buf bytes.Buffer
+		writer, err := NewManifestListWriterV3(&buf, snapshotID, 1, math.MaxInt64, nil)
+		m.Require().NoError(err)
+		manifest := NewManifestFile(3, "overflow.avro", 100, 1, snapshotID).AddedRows(1).Build()
+		err = writer.AddManifests([]ManifestFile{manifest})
+		m.Require().ErrorIs(err, ErrInvalidArgument)
+		m.Require().ErrorContains(err, "overflows int64")
+		m.EqualValues(math.MaxInt64, *writer.NextRowID())
+	})
+
+	m.Run("later validation failure leaves cursor unchanged", func() {
+		var buf bytes.Buffer
+		writer, err := NewManifestListWriterV3(&buf, snapshotID, 1, 10, nil)
+		m.Require().NoError(err)
+		manifest := NewManifestFile(3, "other-snapshot.avro", 100, 1, snapshotID+1).AddedRows(5).Build()
+		err = writer.AddManifests([]ManifestFile{manifest})
+		m.Require().ErrorContains(err, "unassigned sequence number")
+		m.EqualValues(10, *writer.NextRowID())
+	})
+}
+
 func (m *ManifestTestSuite) TestV3ManifestListWriterAssignedRowIDDelta() {
 	// Assigned row-id delta = sum of (existing+added) for all data manifests in list.
 	var buf bytes.Buffer