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 <chrislusf@users.noreply.github.com>

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

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

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