diff --git a/weed/worker/tasks/iceberg/row_lineage.go b/weed/worker/tasks/iceberg/row_lineage.go index c751f4a98b..f6aa85c2aa 100644 --- a/weed/worker/tasks/iceberg/row_lineage.go +++ b/weed/worker/tasks/iceberg/row_lineage.go @@ -368,12 +368,16 @@ func copyFileRows(ctx context.Context, reader *parquet.Reader, entry iceberg.Man }) } -// 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. +// mergedFileFirstRowID is the _row_id of the merged output's first row, or nil when the file must be committed +// without a first_row_id: a null _row_id would inherit first_row_id+position, an id another row already holds. +// Committed without one, the nulls take fresh ids from the table's next-row-id and materialized values still +// override inheritance. func mergedFileFirstRowID(data []byte) (*int64, error) { - reader := parquet.NewReader(bytes.NewReader(data)) - defer reader.Close() - schema := reader.Schema() + file, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) + if err != nil { + return nil, err + } + schema := file.Schema() if schema == nil { return nil, nil } @@ -381,6 +385,11 @@ func mergedFileFirstRowID(data []byte) (*int64, error) { if !ok { return nil, nil } + if rowIDHasNulls(file, leaf.ColumnIndex) { + return nil, nil + } + reader := parquet.NewReader(file) + defer reader.Close() rows := make([]parquet.Row, 1) n, err := reader.ReadRows(rows) if err != nil && err != io.EOF { @@ -402,6 +411,22 @@ func mergedFileFirstRowID(data []byte) (*int64, error) { return nil, nil } +// rowIDHasNulls reports whether any row group's _row_id column chunk counts a null. A chunk whose statistics +// cannot be read counts as having one: a first_row_id named on a hidden null is the failure this guards. +func rowIDHasNulls(file *parquet.File, columnIndex int) bool { + for _, rg := range file.RowGroups() { + chunks := rg.ColumnChunks() + if columnIndex >= len(chunks) { + return true + } + cc, ok := chunks[columnIndex].(*parquet.FileColumnChunk) + if !ok || cc.NullCount() > 0 { + return true + } + } + return false +} + // lineageCodec is the codec of the first data column leaf that declares one, // ZSTD when none does. func lineageCodec(base []parquet.Field) compress.Codec { diff --git a/weed/worker/tasks/iceberg/row_lineage_first_id_test.go b/weed/worker/tasks/iceberg/row_lineage_first_id_test.go new file mode 100644 index 0000000000..07d9929640 --- /dev/null +++ b/weed/worker/tasks/iceberg/row_lineage_first_id_test.go @@ -0,0 +1,63 @@ +package iceberg + +import ( + "bytes" + "testing" + + "github.com/parquet-go/parquet-go" +) + +type lineageRow struct { + ID int64 `parquet:"id"` + RowID *int64 `parquet:"_row_id,optional"` +} + +func lineageFile(t *testing.T, ids ...*int64) []byte { + t.Helper() + var buf bytes.Buffer + w := parquet.NewGenericWriter[lineageRow](&buf) + rows := make([]lineageRow, len(ids)) + for i, id := range ids { + rows[i] = lineageRow{ID: int64(i), RowID: id} + } + if _, err := w.Write(rows); err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +func i64(v int64) *int64 { return &v } + +// A merged v3 file names its first row's id as first_row_id only when every +// row carries a materialized id: a null row would inherit first_row_id plus +// its position, an id another row already holds (high review of the 4.49 +// series; 4.49's own code took the first row's id whatever followed). +func TestMergedFileFirstRowIDOnlyWithoutNullIDs(t *testing.T) { + cases := []struct { + name string + ids []*int64 + want *int64 + }{ + {"every row materialized", []*int64{i64(40), i64(41), i64(42)}, i64(40)}, + {"materialized first, a null later", []*int64{i64(40), nil, i64(42)}, nil}, + {"null first", []*int64{nil, i64(41)}, nil}, + {"all null", []*int64{nil, nil}, nil}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, err := mergedFileFirstRowID(lineageFile(t, c.ids...)) + if err != nil { + t.Fatal(err) + } + switch { + case c.want == nil && got != nil: + t.Fatalf("first_row_id %d, want none", *got) + case c.want != nil && (got == nil || *got != *c.want): + t.Fatalf("first_row_id %v, want %d", got, *c.want) + } + }) + } +}