iceberg: keep a default-CRS geometry column without a CRS after compaction (#11899)

* iceberg: keep a default-CRS geometry column without a CRS after compaction

Writers such as DuckDB write a GEOMETRY column whose Iceberg type has the
default CRS (OGC:CRS84) with no CRS in the Parquet logical type. parquet-go
spells the default out when it writes such a column (geometryType's
LogicalType), even when the column it read declared none, so every
compacted file declared GEOMETRY(crs=OGC:CRS84) where its inputs declared
GEOMETRY. Both mean the same to the Parquet spec, but a reader may treat the
file column as another type than the table column: DuckDB casts it to the
table's GEOMETRY, and a cast column is no longer pruned by its row-group
statistics, so a spatial filter read every row group of a compacted file.

After a bin is merged, the footer's schema elements of every GEOMETRY
column whose Iceberg type has the default CRS (top-level or nested) are
rewritten to declare no CRS. The data, the page indexes and every offset
stay byte for byte; only the footer is re-encoded. An explicit CRS is left
as it is. Retyping the writer's column instead does not work: parquet-go
refuses to convert between two GEOMETRY types that both mean the default,
and its sorting writer converts.

Tests: geometry_crs_test.go, a bin-packed and a sorted compaction of
default-CRS inputs (one of them spelling the default out) write no CRS,
with a control showing parquet-go spells it out on its own; an explicit CRS
is kept; a nested geometry. On master the binpack and sort cases fail with
CRS "OGC:CRS84".

* iceberg: compress the row lineage columns of compacted v3 files

A v3 compaction appends _row_id and _last_updated_sequence_number to the
merged file. The two leaves were built without a codec, so they were
written uncompressed, 16 bytes a row, whatever codec the data columns use;
on a table of points the compacted files grew by about a third.

The lineage leaves now take the codec of the reference file's first leaf
that declares one, ZSTD when none does.

Test: TestCompactCompressesLineageColumns; on master both columns are
written UNCOMPRESSED next to ZSTD data columns.

* iceberg: match list and map geometry by Iceberg name in files without field ids

A Parquet footer path runs through the repeated group of a LIST or MAP
(shapes.list.element, attrs.key_value.value), which Iceberg leaves out of
its field names (shapes.element, attrs.value). The name lookup for a file
without field ids therefore missed every geometry inside a list or a map
and left the default CRS spelled out. Name the paths as Iceberg does,
including the legacy two-level list forms.

* iceberg: rewrite the geometry CRS footer inside the merged file's buffer

Removing a CRS shortens the footer, so the new one fits where the old one
was. Copying the whole merged file to replace a few hundred bytes of
footer doubled the memory of a compaction near the target file size.

* iceberg: take the lineage codec from the data columns, not an input's _row_id

An input that already carries _row_id may list it before its data
columns, and write it uncompressed; the new lineage columns then copied
that codec instead of the data columns'. Choose from the columns the
layout keeps.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>

weed/worker/tasks/iceberg/compact.go

@@ -300,6 +300,14 @@ func (h *Handler) compactDataFiles(
 			glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err)
 			goto binDone
 		}
