From e9c280e09545bddbf6ae29cd682b319bd438e4d0 Mon Sep 17 00:00:00 2001 From: Javier de la Torre Date: Thu, 29 Oct 2026 08:12:46 +0100 Subject: [PATCH] iceberg maintenance: bound decimal, time, uuid, fixed and rescaled timestamp columns on compacted files (#12190) * iceberg maintenance: bound decimal, time, uuid, fixed and rescaled timestamp columns on compacted files The column statistics written for a compacted data file give lower and upper bounds only for booleans, integers, floats, dates, strings, binaries and timestamps already stored in the Iceberg type's unit. Every other column is left unbounded, so a reader can never skip a compacted file on it: a DECIMAL(18,2) column written by DuckDB (stored as INT64) loses its bounds after compaction, as does a millisecond timestamp, a TIME or a UUID. boundLiteral now also converts: - DECIMAL stored as INT32 or INT64 at the column's scale. A decimal stored as a byte array still gets no bound: Parquet compares those statistics as bytes, which does not order negative two's-complement values; - TIME, TIMESTAMP, TIMESTAMPTZ and TIMESTAMP_NS from any Parquet time unit, rescaled to the Iceberg unit; a value rescaled to a coarser unit is rounded down for a lower bound and up for an upper bound so it stays a bound, and a value that would overflow gets none; - UUID and FIXED[n] from a FIXED_LEN_BYTE_ARRAY of the right length; - a DOUBLE column read from a FLOAT file. arrow-go becomes a direct dependency (decimal128, already in the module graph through iceberg-go). * iceberg maintenance: bound TIME(MILLIS) columns stored as INT32 * iceberg maintenance: bound timestamptz_ns columns too --------- Co-authored-by: user.email --- go.mod | 2 +- weed/worker/tasks/iceberg/compact_stats.go | 126 ++++++++++++-- .../tasks/iceberg/compact_stats_types_test.go | 162 ++++++++++++++++++ 3 files changed, 276 insertions(+), 14 deletions(-) create mode 100644 weed/worker/tasks/iceberg/compact_stats_types_test.go diff --git a/go.mod b/go.mod index a369039ea8..37337b465e 100644 --- a/go.mod +++ b/go.mod @@ -117,6 +117,7 @@ require ( github.com/Jille/raft-grpc-transport v1.6.1 github.com/ThreeDotsLabs/watermill v1.5.3 github.com/a-h/templ v0.3.1100 + github.com/apache/arrow-go/v18 v18.8.0 github.com/apache/cassandra-gocql-driver/v2 v2.1.2 github.com/apache/iceberg-go v0.7.0 github.com/apple/foundationdb/bindings/go v0.0.0-20250911184653-27f7192f47c3 @@ -181,7 +182,6 @@ require ( github.com/adrg/xdg v0.5.3 // indirect github.com/anchore/go-lzo v0.1.1 // indirect github.com/antlr4-go/antlr/v4 v4.13.1 // indirect - github.com/apache/arrow-go/v18 v18.8.0 // indirect github.com/apache/thrift v0.24.0 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.10.1 // indirect github.com/bahlo/generic-list-go v0.2.0 // indirect diff --git a/weed/worker/tasks/iceberg/compact_stats.go b/weed/worker/tasks/iceberg/compact_stats.go index 277752c976..601043d4c0 100644 --- a/weed/worker/tasks/iceberg/compact_stats.go +++ b/weed/worker/tasks/iceberg/compact_stats.go @@ -8,7 +8,9 @@ import ( "time" "unicode/utf8" + "github.com/apache/arrow-go/v18/arrow/decimal128" "github.com/apache/iceberg-go" + "github.com/google/uuid" "github.com/parquet-go/parquet-go" "github.com/parquet-go/parquet-go/format" ) @@ -125,8 +127,8 @@ func collectColumnStats(data []byte, schema *iceberg.Schema) (*columnStats, erro if b.invalid || !b.seen { continue } - lo, okLo := boundLiteral(b.min, b.typ, b.field.Type) - hi, okHi := boundLiteral(b.max, b.typ, b.field.Type) + lo, okLo := boundLiteral(b.min, b.typ, b.field.Type, false) + hi, okHi := boundLiteral(b.max, b.typ, b.field.Type, true) if !okLo || !okHi { continue } @@ -182,13 +184,16 @@ func rowGroupOffset(rg *format.RowGroup) int64 { } // boundLiteral converts a Parquet statistics value into a literal of the -// column's Iceberg type, when the conversion is exact. Other types (decimal, -// time, uuid, fixed, timestamps not in the type's unit) get no bound. -func boundLiteral(v parquet.Value, pt parquet.Type, t iceberg.Type) (iceberg.Literal, bool) { +// column's Iceberg type, when the conversion is exact or rounds towards the +// safe side of the bound: roundUp is true for an upper bound, so a time value +// rescaled to a coarser unit stays a bound. Values a reader could not order the +// same way as the writer (NaN, invalid UTF-8, a decimal at another scale or +// stored as bytes, an unknown time unit) get no bound. +func boundLiteral(v parquet.Value, pt parquet.Type, t iceberg.Type, roundUp bool) (iceberg.Literal, bool) { if v.IsNull() { return nil, false } - switch t.(type) { + switch t := t.(type) { case iceberg.BooleanType: return iceberg.BoolLiteral(v.Boolean()), v.Kind() == parquet.Boolean case iceberg.Int32Type: @@ -204,33 +209,128 @@ func boundLiteral(v parquet.Value, pt parquet.Type, t iceberg.Type) (iceberg.Lit f := v.Float() return iceberg.Float32Literal(f), v.Kind() == parquet.Float && !math.IsNaN(float64(f)) case iceberg.Float64Type: - f := v.Double() - return iceberg.Float64Literal(f), v.Kind() == parquet.Double && !math.IsNaN(f) + switch v.Kind() { + case parquet.Double: + f := v.Double() + return iceberg.Float64Literal(f), !math.IsNaN(f) + case parquet.Float: + f := v.Float() + return iceberg.Float64Literal(float64(f)), !math.IsNaN(float64(f)) + } case iceberg.DateType: return iceberg.DateLiteral(iceberg.Date(v.Int32())), v.Kind() == parquet.Int32 + case iceberg.TimeType: + var i64 int64 + switch v.Kind() { + case parquet.Int64: + i64 = v.Int64() + case parquet.Int32: // TIME(MILLIS) is stored as INT32 + i64 = int64(v.Int32()) + default: + return nil, false + } + if us, ok := convertTimeUnit(i64, pt, time.Microsecond, roundUp); ok { + return iceberg.TimeLiteral(iceberg.Time(us)), true + } case iceberg.TimestampType, iceberg.TimestampTzType: - return iceberg.TimestampLiteral(iceberg.Timestamp(v.Int64())), v.Kind() == parquet.Int64 && timeUnit(pt) == time.Microsecond - case iceberg.TimestampNsType: - return iceberg.TimestampNsLiteral(iceberg.TimestampNano(v.Int64())), v.Kind() == parquet.Int64 && timeUnit(pt) == time.Nanosecond + if v.Kind() == parquet.Int64 { + if us, ok := convertTimeUnit(v.Int64(), pt, time.Microsecond, roundUp); ok { + return iceberg.TimestampLiteral(iceberg.Timestamp(us)), true + } + } + case iceberg.TimestampNsType, iceberg.TimestampTzNsType: + if v.Kind() == parquet.Int64 { + if ns, ok := convertTimeUnit(v.Int64(), pt, time.Nanosecond, roundUp); ok { + return iceberg.TimestampNsLiteral(iceberg.TimestampNano(ns)), true + } + } case iceberg.StringType: b := v.ByteArray() return iceberg.StringLiteral(string(b)), v.Kind() == parquet.ByteArray && utf8.Valid(b) case iceberg.BinaryType: return iceberg.BinaryLiteral(bytes.Clone(v.ByteArray())), v.Kind() == parquet.ByteArray + case iceberg.FixedType: + if v.Kind() == parquet.FixedLenByteArray && len(v.ByteArray()) == t.Len() { + return iceberg.FixedLiteral(bytes.Clone(v.ByteArray())), true + } + case iceberg.UUIDType: + if v.Kind() == parquet.FixedLenByteArray && len(v.ByteArray()) == 16 { + var u uuid.UUID + copy(u[:], v.ByteArray()) + return iceberg.UUIDLiteral(u), true + } + case iceberg.DecimalType: + // Only the integer encodings (precision up to 18, what DuckDB writes for + // DECIMAL(18,s)): the statistics of a byte-array decimal are compared + // as bytes, which does not order negative two's-complement values. + if scale, ok := parquetDecimalScale(pt); !ok || scale != t.Scale() { + return nil, false + } + switch v.Kind() { + case parquet.Int32: + return iceberg.DecimalLiteral(iceberg.Decimal{Val: decimal128.FromI64(int64(v.Int32())), Scale: t.Scale()}), true + case parquet.Int64: + return iceberg.DecimalLiteral(iceberg.Decimal{Val: decimal128.FromI64(v.Int64()), Scale: t.Scale()}), true + } } return nil, false } +// convertTimeUnit rescales a time value from the Parquet column's unit to the +// unit the Iceberg type stores, rounding towards the safe side of a bound. +func convertTimeUnit(v int64, pt parquet.Type, target time.Duration, roundUp bool) (int64, bool) { + unit := timeUnit(pt) + if unit == 0 { + return 0, false + } + if unit == target { + return v, true + } + if unit > target { + factor := int64(unit / target) + if v > math.MaxInt64/factor || v < math.MinInt64/factor { + return 0, false + } + return v * factor, true + } + factor := int64(target / unit) + q := v / factor + if roundUp && v%factor != 0 && v > 0 { + q++ + } + if !roundUp && v%factor != 0 && v < 0 { + q-- + } + return q, true +} + func timeUnit(pt parquet.Type) time.Duration { if pt == nil || pt.LogicalType() == nil { return 0 } - if ts, ok := pt.LogicalType().Value.(*format.TimestampType); ok && ts.Unit.Value != nil { - return ts.Unit.Value.Duration() + switch lt := pt.LogicalType().Value.(type) { + case *format.TimestampType: + if lt.Unit.Value != nil { + return lt.Unit.Value.Duration() + } + case *format.TimeType: + if lt.Unit.Value != nil { + return lt.Unit.Value.Duration() + } } return 0 } +func parquetDecimalScale(pt parquet.Type) (int, bool) { + if pt == nil || pt.LogicalType() == nil { + return 0, false + } + if d, ok := pt.LogicalType().Value.(*format.DecimalType); ok { + return int(d.Scale), true + } + return 0, false +} + // truncateBounds applies truncate(16) to string and binary bounds. A prefix is // still a lower bound; a shortened upper bound is not, so a value longer than // the limit keeps no upper bound (the reference implementation increments the diff --git a/weed/worker/tasks/iceberg/compact_stats_types_test.go b/weed/worker/tasks/iceberg/compact_stats_types_test.go new file mode 100644 index 0000000000..122cddd030 --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_stats_types_test.go @@ -0,0 +1,162 @@ +package iceberg + +import ( + "bytes" + "strings" + "testing" + + "github.com/apache/arrow-go/v18/arrow/decimal128" + "github.com/apache/iceberg-go" + "github.com/google/uuid" + "github.com/parquet-go/parquet-go" +) + +// The bound converter covers the types engines write besides the plain +// numbers and strings: a compacted file whose DECIMAL, TIME, UUID or +// millisecond-timestamp column has no bounds can never be skipped on it. +func TestBoundLiteralTypes(t *testing.T) { + u := uuid.MustParse("0b6a4f7e-6c1d-4c7a-9a52-2c51c0d6a9f1") + cases := []struct { + name string + v parquet.Value + pt parquet.Type + it iceberg.Type + roundUp bool + want iceberg.Literal // nil = no bound + }{ + {"decimal(18,2) as INT64", parquet.Int64Value(-12345), parquet.Decimal(2, 18, parquet.Int64Type).Type(), iceberg.DecimalTypeOf(18, 2), false, + iceberg.DecimalLiteral(iceberg.Decimal{Val: decimal128.FromI64(-12345), Scale: 2})}, + {"decimal(9,2) as INT32", parquet.Int32Value(705), parquet.Decimal(2, 9, parquet.Int32Type).Type(), iceberg.DecimalTypeOf(9, 2), true, + iceberg.DecimalLiteral(iceberg.Decimal{Val: decimal128.FromI64(705), Scale: 2})}, + {"decimal at another scale: none", parquet.Int64Value(1), parquet.Decimal(3, 18, parquet.Int64Type).Type(), iceberg.DecimalTypeOf(18, 2), false, nil}, + {"decimal(38,2) as bytes: none", parquet.FixedLenByteArrayValue(make([]byte, 16)), parquet.Decimal(2, 38, parquet.FixedLenByteArrayType(16)).Type(), iceberg.DecimalTypeOf(38, 2), false, nil}, + {"time (us)", parquet.Int64Value(3_600_000_000), parquet.Time(parquet.Microsecond).Type(), iceberg.PrimitiveTypes.Time, false, + iceberg.TimeLiteral(3_600_000_000)}, + {"time (ms) is INT32", parquet.Int32Value(3_600_000), parquet.Time(parquet.Millisecond).Type(), iceberg.PrimitiveTypes.Time, false, + iceberg.TimeLiteral(3_600_000_000)}, + {"timestamp ms, lower", parquet.Int64Value(1_700_000_000_123), parquet.Timestamp(parquet.Millisecond).Type(), iceberg.PrimitiveTypes.Timestamp, false, + iceberg.TimestampLiteral(1_700_000_000_123_000)}, + {"timestamp ns to us, upper rounds up", parquet.Int64Value(1_700_000_000_000_000_001), parquet.Timestamp(parquet.Nanosecond).Type(), iceberg.PrimitiveTypes.TimestampTz, true, + iceberg.TimestampLiteral(1_700_000_000_000_001)}, + {"timestamp ns to us, lower rounds down", parquet.Int64Value(1_700_000_000_000_000_001), parquet.Timestamp(parquet.Nanosecond).Type(), iceberg.PrimitiveTypes.TimestampTz, false, + iceberg.TimestampLiteral(1_700_000_000_000_000)}, + {"negative timestamp ns to us, lower rounds down", parquet.Int64Value(-1_500), parquet.Timestamp(parquet.Nanosecond).Type(), iceberg.PrimitiveTypes.TimestampTz, false, + iceberg.TimestampLiteral(-2)}, + {"negative timestamp ns to us, upper rounds up", parquet.Int64Value(-1_500), parquet.Timestamp(parquet.Nanosecond).Type(), iceberg.PrimitiveTypes.TimestampTz, true, + iceberg.TimestampLiteral(-1)}, + {"timestamptz ns us rescale", parquet.Int64Value(1_700_000_000_123), parquet.Timestamp(parquet.Microsecond).Type(), iceberg.TimestampTzNsType{}, false, + iceberg.TimestampNsLiteral(1_700_000_000_123_000)}, + {"timestamp ms to us overflows: none", parquet.Int64Value(1 << 62), parquet.Timestamp(parquet.Millisecond).Type(), iceberg.PrimitiveTypes.Timestamp, false, nil}, + {"uuid", parquet.FixedLenByteArrayValue(u[:]), parquet.UUID().Type(), iceberg.PrimitiveTypes.UUID, false, iceberg.UUIDLiteral(u)}, + {"fixed[4]", parquet.FixedLenByteArrayValue([]byte{1, 2, 3, 4}), parquet.FixedLenByteArrayType(4), iceberg.FixedTypeOf(4), false, + iceberg.FixedLiteral([]byte{1, 2, 3, 4})}, + {"fixed of another length: none", parquet.FixedLenByteArrayValue([]byte{1, 2, 3}), parquet.FixedLenByteArrayType(3), iceberg.FixedTypeOf(4), false, nil}, + {"double from FLOAT", parquet.FloatValue(1.5), parquet.FloatType, iceberg.PrimitiveTypes.Float64, false, iceberg.Float64Literal(1.5)}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, ok := boundLiteral(c.v, c.pt, c.it, c.roundUp) + if c.want == nil { + if ok { + t.Fatalf("want no bound, got %v", got) + } + return + } + if !ok || !got.Equals(c.want) { + t.Fatalf("got %v (ok %v), want %v", got, ok, c.want) + } + }) + } +} + +// A file written with a DECIMAL(18,2) stored as INT64, a millisecond +// timestamp, a nanosecond timestamp read as a microsecond timestamptz, a +// TIME, a UUID and a FIXED[4] column, in two row groups, gets lower and upper +// bounds for each of them. +func TestCollectColumnStatsBoundsMoreTypes(t *testing.T) { + icebergSchema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "price", Type: iceberg.DecimalTypeOf(18, 2), Required: true}, + iceberg.NestedField{ID: 2, Name: "ts_ms", Type: iceberg.PrimitiveTypes.Timestamp, Required: true}, + iceberg.NestedField{ID: 3, Name: "ts_ns", Type: iceberg.PrimitiveTypes.TimestampTz, Required: true}, + iceberg.NestedField{ID: 4, Name: "tod", Type: iceberg.PrimitiveTypes.Time, Required: true}, + iceberg.NestedField{ID: 5, Name: "id", Type: iceberg.PrimitiveTypes.UUID, Required: true}, + iceberg.NestedField{ID: 6, Name: "code", Type: iceberg.FixedTypeOf(4), Required: true}, + ) + schema := parquet.NewSchema("table", parquet.Group{ + "price": parquet.FieldID(parquet.Decimal(2, 18, parquet.Int64Type), 1), + "ts_ms": parquet.FieldID(parquet.Timestamp(parquet.Millisecond), 2), + "ts_ns": parquet.FieldID(parquet.Timestamp(parquet.Nanosecond), 3), + "tod": parquet.FieldID(parquet.Time(parquet.Microsecond), 4), + "id": parquet.FieldID(parquet.UUID(), 5), + "code": parquet.FieldID(parquet.Leaf(parquet.FixedLenByteArrayType(4)), 6), + }) + col := map[string]int{} + for i, p := range schema.Columns() { + col[strings.Join(p, ".")] = i + } + u1 := uuid.MustParse("00000000-0000-0000-0000-000000000001") + u2 := uuid.MustParse("ffffffff-0000-0000-0000-000000000002") + type row struct { + price, tsMS, tsNS, tod int64 + id uuid.UUID + code []byte + } + groups := [][]row{ + {{-12345, 1_700_000_000_123, -1_500, 3_600_000_000, u2, []byte{9, 9, 9, 9}}}, + {{67890, 1_700_000_001_000, 2_500, 7_200_000_000, u1, []byte{1, 2, 3, 4}}}, + } + var buf bytes.Buffer + w := parquet.NewGenericWriter[any](&buf, schema) + for _, g := range groups { + for _, r := range g { + pr := make(parquet.Row, len(col)) + pr[col["price"]] = parquet.Int64Value(r.price).Level(0, 0, col["price"]) + pr[col["ts_ms"]] = parquet.Int64Value(r.tsMS).Level(0, 0, col["ts_ms"]) + pr[col["ts_ns"]] = parquet.Int64Value(r.tsNS).Level(0, 0, col["ts_ns"]) + pr[col["tod"]] = parquet.Int64Value(r.tod).Level(0, 0, col["tod"]) + pr[col["id"]] = parquet.FixedLenByteArrayValue(r.id[:]).Level(0, 0, col["id"]) + pr[col["code"]] = parquet.FixedLenByteArrayValue(r.code).Level(0, 0, col["code"]) + if _, err := w.WriteRows([]parquet.Row{pr}); err != nil { + t.Fatalf("write row: %v", err) + } + } + if err := w.Flush(); err != nil { + t.Fatalf("flush: %v", err) + } + } + if err := w.Close(); err != nil { + t.Fatalf("close writer: %v", err) + } + + stats, err := collectColumnStats(buf.Bytes(), icebergSchema) + if err != nil { + t.Fatal(err) + } + dec := func(v int64) iceberg.Literal { + return iceberg.DecimalLiteral(iceberg.Decimal{Val: decimal128.FromI64(v), Scale: 2}) + } + want := [][2]iceberg.Literal{ // by field id, from 1 + {dec(-12345), dec(67890)}, + {iceberg.TimestampLiteral(1_700_000_000_123_000), iceberg.TimestampLiteral(1_700_000_001_000_000)}, + {iceberg.TimestampLiteral(-2), iceberg.TimestampLiteral(3)}, // rounded outwards + {iceberg.TimeLiteral(3_600_000_000), iceberg.TimeLiteral(7_200_000_000)}, + {iceberg.UUIDLiteral(u1), iceberg.UUIDLiteral(u2)}, + {iceberg.FixedLiteral([]byte{1, 2, 3, 4}), iceberg.FixedLiteral([]byte{9, 9, 9, 9})}, + } + for i, w := range want { + id := i + 1 + name := icebergSchema.Field(i).Name + for j, got := range [][]byte{stats.lower[id], stats.upper[id]} { + kind := [2]string{"lower", "upper"}[j] + exp, err := w[j].MarshalBinary() + if err != nil { + t.Fatal(err) + } + if got == nil { + t.Errorf("%s: no %s bound", name, kind) + } else if !bytes.Equal(got, exp) { + t.Errorf("%s: %s bound %x, want %x", name, kind, got, exp) + } + } + } +}