From ecd778a4a0c62790c51b9352033d49a6597d22ba Mon Sep 17 00:00:00 2001 From: Javier de la Torre Date: Sat, 24 Oct 2026 07:53:43 +0200 Subject: [PATCH] iceberg maintenance: a merged file with null row ids names no first_row_id (#11973) * iceberg maintenance: a merged file with null row ids names no first_row_id Compaction stamps a merged v3 data file with the _row_id of its first row as first_row_id (mergedFileFirstRowID). When the bin mixed rows that carry a materialized _row_id with rows that carry none (a data file with no first_row_id, as a v2 table's files are after an upgrade to v3, gives null ids), the null rows then inherit first_row_id + their position in the file: ids that other rows already hold, so two rows answer to the same row id. A merged file now names a first_row_id only when its footer shows no null _row_id in any row group. Otherwise the commit assigns the file a fresh range from the table's next-row-id: the null rows get new, unused ids and the materialized ones keep theirs (a materialized value overrides inheritance). Test: TestMergedFileFirstRowIDOnlyWithoutNullIDs ("materialized first, a null later" fails without the fix). * iceberg maintenance: name no first_row_id when the merged file has a null _row_id mergedFileFirstRowID stamped the merged file with its first row's id without checking later rows. A null _row_id in a merged file inherits first_row_id+position, an id another row already holds, so the commit produced two rows answering to the same row id. Check the footer's null counts for the _row_id column across row groups first: any null means the file is committed without a first_row_id, so its nulls take fresh ids from the table's next-row-id while materialized values still override inheritance. * shell: retry filer discovery until master knows a filer A shell started against a freshly booted cluster (weed mini, the s3tables/lance test harnesses) can list cluster nodes before the filer's first heartbeat lands on the master. The empty list left FilerAddress unset and the next command died with "dial tcp: missing address". Poll ListClusterNodes until a filer appears or the connect timeout expires. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: discover the filer on first use instead of at startup The startup retry loop made every weed shell invocation wait out the connect timeout on clusters that have no filer, even for commands that only talk to the master. Discovery now runs once, on the first command that actually needs a filer, keeping the retry-until-registered fix for the fresh-cluster race without stalling master-only commands. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * shell: stop filer discovery retries on non-transient errors A master that keeps answering ListClusterNodes with a permanent error (permission denied, unimplemented) cannot gain a filer by retrying, so discovery returns instead of polling until the deadline. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Javier de la Torre Co-authored-by: Chris Lu Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Chris Lu --- weed/worker/tasks/iceberg/row_lineage.go | 35 +++++++++-- .../iceberg/row_lineage_first_id_test.go | 63 +++++++++++++++++++ 2 files changed, 93 insertions(+), 5 deletions(-) create mode 100644 weed/worker/tasks/iceberg/row_lineage_first_id_test.go 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) + } + }) + } +}