+		// A default-CRS geometry column declares no CRS (geometry_crs.go). A
+		// footer that cannot be rewritten leaves the file as written: valid,
+		// only labelled differently.
+		if fixed, _, crsErr := withDefaultGeometryCRSUnset(mergedData, schema); crsErr != nil {
+			glog.Warningf("iceberg compact: geometry CRS left as written in bin %d: %v", binIdx, crsErr)
+		} else {
+			mergedData = fixed
+		}
 
 		// Write merged file to filer
 		if err := ensureFilerDir(ctx, filerClient, dataDir); err != nil {

weed/worker/tasks/iceberg/geometry_crs.go

@@ -0,0 +1,204 @@
+package iceberg
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"strings"
+
+	"github.com/apache/iceberg-go"
+	"github.com/parquet-go/parquet-go/deprecated"
+	"github.com/parquet-go/parquet-go/encoding/thrift"
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// A GEOMETRY column whose Iceberg type has the default CRS (OGC:CRS84) is
+// written with no CRS in the Parquet logical type, as the input files from
+// most writers declare it. parquet-go spells the default out when it writes
+// such a column, even when the column it read declared none. Both mean the
+// same, but a reader may then see the file column as a different type from
+// the table column (DuckDB casts it and no longer prunes row groups by their
+// statistics). An explicit CRS is left as it is.
+//
+// The footer is changed after the file is written rather than by giving the
+// writer another column type: parquet-go refuses to convert between two
+// GEOMETRY types that both mean the default, and its sorting writer
+// converts. Only the footer's schema elements change; the data, the page
+// indexes and every offset into the file stay byte for byte.
+
+var parquetMagic = []byte("PAR1")
+
+// withDefaultGeometryCRSUnset returns data with the CRS of every GEOMETRY
+// column whose Iceberg type has the default CRS left unset in the footer, and
+// whether anything changed. A file that needs no change comes back as it was;
+// so does one whose footer cannot be read (encrypted, not Parquet), with an
+// error. A changed file is written into data's buffer when it fits, so the
+// caller uses the result in place of data.
+func withDefaultGeometryCRSUnset(data []byte, schema *iceberg.Schema) ([]byte, bool, error) {
+	if schema == nil || !hasDefaultCRSGeometry(schema) {
+		return data, false, nil
+	}
+	n := len(data)
+	if n < 12 || !bytes.Equal(data[n-4:], parquetMagic) || !bytes.Equal(data[:4], parquetMagic) {
+		return data, false, fmt.Errorf("not a plaintext parquet file")
+	}
+	footerLen := int(binary.LittleEndian.Uint32(data[n-8 : n-4]))
+	footerStart := n - 8 - footerLen
+	if footerLen <= 0 || footerStart < 4 {
+		return data, false, fmt.Errorf("footer length %d out of range", footerLen)
+	}
+	var protocol thrift.CompactProtocol
+	var md format.FileMetaData
+	if err := thrift.Unmarshal(&protocol, data[footerStart:n-8], &md); err != nil {
+		return data, false, fmt.Errorf("decode footer: %w", err)
+	}
+	changed := false
+	for i, path := range schemaElementPaths(md.Schema) {
+		el := &md.Schema[i]
+		g, ok := el.LogicalType.Value.(*format.GeometryType)
+		if !ok || g.CRS != iceberg.DefaultGeoCRS {
+			continue
+		}
+		var field iceberg.NestedField
+		if el.FieldID > 0 {
+			field, ok = schema.FindFieldByID(int(el.FieldID))
+		} else {
+			field, ok = schema.FindFieldByName(strings.Join(path, "."))
+		}
+		if !ok {
+			continue
+		}
+		if gt, isGeo := field.Type.(iceberg.GeometryType); !isGeo || gt.CRS() != iceberg.DefaultGeoCRS {
+			continue
+		}
+		el.LogicalType = format.LogicalType{Value: &format.GeometryType{}}
+		changed = true
+	}
+	if !changed {
+		return data, false, nil
+	}
+	footer, err := thrift.Marshal(&protocol, &md)
+	if err != nil {
+		return data, false, fmt.Errorf("encode footer: %w", err)
+	}
+	// The new footer is written over the old one in data's own buffer when it
+	// fits, as it does when a CRS is removed: the file is as large as a
+	// compaction target, and copying it would double the memory it takes.
+	out := data[:footerStart]
+	if cap(data) < footerStart+len(footer)+8 {
+		out = make([]byte, footerStart, footerStart+len(footer)+8)
+		copy(out, data[:footerStart])
+	}
+	out = append(out, footer...)
+	out = binary.LittleEndian.AppendUint32(out, uint32(len(footer)))
+	out = append(out, parquetMagic...)
+	return out, true, nil
+}
+
+func hasDefaultCRSGeometry(schema *iceberg.Schema) bool {
+	for _, f := range schema.Fields() {
+		if typeHasDefaultCRSGeometry(f.Type) {
+			return true
+		}
+	}
+	return false
+}
+
+// typeHasDefaultCRSGeometry reports whether the type is, or nests, a geometry
+// with the default CRS.
+func typeHasDefaultCRSGeometry(t iceberg.Type) bool {
+	switch v := t.(type) {
+	case iceberg.GeometryType:
+		return v.CRS() == iceberg.DefaultGeoCRS
+	case *iceberg.StructType:
+		for _, f := range v.FieldList {
+			if typeHasDefaultCRSGeometry(f.Type) {
+				return true
+			}
+		}
+	case *iceberg.ListType:
+		return typeHasDefaultCRSGeometry(v.Element)
+	case *iceberg.MapType:
+		return typeHasDefaultCRSGeometry(v.KeyType) || typeHasDefaultCRSGeometry(v.ValueType)
+	}
+	return false
+}
+
+// schemaElementPaths is each flattened schema element's Iceberg field path
+// from the root (the root's is empty): the footer lists the tree depth first,
+// every group followed by its num_children children. The repeated group of a
+// LIST or MAP is not an Iceberg field, so it takes its parent's path; under it
+// a list's element is named "element" and a map's two children "key" and
+// "value", as Iceberg names them, whatever the writer called them.
+func schemaElementPaths(elements []format.SchemaElement) [][]string {
+	paths := make([][]string, len(elements))
+	var walk func(i int, path []string) int
+	// children walks the n children of a group whose own path is path,
+	// naming the k-th by name(k) when name is set.
+	children := func(next, n int, path []string, name func(k int, el *format.SchemaElement) string) int {
+		for c := 0; c < n && next < len(elements); c++ {
+			child := path
+			if name != nil {
+				child = append(append([]string{}, path...), name(c, &elements[next]))
+			} else {
+				child = append(append([]string{}, path...), elements[next].Name)
+			}
+			next = walk(next, child)
+		}
+		return next
+	}
+	walk = func(i int, path []string) int {
+		if i >= len(elements) {
+			return i
+		}
+		paths[i] = path
+		el := &elements[i]
+		n := int(el.NumChildren.V)
+		if n != 1 || i+1 >= len(elements) {
+			return children(i+1, n, path, nil)
+		}
+		repeated := &elements[i+1]
+		if rt, ok := repeated.RepetitionType.Get(); !ok || rt != format.Repeated {
+			return children(i+1, n, path, nil)
+		}
+		switch {
+		case isListAnnotated(el):
+			// A repeated group of one field wraps the element, unless its
+			// name marks the legacy form where the group is the element.
+			if rn := int(repeated.NumChildren.V); rn == 1 && repeated.Name != "array" && repeated.Name != el.Name+"_tuple" {
+				paths[i+1] = path
+				return children(i+2, 1, path, func(int, *format.SchemaElement) string { return "element" })
+			}
+			return children(i+1, 1, path, func(int, *format.SchemaElement) string { return "element" })
+		case isMapAnnotated(el):
+			if int(repeated.NumChildren.V) == 2 {
+				paths[i+1] = path
+				return children(i+2, 2, path, func(k int, _ *format.SchemaElement) string {
+					if k == 0 {
+						return "key"
+					}
+					return "value"
+				})
+			}
+		}
+		return children(i+1, n, path, nil)
+	}
+	walk(0, nil)
+	return paths
+}
+
+func isListAnnotated(el *format.SchemaElement) bool {
+	if _, ok := el.LogicalType.Value.(*format.ListType); ok {
+		return true
+	}
+	ct, ok := el.ConvertedType.Get()
+	return ok && ct == deprecated.List
+}
+
+func isMapAnnotated(el *format.SchemaElement) bool {
+	if _, ok := el.LogicalType.Value.(*format.MapType); ok {
+		return true
+	}
+	ct, ok := el.ConvertedType.Get()
+	return ok && (ct == deprecated.Map || ct == deprecated.MapKeyValue)
+}

weed/worker/tasks/iceberg/geometry_crs_test.go

@@ -0,0 +1,251 @@
+package iceberg
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
+)
+
+// unsetDefaultCRS makes an input file declare the default CRS as no CRS at
+// all, as DuckDB and other writers do; parquet.Geometry("") spells it out.
+func unsetDefaultCRS(t *testing.T, data []byte) []byte {
+	t.Helper()
+	out, changed, err := withDefaultGeometryCRSUnset(data, geoTestSchema(iceberg.GeometryType{}))
+	if err != nil || !changed {
+		t.Fatalf("unset the input's default CRS: changed %v, %v", changed, err)
+	}
+	return out
+}
+
+func fiveGeoFiles() []geoFile {
+	var files []geoFile
+	for f := 0; f < 5; f++ {
+		var rows []geoRow
+		for i := 0; i < 40; i++ {
+			id := int64(f*40 + i)
+			rows = append(rows, geoRow{ID: id, Geom: wkbPoint(float64(id%20)-10, float64(id/20)-5)})
+		}
+		files = append(files, geoFile{Name: fmt.Sprintf("g%d.parquet", f), Rows: rows})
+	}
+	return files
+}
+
+// The Parquet CRS of a default-CRS geometry column survives compaction as the
+// inputs declared it: none. The control shows what parquet-go does with the
+// same schema on its own: it spells the default out.
+func TestCompactWritesDefaultCRSGeometryWithoutCRS(t *testing.T) {
+	control := unsetDefaultCRS(t, writeGeoParquet(t, []geoRow{{1, wkbPoint(1, 2)}}, ""))
+	if crs, ok := footerGeometryCRS(t, control, "geom"); !ok || crs != "" {
+		t.Fatalf("setup: an input file declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	f, err := parquet.OpenFile(bytes.NewReader(control), int64(len(control)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var plain bytes.Buffer
+	w := parquet.NewWriter(&plain, f.Schema())
+	if _, err := w.WriteRows([]parquet.Row{{parquet.ByteArrayValue(wkbPoint(1, 2)).Level(0, 1, 0), parquet.Int64Value(1).Level(0, 0, 1)}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if crs, _ := footerGeometryCRS(t, plain.Bytes(), "geom"); crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("control: parquet-go copied the schema with CRS %q; the fix is for %q", crs, iceberg.DefaultGeoCRS)
+	}
+
+	for _, strategy := range []string{"binpack", "sort"} {
+		t.Run(strategy, func(t *testing.T) {
+			fs, client := startFakeFiler(t)
+			setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+			opt := geoTableOptions{fixup: unsetDefaultCRS}
+			config := Config{}
+			if strategy == "sort" {
+				so, err := table.NewSortOrder(1, []table.SortField{{SourceIDs: []int{geoTestIDField}, Transform: iceberg.IdentityTransform{}, Direction: table.SortASC, NullOrder: table.NullsFirst}})
+				if err != nil {
+					t.Fatal(err)
+				}
+				opt.sortOrder = &so
+				config.RewriteStrategy, config.SortBufferRows = "sort", minSortBufferRows
+			}
+			files := fiveGeoFiles()
+			files[2].ParquetCRSSet = true // an input that spells the default out
+			populateGeoTable(t, fs, setup, files, opt)
+			if result := compactGeo(t, client, setup, config); !strings.Contains(result, "compacted 5 files") {
+				t.Fatalf("result %q", result)
+			}
+			for name, data := range compactedFiles(t, fs, setup) {
+				if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != "" {
+					t.Errorf("%s: geometry column declares CRS %q (geometry %v), want none", name, crs, ok)
+				}
+				got, err := parquet.Read[struct {
+					ID int64 `parquet:"id"`
+				}](bytes.NewReader(data), int64(len(data)))
+				if err != nil || len(got) != 200 {
+					t.Errorf("%s: the rewritten file reads %d rows, %v", name, len(got), err)
+				}
+			}
+		})
+	}
+}
+
+// An explicit CRS is not the default: it is written as the input declared it.
+func TestCompactKeepsExplicitGeometryCRS(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	g, err := iceberg.GeometryTypeOf("EPSG:2913")
+	if err != nil {
+		t.Fatal(err)
+	}
+	populateGeoTable(t, fs, setup, fiveGeoFiles(), geoTableOptions{icebergType: g, parquetCRS: "EPSG:2913"})
+	compactGeo(t, client, setup, Config{})
+	for name, data := range compactedFiles(t, fs, setup) {
+		if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != "EPSG:2913" {
+			t.Errorf("%s: geometry CRS %q (geometry %v), want EPSG:2913", name, crs, ok)
+		}
+	}
+}
+
+// A default-CRS geometry nested in a struct is written without a CRS too.
+func TestDefaultCRSUnsetForNestedGeometry(t *testing.T) {
+	schema := parquet.NewSchema("table", parquet.Group{
+		"id":  parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), 1),
+		"loc": parquet.FieldID(parquet.Group{"geom": parquet.FieldID(parquet.Optional(parquet.Geometry("")), 3)}, 2),
+	})
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[any](&buf, schema)
+	cols := map[string]int{}
+	for i, p := range schema.Columns() {
+		cols[strings.Join(p, ".")] = i
+	}
+	row := make(parquet.Row, 2)
+	row[cols["id"]] = parquet.Int64Value(1).Level(0, 0, cols["id"])
+	row[cols["loc.geom"]] = parquet.ByteArrayValue(wkbPoint(1, 2)).Level(0, 1, cols["loc.geom"])
+	if _, err := w.WriteRows([]parquet.Row{row}); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if crs, ok := footerGeometryCRS(t, buf.Bytes(), "geom"); !ok || crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("setup: the nested column declares %q (geometry %v); the control is the spelled-out default", crs, ok)
+	}
+	ice := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
+		iceberg.NestedField{ID: 2, Name: "loc", Type: &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 3, Name: "geom", Type: iceberg.GeometryType{}}}}})
+	out, changed, err := withDefaultGeometryCRSUnset(buf.Bytes(), ice)
+	if err != nil || !changed {
+		t.Fatalf("changed %v, %v", changed, err)
+	}
+	if crs, ok := footerGeometryCRS(t, out, "geom"); !ok || crs != "" {
+		t.Errorf("the nested geometry declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	got, err := parquet.Read[struct {
+		ID int64 `parquet:"id"`
+	}](bytes.NewReader(out), int64(len(out)))
+	if err != nil || len(got) != 1 || got[0].ID != 1 {
+		t.Errorf("the rewritten file reads %v, %v", got, err)
+	}
+}
+
+// footerGeometryCRSs is the CRS of every GEOMETRY element in the footer, by
+// schema element index and name.
+func footerGeometryCRSs(t *testing.T, data []byte) map[string]string {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	out := map[string]string{}
+	for i, el := range f.Metadata().Schema {
+		if g, ok := el.LogicalType.Value.(*format.GeometryType); ok {
+			out[fmt.Sprintf("%d:%s", i, el.Name)] = g.CRS
+		}
+	}
+	return out
+}
+
+// A file without field ids is matched to the table by name: through the
+// LIST and MAP wrapper groups too, which Iceberg's field names leave out.
+func TestDefaultCRSUnsetWithoutFieldIDs(t *testing.T) {
+	geom := func() parquet.Node { return parquet.Optional(parquet.Geometry("")) }
+	element := func(id int, typ iceberg.Type) *iceberg.ListType {
+		return &iceberg.ListType{ElementID: id, Element: typ, ElementRequired: false}
+	}
+	cases := []struct {
+		name    string
+		parquet parquet.Group
+		iceberg []iceberg.NestedField
+	}{
+		{"top level", parquet.Group{"id": parquet.Int(64), "geom": geom()},
+			[]iceberg.NestedField{{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, {ID: 2, Name: "geom", Type: iceberg.GeometryType{}}}},
+		{"list", parquet.Group{"shapes": parquet.Optional(parquet.List(geom()))},
+			[]iceberg.NestedField{{ID: 1, Name: "shapes", Type: element(2, iceberg.GeometryType{})}}},
+		{"list of structs", parquet.Group{"items": parquet.Optional(parquet.List(parquet.Group{"geom": geom()}))},
+			[]iceberg.NestedField{{ID: 1, Name: "items", Type: element(2, &iceberg.StructType{FieldList: []iceberg.NestedField{{ID: 3, Name: "geom", Type: iceberg.GeometryType{}}}})}}},
+		{"map", parquet.Group{"attrs": parquet.Optional(parquet.Map(parquet.String(), geom()))},
+			[]iceberg.NestedField{{ID: 1, Name: "attrs", Type: &iceberg.MapType{KeyID: 2, KeyType: iceberg.PrimitiveTypes.String, ValueID: 3, ValueType: iceberg.GeometryType{}}}}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var buf bytes.Buffer
+			w := parquet.NewGenericWriter[any](&buf, parquet.NewSchema("table", tc.parquet))
+			if err := w.Close(); err != nil {
+				t.Fatal(err)
+			}
+			before := footerGeometryCRSs(t, buf.Bytes())
+			if len(before) != 1 {
+				t.Fatalf("setup: geometry columns %v, want one", before)
+			}
+			for path, crs := range before {
+				if crs != iceberg.DefaultGeoCRS {
+					t.Fatalf("setup: %s declares %q; the control is the spelled-out default", path, crs)
+				}
+			}
+			out, changed, err := withDefaultGeometryCRSUnset(buf.Bytes(), iceberg.NewSchema(0, tc.iceberg...))
+			if err != nil || !changed {
+				t.Fatalf("changed %v, %v", changed, err)
+			}
+			for path, crs := range footerGeometryCRSs(t, out) {
+				if crs != "" {
+					t.Errorf("%s declares CRS %q, want none", path, crs)
+				}
+			}
+		})
+	}
+}
+
+// The footer is replaced inside the file's own buffer: a compacted file can be
+// as large as the target file size, and a copy of it would double the memory
+// the rewrite needs.
+func TestDefaultCRSUnsetReusesBuffer(t *testing.T) {
+	data := writeGeoParquet(t, []geoRow{{1, wkbPoint(1, 2)}, {2, wkbPoint(3, 4)}}, "")
+	if crs, ok := footerGeometryCRS(t, data, "geom"); !ok || crs != iceberg.DefaultGeoCRS {
+		t.Fatalf("setup: CRS %q (geometry %v); the control is the spelled-out default", crs, ok)
+	}
+	first := &data[0]
+	out, changed, err := withDefaultGeometryCRSUnset(data, geoTestSchema(iceberg.GeometryType{}))
+	if err != nil || !changed {
+		t.Fatalf("changed %v, %v", changed, err)
+	}
+	if &out[0] != first {
+		t.Errorf("the rewrite copied the file into a new buffer")
+	}
+	if crs, ok := footerGeometryCRS(t, out, "geom"); !ok || crs != "" {
+		t.Errorf("geometry declares CRS %q (geometry %v), want none", crs, ok)
+	}
+	got, err := parquet.Read[struct {
+		ID   int64  `parquet:"id"`
+		Geom []byte `parquet:"geom,optional"`
+	}](bytes.NewReader(out), int64(len(out)))
+	if err != nil || len(got) != 2 || got[1].ID != 2 || !bytes.Equal(got[1].Geom, wkbPoint(3, 4)) {
+		t.Errorf("the rewritten file reads %v, %v", got, err)
+	}
+}

