You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/06/29 16:58:50 UTC
(iceberg-go) branch main updated: feat: let WriteManifest set and return v3 first_row_id (#1321)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new bf9cb31a feat: let WriteManifest set and return v3 first_row_id (#1321)
bf9cb31a is described below
commit bf9cb31a0ce97e5b24ec346d34592dec670ee6d8
Author: Hardik Borse <17...@users.noreply.github.com>
AuthorDate: Mon Jun 29 22:28:44 2026 +0530
feat: let WriteManifest set and return v3 first_row_id (#1321)
**Summary**
* add `WithManifestFileFirstRowID`, a `ManifestFileOption` that sets
`first_row_id` on a v3+ data manifest
* forward variadic `ManifestFileOption` args through `WriteManifest` to
`ToManifestFile`
**Why**
`WriteManifest` predates the v3 work in #735 and forwarded no options to
`ToManifestFile`, so `ManifestFile.FirstRowID()` was always `nil` for
manifests written through this public helper, even though
`ManifestWriter`/`ToManifestFile` already supported setting it via
`opts`. Library consumers had no way to set or observe `first_row_id`
through `WriteManifest`.
`WriteManifest`'s signature change is purely additive (`opts` is the
last, variadic parameter), so existing call sites are unaffected.
**Testing**
* `go test ./... -run '^TestWriteManifestWithFirstRowIDOption$'
-count=1`
* `go test ./... -count=1`
* `gofmt -l .`
* `golangci-lint run`
Fixes #1274
---------
Signed-off-by: the-onewho-knocks <bo...@gmail.com>
---
manifest.go | 62 +++++++++++++++++++--
manifest_test.go | 162 ++++++++++++++++++++++++++++++++++++++++++++++++++-----
2 files changed, 206 insertions(+), 18 deletions(-)
diff --git a/manifest.go b/manifest.go
index a49f9263..0566abc2 100644
--- a/manifest.go
+++ b/manifest.go
@@ -1306,6 +1306,10 @@ func (w *ManifestWriter) ToManifestFile(location string, length int64, opts ...M
apply(&mf)
}
+ if mf.Content == ManifestContentDeletes && mf.FirstRowIDValue != nil {
+ return nil, errors.New("first_row_id must not be set on delete manifests")
+ }
+
return &mf, nil
}
@@ -1551,11 +1555,13 @@ 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.FirstRowIDValue == nil {
- if m.nextRowID != nil {
- firstRowID := *m.nextRowID
- wrapped.FirstRowIDValue = &firstRowID
- *m.nextRowID += wrapped.ExistingRowsCount + wrapped.AddedRowsCount
+ if wrapped.Content == ManifestContentData {
+ if wrapped.FirstRowIDValue == nil {
+ if m.nextRowID != nil {
+ firstRowID := *m.nextRowID
+ wrapped.FirstRowIDValue = &firstRowID
+ *m.nextRowID += wrapped.ExistingRowsCount + wrapped.AddedRowsCount
+ }
}
}
}
@@ -1651,6 +1657,52 @@ func WriteManifest(
return w.ToManifestFile(filename, cnt.Count)
}
+// WriteManifestV3 writes a v3 data manifest and assigns first_row_id.
+// The returned ManifestFile has FirstRowID() set to firstRowID.
+// nextFirstRowID is firstRowID + AddedRowsCount + ExistingRowsCount
+// for the caller to chain to the next manifest. Use this only for
+// reconstruction — manifests committed through a v3 list writer get
+// first_row_id from AddManifests; pre-setting collides with that allocation.
+func WriteManifestV3(
+ filename string,
+ out io.Writer,
+ firstRowID int64,
+ spec PartitionSpec,
+ schema *Schema,
+ snapshotID int64,
+ entries []ManifestEntry,
+) (mf ManifestFile, nextFirstRowID int64, err error) {
+ cnt := &internal.CountingWriter{W: out}
+
+ w, err := NewManifestWriter(3, cnt, spec, schema, snapshotID)
+ if err != nil {
+ return nil, 0, err
+ }
+ defer internal.CheckedClose(w, &err)
+
+ for _, entry := range entries {
+ if err := w.addEntry(entry.(*manifestEntry)); err != nil {
+ return nil, 0, err
+ }
+ }
+
+ if err := w.Close(); err != nil {
+ return nil, 0, err
+ }
+
+ mf, err = w.ToManifestFile(filename, cnt.Count)
+ if err != nil {
+ return nil, 0, err
+ }
+
+ raw := mf.(*manifestFile)
+ v := firstRowID
+ raw.FirstRowIDValue = &v
+ nextFirstRowID = firstRowID + raw.AddedRowsCount + raw.ExistingRowsCount
+
+ return raw, nextFirstRowID, nil
+}
+
// ManifestEntryStatus defines constants for the entry status of
// existing, added or deleted.
type ManifestEntryStatus int8
diff --git a/manifest_test.go b/manifest_test.go
index 27f24f33..86b024f3 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -453,7 +453,8 @@ var (
},
}
- testSchema = NewSchema(0,
+ testSchema = NewSchema(
+ 0,
NestedField{ID: 1, Name: "VendorID", Type: PrimitiveTypes.Int32, Required: true},
NestedField{ID: 2, Name: "tpep_pickup_datetime", Type: PrimitiveTypes.Timestamp, Required: true},
NestedField{ID: 3, Name: "tpep_dropoff_datetime", Type: PrimitiveTypes.Timestamp, Required: true},
@@ -1159,6 +1160,129 @@ func (m *ManifestTestSuite) TestV3DataManifestFirstRowIDInheritanceSkipsDeletedE
m.EqualValues(1000+liveCount, *read[2].DataFile().FirstRowID())
}
+func (m *ManifestTestSuite) TestWriteManifestV3() {
+ partitionSpec := NewPartitionSpecID(1,
+ PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name: "x", Transform: IdentityTransform{}})
+ count := int64(10)
+ entries := []ManifestEntry{
+ &manifestEntry{
+ EntryStatus: EntryStatusADDED,
+ Snapshot: &entrySnapshotID,
+ Data: &dataFile{
+ Content: EntryContentData,
+ Path: "/data/file1.parquet",
+ Format: ParquetFile,
+ PartitionData: map[string]any{"x": int(1)},
+ RecordCount: count,
+ FileSize: 1000,
+ BlockSizeInBytes: 64 * 1024,
+ FirstRowIDField: nil,
+ },
+ },
+ &manifestEntry{
+ EntryStatus: EntryStatusADDED,
+ Snapshot: &entrySnapshotID,
+ Data: &dataFile{
+ Content: EntryContentData,
+ Path: "/data/file2.parquet",
+ Format: ParquetFile,
+ PartitionData: map[string]any{"x": int(2)},
+ RecordCount: count,
+ FileSize: 2000,
+ BlockSizeInBytes: 64 * 1024,
+ FirstRowIDField: nil,
+ },
+ },
+ }
+
+ m.Run("sets first_row_id on returned manifest", func() {
+ var buf bytes.Buffer
+ mf, _, err := WriteManifestV3("/manifest.avro", &buf, 500, partitionSpec, testSchema, entrySnapshotID, entries)
+ m.Require().NoError(err)
+ m.Require().NotNil(mf.FirstRowID())
+ m.EqualValues(500, *mf.FirstRowID())
+ })
+
+ m.Run("returns nextFirstRowID", func() {
+ var buf bytes.Buffer
+ // Two entries each with count=10, all added => 20 rows total
+ _, nextID, err := WriteManifestV3("/manifest.avro", &buf, 500, partitionSpec, testSchema, entrySnapshotID, entries)
+ m.Require().NoError(err)
+ m.EqualValues(520, nextID) // 500 + 10 + 10
+ })
+
+ m.Run("read inheritance from manifest", func() {
+ var buf bytes.Buffer
+ firstRowID := int64(500)
+ mf, _, err := WriteManifestV3("/manifest.avro", &buf, firstRowID, partitionSpec, testSchema, entrySnapshotID, entries)
+ m.Require().NoError(err)
+
+ read, err := ReadManifest(mf, bytes.NewReader(buf.Bytes()), false)
+ m.Require().NoError(err)
+ m.Require().Len(read, 2)
+ m.Require().NotNil(read[0].DataFile().FirstRowID())
+ m.EqualValues(firstRowID, *read[0].DataFile().FirstRowID())
+ m.Require().NotNil(read[1].DataFile().FirstRowID())
+ m.EqualValues(firstRowID+count, *read[1].DataFile().FirstRowID())
+ })
+
+ m.Run("WriteManifest backward compat returns nil first_row_id", func() {
+ var buf bytes.Buffer
+ mf, err := WriteManifest("/manifest.avro", &buf, 3, partitionSpec, testSchema, entrySnapshotID, entries)
+ m.Require().NoError(err)
+ m.Nil(mf.FirstRowID())
+ })
+}
+
+func (m *ManifestTestSuite) TestAddManifestsPresetAndNilFirstRowIDNoOverlap() {
+ partitionSpec := NewPartitionSpecID(1,
+ PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name: "x", Transform: IdentityTransform{}})
+ commitSnapID := int64(100)
+ seqNum := int64(1)
+ count := int64(1)
+ entries := []ManifestEntry{
+ &manifestEntry{
+ EntryStatus: EntryStatusADDED,
+ Snapshot: &commitSnapID,
+ Data: &dataFile{
+ Content: EntryContentData,
+ Path: "/data/m1.parquet",
+ Format: ParquetFile,
+ PartitionData: map[string]any{"x": int(1)},
+ RecordCount: count,
+ FileSize: 1000,
+ BlockSizeInBytes: 64 * 1024,
+ },
+ },
+ }
+
+ var buf1 bytes.Buffer
+ m1, _, err := WriteManifestV3("/m1.avro", &buf1, 0, partitionSpec, testSchema, commitSnapID, entries)
+ m.Require().NoError(err)
+ m.Require().NotNil(m1.FirstRowID())
+ m.EqualValues(0, *m1.FirstRowID())
+
+ var buf2 bytes.Buffer
+ m2, err := WriteManifest("/m2.avro", &buf2, 3, partitionSpec, testSchema, commitSnapID, entries)
+ m.Require().NoError(err)
+ m.Nil(m2.FirstRowID())
+
+ var listBuf bytes.Buffer
+ writer, err := NewManifestListWriterV3(&listBuf, commitSnapID, seqNum, 1, nil)
+ m.Require().NoError(err)
+ m.Require().NoError(writer.AddManifests([]ManifestFile{m1, m2}))
+ m.Require().NoError(writer.Close())
+
+ list, err := ReadManifestList(bytes.NewReader(listBuf.Bytes()))
+ m.Require().NoError(err)
+ m.Require().Len(list, 2)
+ m.Require().NotNil(list[0].FirstRowID())
+ m.EqualValues(0, *list[0].FirstRowID())
+ m.Require().NotNil(list[1].FirstRowID())
+ m.EqualValues(1, *list[1].FirstRowID())
+ m.EqualValues(2, *writer.NextRowID())
+}
+
func (m *ManifestTestSuite) TestReadManifestListIncompleteSchema() {
// Verify that reading a manifest list whose embedded schema references
// an undefined named type ("field_summary" without its definition)
@@ -1279,7 +1403,8 @@ func (m *ManifestTestSuite) TestReadManifestListIncompleteSchema() {
// We'll generate a file that is missing part of its schema
sch, err := internal.NewManifestFileSchema(2)
m.NoError(err)
- wr, err := ocf.NewWriter(&buf, sch,
+ wr, err := ocf.NewWriter(
+ &buf, sch,
ocf.WithSchema(incompleteSchema),
ocf.WithMetadata(map[string][]byte{
"format-version": {'2'},
@@ -1322,7 +1447,8 @@ func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
file, err := WriteManifest(
"s3://bucket/namespace/table/metadata/abcd-0123.avro", &buf, 2,
partitionSpec,
- NewSchema(123,
+ NewSchema(
+ 123,
NestedField{ID: 1, Name: "id", Type: Int64Type{}},
NestedField{ID: 2, Name: "name", Type: StringType{}},
),
@@ -1396,7 +1522,8 @@ func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
m.NoError(err)
sch, err := internal.NewManifestEntrySchema(partitionSchema, 2)
m.NoError(err)
- wr, err := ocf.NewWriter(&buf, sch,
+ wr, err := ocf.NewWriter(
+ &buf, sch,
ocf.WithSchema(incompleteSchema),
ocf.WithMetadata(map[string][]byte{
"format-version": {'2'},
@@ -1751,7 +1878,8 @@ func (m *ManifestTestSuite) TestManifestEntryBuilder() {
entry := NewManifestEntryBuilder(
EntryStatusEXISTING,
&snapshotEntryID,
- dataFileBuilder.Build()).Build()
+ dataFileBuilder.Build(),
+ ).Build()
m.Assert().Equal(EntryStatusEXISTING, entry.Status())
m.Assert().EqualValues(1, entry.SnapshotID())
@@ -1849,10 +1977,12 @@ func (m *ManifestTestSuite) TestManifestWriterMeta() {
}
func TestReadManifestDecodesNilLogicalPartitionValueFromNullableUnion(t *testing.T) {
- schema := NewSchema(0,
+ schema := NewSchema(
+ 0,
NestedField{ID: 1, Name: "dt", Type: PrimitiveTypes.Date},
)
- partitionSpec := NewPartitionSpecID(1,
+ partitionSpec := NewPartitionSpecID(
+ 1,
PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name: "dt", Transform: IdentityTransform{}},
)
@@ -2181,10 +2311,12 @@ func (w *limitedWriter) Write(p []byte) (int, error) {
}
func (m *ManifestTestSuite) TestManifestWriterDoesNotCommitStateOnEncodeError() {
- schema := NewSchema(0,
+ schema := NewSchema(
+ 0,
NestedField{ID: 1, Name: "dt", Type: PrimitiveTypes.Date},
)
- partitionSpec := NewPartitionSpecID(1,
+ partitionSpec := NewPartitionSpecID(
+ 1,
PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name: "dt", Transform: IdentityTransform{}},
)
@@ -2438,7 +2570,8 @@ func (m *ManifestTestSuite) TestManifestRoundTripSortOrderID() {
file, err := WriteManifest(
"s3://bucket/ns/table/metadata/round-trip.avro", &buf, 2,
partitionSpec,
- NewSchema(0,
+ NewSchema(
+ 0,
NestedField{ID: 1, Name: "id", Type: Int64Type{}},
),
snapshotID,
@@ -2496,7 +2629,8 @@ func (m *ManifestTestSuite) assertWriteOmitsDistinctCounts(version int) {
file, err := WriteManifest(
"s3://bucket/ns/table/metadata/distinct.avro", &buf, version,
partitionSpec,
- NewSchema(0,
+ NewSchema(
+ 0,
NestedField{ID: 1, Name: "id", Type: Int64Type{}, Required: true},
),
snapshotID,
@@ -2531,7 +2665,8 @@ func (m *ManifestTestSuite) assertWriteOmitsDistinctCounts(version int) {
// the distinct counts come back populated.
func (m *ManifestTestSuite) TestReadManifestLegacyDistinctCounts() {
partitionSpec := NewPartitionSpec()
- tableSchema := NewSchema(0,
+ tableSchema := NewSchema(
+ 0,
NestedField{ID: 1, Name: "id", Type: Int64Type{}, Required: true},
)
partitionSchema, err := partitionTypeToAvroSchema(partitionSpec.PartitionType(tableSchema))
@@ -2559,7 +2694,8 @@ func (m *ManifestTestSuite) TestReadManifestLegacyDistinctCounts() {
}
var buf bytes.Buffer
- wr, err := ocf.NewWriter(&buf, legacySchema,
+ wr, err := ocf.NewWriter(
+ &buf, legacySchema,
ocf.WithSchema(legacySchema.String()),
ocf.WithMetadata(map[string][]byte{
"format-version": []byte("2"),