diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index e63e9e4a4a..7936df7d26 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/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 { diff --git a/weed/worker/tasks/iceberg/geometry_crs.go b/weed/worker/tasks/iceberg/geometry_crs.go new file mode 100644 index 0000000000..3270aef675 --- /dev/null +++ b/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) +} diff --git a/weed/worker/tasks/iceberg/geometry_crs_test.go b/weed/worker/tasks/iceberg/geometry_crs_test.go new file mode 100644 index 0000000000..04a1df9102 --- /dev/null +++ b/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) + } +} diff --git a/weed/worker/tasks/iceberg/geometry_fixture_test.go b/weed/worker/tasks/iceberg/geometry_fixture_test.go new file mode 100644 index 0000000000..3e5966f0e8 --- /dev/null +++ b/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 +} diff --git a/weed/worker/tasks/iceberg/row_lineage.go b/weed/worker/tasks/iceberg/row_lineage.go index af53329d6e..c751f4a98b 100644 --- a/weed/worker/tasks/iceberg/row_lineage.go +++ b/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 +} diff --git a/weed/worker/tasks/iceberg/row_lineage_test.go b/weed/worker/tasks/iceberg/row_lineage_test.go index 83bc1bed19..35af667a17 100644 --- a/weed/worker/tasks/iceberg/row_lineage_test.go +++ b/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) + } + } +}