weed/worker/tasks/iceberg/geometry_fixture_test.go

@@ -0,0 +1,360 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"math"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+// Fixtures for v3 tables with a geometry column: Parquet files with field
+// ids, ZSTD and a native GEOMETRY column, and manifests that carry the v3
+// geometry bounds of each file, as DuckDB writes them.
+
+const (
+	geoTestIDField   = 1
+	geoTestGeomField = 2
+)
+
+func wkbPoint(x, y float64) []byte {
+	b := make([]byte, 21)
+	b[0] = 1 // little endian
+	binary.LittleEndian.PutUint32(b[1:], 1)
+	binary.LittleEndian.PutUint64(b[5:], math.Float64bits(x))
+	binary.LittleEndian.PutUint64(b[13:], math.Float64bits(y))
+	return b
+}
+
+// wkbBox is the polygon of one closed ring around the box.
+func wkbBox(xmin, ymin, xmax, ymax float64) []byte {
+	ring := [][2]float64{{xmin, ymin}, {xmax, ymin}, {xmax, ymax}, {xmin, ymax}, {xmin, ymin}}
+	b := make([]byte, 0, 13+16*len(ring))
+	b = append(b, 1)
+	b = binary.LittleEndian.AppendUint32(b, 3)
+	b = binary.LittleEndian.AppendUint32(b, 1)
+	b = binary.LittleEndian.AppendUint32(b, uint32(len(ring)))
+	for _, p := range ring {
+		b = binary.LittleEndian.AppendUint64(b, math.Float64bits(p[0]))
+		b = binary.LittleEndian.AppendUint64(b, math.Float64bits(p[1]))
+	}
+	return b
+}
+
+func geoLE64(v int64) []byte {
+	b := make([]byte, 8)
+	binary.LittleEndian.PutUint64(b, uint64(v))
+	return b
+}
+
+func xyBound(x, y float64) []byte {
+	b := make([]byte, 16)
+	binary.LittleEndian.PutUint64(b, math.Float64bits(x))
+	binary.LittleEndian.PutUint64(b[8:], math.Float64bits(y))
+	return b
+}
+
+func decodeXYBound(t *testing.T, b []byte) (float64, float64) {
+	t.Helper()
+	if len(b) < 16 {
+		t.Fatalf("geometry bound of %d bytes, want 16", len(b))
+	}
+	return math.Float64frombits(binary.LittleEndian.Uint64(b)), math.Float64frombits(binary.LittleEndian.Uint64(b[8:]))
+}
+
+type geoRow struct {
+	ID   int64
+	Geom []byte // nil: a null geometry
+}
+
+type geoFile struct {
+	Name          string
+	Rows          []geoRow
+	ParquetCRSSet bool // declare the default CRS in the Parquet logical type
+}
+
+type geoTableOptions struct {
+	icebergType iceberg.Type     // the geometry column's Iceberg type (default: geometry with the default CRS)
+	parquetCRS  string           // the CRS of the Parquet GEOMETRY type
+	sortOrder   *table.SortOrder // nil: unsorted
+	properties  iceberg.Properties
+	// fixup, when set, rewrites each input file after it is written
+	fixup func(t *testing.T, data []byte) []byte
+}
+
+func geoTestSchema(geomType iceberg.Type) *iceberg.Schema {
+	return iceberg.NewSchema(0,
+		iceberg.NestedField{ID: geoTestIDField, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
+		iceberg.NestedField{ID: geoTestGeomField, Type: geomType, Name: "geom", Required: false},
+	)
+}
+
+func writeGeoParquet(t *testing.T, rows []geoRow, crs string) []byte {
+	t.Helper()
+	schema := parquet.NewSchema("table", parquet.Group{
+		"id":   parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), geoTestIDField),
+		"geom": parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Geometry(crs)), &zstd.Codec{}), geoTestGeomField),
+	})
+	geomCol, idCol := -1, -1
+	for i, p := range schema.Columns() {
+		switch strings.Join(p, ".") {
+		case "geom":
+			geomCol = i
+		case "id":
+			idCol = i
+		}
+	}
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[any](&buf, schema)
+	for _, r := range rows {
+		row := make(parquet.Row, 2)
+		if r.Geom == nil {
+			row[geomCol] = parquet.NullValue().Level(0, 0, geomCol)
+		} else {
+			row[geomCol] = parquet.ByteArrayValue(r.Geom).Level(0, 1, geomCol)
+		}
+		row[idCol] = parquet.Int64Value(r.ID).Level(0, 0, idCol)
+		if _, err := w.WriteRows([]parquet.Row{row}); err != nil {
+			t.Fatalf("write row: %v", err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close writer: %v", err)
+	}
+	return buf.Bytes()
+}
+
+// footerGeoBox is the union of the file's row-group geometry boxes.
+func footerGeoBox(t *testing.T, data []byte) (xmin, ymin, xmax, ymax float64, ok bool) {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	xmin, ymin, xmax, ymax = math.Inf(1), math.Inf(1), math.Inf(-1), math.Inf(-1)
+	for _, rg := range f.Metadata().RowGroups {
+		for _, c := range rg.Columns {
+			if strings.Join(c.MetaData.PathInSchema, ".") != "geom" || len(c.MetaData.GeospatialStatistics.GeoSpatialTypes) == 0 {
+				continue
+			}
+			bb := c.MetaData.GeospatialStatistics.BBox
+			xmin, ymin = math.Min(xmin, bb.XMin), math.Min(ymin, bb.YMin)
+			xmax, ymax = math.Max(xmax, bb.XMax), math.Max(ymax, bb.YMax)
+			ok = true
+		}
+	}
+	return
+}
+
+// footerGeometryCRS is the CRS the file's GEOMETRY logical type declares, and
+// whether the column is GEOMETRY at all.
+func footerGeometryCRS(t *testing.T, data []byte, column string) (string, bool) {
+	t.Helper()
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	for _, el := range f.Metadata().Schema {
+		if el.Name != column {
+			continue
+		}
+		if g, ok := el.LogicalType.Value.(*format.GeometryType); ok {
+			return g.CRS, true
+		}
+		return "", false
+	}
+	t.Fatalf("no column %s in the footer", column)
+	return "", false
+}
+
+// populateGeoTable writes a v3 geometry table with one append snapshot and
+// one manifest per file.
+func populateGeoTable(t *testing.T, fs *fakeFilerServer, setup tableSetup, files []geoFile, opt geoTableOptions) table.Metadata {
+	t.Helper()
+	geomType := opt.icebergType
+	if geomType == nil {
+		geomType = iceberg.GeometryType{}
+	}
+	schema := geoTestSchema(geomType)
+	spec := *iceberg.UnpartitionedSpec
+	location := "s3://" + setup.BucketName + "/" + setup.dataPath()
+	props := iceberg.Properties{"format-version": "3"}
+	for k, v := range opt.properties {
+		props[k] = v
+	}
+	sortOrder := table.UnsortedSortOrder
+	if opt.sortOrder != nil {
+		sortOrder = *opt.sortOrder
+	}
+	meta, err := table.NewMetadata(schema, &spec, sortOrder, location, props)
+	if err != nil {
+		t.Fatalf("create metadata: %v", err)
+	}
+	bucketPath := path.Join(s3tables.TablesPath, setup.BucketName)
+	tableFilerPath := path.Join(bucketPath, setup.dataPath())
+	metaDir := path.Join(tableFilerPath, "metadata")
+	dataDir := path.Join(tableFilerPath, "data")
+	putFile := func(dir, name string, data []byte) {
+		fs.putEntry(dir, name, &filer_pb.Entry{Name: name, Content: data,
+			Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(data))}})
+	}
+	snapID := int64(1)
+	var manifests []iceberg.ManifestFile
+	for i, f := range files {
+		crs := opt.parquetCRS
+		if f.ParquetCRSSet && crs == "" {
+			crs = iceberg.DefaultGeoCRS
+		}
+		data := writeGeoParquet(t, f.Rows, crs)
+		if opt.fixup != nil && !f.ParquetCRSSet {
+			data = opt.fixup(t, data)
+		}
+		putFile(dataDir, f.Name, data)
+		dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, setup.fileRef("data", f.Name), iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(f.Rows)), int64(len(data)))
+		if err != nil {
+			t.Fatalf("data file %s: %v", f.Name, err)
+		}
+		lo, hi := int64(math.MaxInt64), int64(math.MinInt64)
+		for _, r := range f.Rows {
+			lo, hi = min(lo, r.ID), max(hi, r.ID)
+		}
+		lower := map[int][]byte{geoTestIDField: geoLE64(lo)}
+		upper := map[int][]byte{geoTestIDField: geoLE64(hi)}
+		if xmin, ymin, xmax, ymax, ok := footerGeoBox(t, data); ok {
+			lower[geoTestGeomField], upper[geoTestGeomField] = xyBound(xmin, ymin), xyBound(xmax, ymax)
+		}
+		dfb.LowerBoundValues(lower)
+		dfb.UpperBoundValues(upper)
+		entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
+		name := fmt.Sprintf("data-manifest-%d.avro", i)
+		var buf bytes.Buffer
+		mf, err := iceberg.WriteManifest(setup.fileRef("metadata", name), &buf, 3, spec, schema, snapID, []iceberg.ManifestEntry{entry})
+		if err != nil {
+			t.Fatalf("write manifest: %v", err)
+		}
+		putFile(metaDir, name, buf.Bytes())
+		manifests = append(manifests, mf)
+	}
+	var mlBuf bytes.Buffer
+	mlw, err := iceberg.NewManifestListWriterV3(&mlBuf, snapID, 1, 0, nil)
+	if err != nil {
+		t.Fatalf("manifest list writer: %v", err)
+	}
+	if err := mlw.AddManifests(manifests); err != nil {
+		t.Fatalf("add manifests: %v", err)
+	}
+	if err := mlw.Close(); err != nil {
+		t.Fatalf("close manifest list: %v", err)
+	}
+	putFile(metaDir, "snap-1.avro", mlBuf.Bytes())
+	firstRowID := int64(0)
+	addedRows := *mlw.NextRowID()
+	snap := table.Snapshot{SnapshotID: snapID, SequenceNumber: 1, TimestampMs: time.Now().UnixMilli(),
+		ManifestList: setup.fileRef("metadata", "snap-1.avro"),
+		Summary:      &table.Summary{Operation: table.OpAppend, Properties: map[string]string{}},
+		FirstRowID:   &firstRowID, AddedRows: &addedRows}
+	builder, err := table.MetadataBuilderFromBase(meta, location)
+	if err != nil {
+		t.Fatalf("metadata builder: %v", err)
+	}
+	if err := builder.AddSnapshot(&snap); err != nil {
+		t.Fatalf("add snapshot: %v", err)
+	}
+	if err := builder.SetSnapshotRef(table.MainBranch, snapID, table.BranchRef); err != nil {
+		t.Fatalf("set ref: %v", err)
+	}
+	if meta, err = builder.Build(); err != nil {
+		t.Fatalf("build metadata: %v", err)
+	}
+	full, _ := json.Marshal(meta)
+	xattr, _ := json.Marshal(map[string]interface{}{
+		"metadataVersion":  1,
+		"metadataLocation": setup.fileRef("metadata", "v1.metadata.json"),
+		"metadata":         map[string]interface{}{"fullMetadata": json.RawMessage(full)},
+	})
+	fs.putEntry(s3tables.TablesPath, setup.BucketName, &filer_pb.Entry{
+		Name: setup.BucketName, IsDirectory: true, Extended: map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")},
+	})
+	fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{Name: setup.Namespace, IsDirectory: true})
+	fs.putEntry(path.Join(bucketPath, setup.Namespace), setup.TableName, &filer_pb.Entry{
+		Name: setup.TableName, IsDirectory: true, Extended: map[string][]byte{s3tables.ExtendedKeyMetadata: xattr},
+	})
+	return meta
+}
+
+func compactGeo(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, config Config) string {
+	t.Helper()
+	if config.TargetFileSizeBytes == 0 {
+		config.TargetFileSizeBytes = 256 * 1024 * 1024
+	}
+	if config.MinInputFiles == 0 {
+		config.MinInputFiles = 2
+	}
+	if config.MaxCommitRetries == 0 {
+		config.MaxCommitRetries = 3
+	}
+	config.ApplyDeletes = true
+	result, _, err := NewHandler(nil).compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	return result
+}
+
+// compactedFiles is every compact-*.parquet in the table's data directory.
+func compactedFiles(t *testing.T, fs *fakeFilerServer, setup tableSetup) map[string][]byte {
+	t.Helper()
+	out := map[string][]byte{}
+	for _, e := range fs.listDir(path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")) {
+		if strings.HasPrefix(e.Name, "compact-") {
+			out[e.Name] = e.Content
+		}
+	}
+	if len(out) == 0 {
+		t.Fatal("no compacted file")
+	}
+	return out
+}
+
+// currentDataEntries is every entry of the current snapshot's data manifests.
+func currentDataEntries(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) []iceberg.ManifestEntry {
+	t.Helper()
+	ctx := context.Background()
+	state, err := loadCurrentMetadata(ctx, client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("loadCurrentMetadata: %v", err)
+	}
+	manifests, err := loadCurrentManifests(ctx, client, setup.BucketName, state.DataPath, state.Metadata)
+	if err != nil {
+		t.Fatalf("loadCurrentManifests: %v", err)
+	}
+	var out []iceberg.ManifestEntry
+	for _, mf := range manifests {
+		if mf.ManifestContent() != iceberg.ManifestContentData {
+			continue
+		}
+		data, err := loadFileByIcebergPath(ctx, client, setup.BucketName, state.DataPath, mf.FilePath())
+		if err != nil {
+			t.Fatalf("load manifest: %v", err)
+		}
+		entries, err := iceberg.ReadManifest(mf, bytes.NewReader(data), true)
+		if err != nil {
+			t.Fatalf("read manifest: %v", err)
+		}
+		out = append(out, entries...)
+	}
+	return out
+}

