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 <jatorre@lakehousebox.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

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 {

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)
+			}
+		})
+	}
+}