weed/worker/tasks/iceberg/row_lineage.go

@@ -10,6 +10,8 @@ import (
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/table"
 	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress"
+	"github.com/parquet-go/parquet-go/compress/zstd"
 )
 
 // Row lineage (Iceberg v3, spec "Row Lineage" and "Snapshot Row IDs").
@@ -170,9 +172,16 @@ func newLineageLayout(reference *parquet.Schema) (*lineageLayout, error) {
 		offsets = append(offsets, leaves)
 		leaves += leafCount(f)
 	}
+	// The lineage columns are compressed like the reference's data columns
+	// (not like a lineage column it already has): a leaf that declares no
+	// codec is written uncompressed, 16 bytes a row.
+	codec := lineageCodec(base)
+	lineageLeaf := func(id int) parquet.Node {
+		return parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Leaf(parquet.Int64Type)), codec), id)
+	}
 	fields := append(append([]parquet.Field{}, base...),
-		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.RowIDFieldID), name: iceberg.RowIDColumnName},
-		lineageField{Node: parquet.FieldID(parquet.Optional(parquet.Leaf(parquet.Int64Type)), iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName},
+		lineageField{Node: lineageLeaf(iceberg.RowIDFieldID), name: iceberg.RowIDColumnName},
+		lineageField{Node: lineageLeaf(iceberg.LastUpdatedSequenceNumberFieldID), name: iceberg.LastUpdatedSequenceNumberColumnName},
 	)
 	return &lineageLayout{
 		schema:      parquet.NewSchema(reference.Name(), &orderedGroup{Node: reference, fields: fields}),
@@ -392,3 +401,26 @@ func mergedFileFirstRowID(data []byte) (*int64, error) {
 	}
 	return nil, nil
 }
+
+// lineageCodec is the codec of the first data column leaf that declares one,
+// ZSTD when none does.
+func lineageCodec(base []parquet.Field) compress.Codec {
+	for _, f := range base {
+		if c := columnCodec(f); c != nil {
+			return c
+		}
+	}
+	return &zstd.Codec{}
+}
+
+func columnCodec(n parquet.Node) compress.Codec {
+	if n.Leaf() {
+		return n.Compression()
+	}
+	for _, f := range n.Fields() {
+		if c := columnCodec(f); c != nil {
+			return c
+		}
+	}
+	return nil
+}

weed/worker/tasks/iceberg/row_lineage_test.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/parquet-go/parquet-go/format"
 
 	filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
@@ -382,3 +384,59 @@ func TestCompactDataFilesV3RowLineageManifestRead(t *testing.T) {
 		}
 	}
 }
+
+// The lineage columns a v3 compaction adds are compressed like the data
+// columns.
+func TestCompactCompressesLineageColumns(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "geo"}
+	populateGeoTable(t, fs, setup, fiveGeoFiles(), geoTableOptions{})
+	compactGeo(t, client, setup, Config{})
+	for name, data := range compactedFiles(t, fs, setup) {
+		f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		seen := 0
+		for _, rg := range f.Metadata().RowGroups {
+			for _, c := range rg.Columns {
+				col := strings.Join(c.MetaData.PathInSchema, ".")
+				if col != iceberg.RowIDColumnName && col != iceberg.LastUpdatedSequenceNumberColumnName {
+					continue
+				}
+				seen++
+				if c.MetaData.Codec != format.Zstd {
+					t.Errorf("%s: %s written with codec %v, want ZSTD like the data columns", name, col, c.MetaData.Codec)
+				}
+			}
+		}
+		if seen == 0 {
+			t.Errorf("%s: no lineage column", name)
+		}
+	}
+}
+
+// The lineage codec follows the data columns: an input whose own _row_id
+// comes first and is uncompressed does not leave the new lineage columns
+// uncompressed.
+func TestLineageCodecFollowsDataColumns(t *testing.T) {
+	schema := parquet.NewSchema("table", parquet.Group{
+		iceberg.RowIDColumnName: parquet.FieldID(parquet.Compressed(parquet.Optional(parquet.Int(64)), &parquet.Uncompressed), iceberg.RowIDFieldID),
+		"id":                    parquet.FieldID(parquet.Compressed(parquet.Int(64), &zstd.Codec{}), 1),
+	})
+	if first := schema.Fields()[0].Name(); first != iceberg.RowIDColumnName {
+		t.Fatalf("setup: first column %s, want %s", first, iceberg.RowIDColumnName)
+	}
+	layout, err := newLineageLayout(schema)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, f := range layout.schema.Fields() {
+		if f.Name() != iceberg.RowIDColumnName && f.Name() != iceberg.LastUpdatedSequenceNumberColumnName {
+			continue
+		}
+		if c := f.Compression(); c == nil || c.CompressionCodec() != format.Zstd {
+			t.Errorf("%s compressed with %v, want ZSTD like the data column", f.Name(), c)
+		}
+	}
+}