From 1d0b97f4c69a53ccf70a6f42cdb1a29c7c79f996 Mon Sep 17 00:00:00 2001 From: Dmitriy Pavlov Date: Thu, 3 Sep 2026 03:24:23 +0300 Subject: [PATCH] avro: field time.Time <> iceberg.date (#11091) * iceberg: normalize foreign day partitions during manifest rewrite * test: cover manifest rewrite with foreign day partitions * iceberg: restore every foreign partition value, not just day transforms iceberg-go takes a partition field's logical type from the last branch of its Avro union, so a writer that spells an optional partition [, null] rather than [null, ] leaves the value as whatever the Avro decoder produced. A day or date partition then arrives as a time.Time the manifest writer cannot encode, and a time partition is worse: time.Duration converts to int64 nanoseconds and silently records the wrong value. ReadManifest sits next to ReadManifestList, the other shim for what foreign writers put on the wire, and converts each partition value back to the Iceberg representation for its field type. Claude-Session: https://claude.ai/code/session_01FdQyRuWF9SuCnPn21iH9yR * iceberg: read manifests that carry partition values through the shim Compaction, delete rewrite and their detection passes read entries and write the same partition values back into new manifests, so they fail on a foreign day partition exactly as manifest rewrite does. Where filters see it too: literalMatchesActual falls through to fmt.Sprint, so a time.Time renders as a timestamp and never matches the day the user asked for. The two remaining readers, orphan collection and the admin preview, only look at file paths and stay on iceberg.ReadManifest. Claude-Session: https://claude.ai/code/session_01FdQyRuWF9SuCnPn21iH9yR * iceberg: convert partition values before the writer rebinds logical types Dimonyga checked the manifests of a live Doris table: every input spells the partition union null-first, with the date logical type present, so the union ordering is not what breaks the merge. The conversion is lazy. iceberg-go converts what the Avro decoder returned on the first Partition() call, using the logical types read from the manifest being parsed, and ManifestWriter.addEntry rebinds them to the manifest it is about to write before it makes that call. A day partition is where the two disagree -- iceberg-go's day transform reports an int32 result type, so the manifest it writes carries no date logical type at all -- and an entry nobody looked at in between converts against that and keeps its time.Time. That is why only rewrite_manifests failed: compaction and delete rewrite group entries by partitionKey(df.Partition()) first, which converts them, and a where filter does the same. Reading every entry's partition here converts them all while the manifest's own logical types are still in place. Claude-Session: https://claude.ai/code/session_01FdQyRuWF9SuCnPn21iH9yR --------- Co-authored-by: Chris Lu --- go.mod | 2 +- weed/s3api/s3tables/iceberg_manifest.go | 91 ++++++++ weed/s3api/s3tables/iceberg_manifest_test.go | 199 ++++++++++++++++++ .../s3tables/s3tablestest/foreign_manifest.go | 150 +++++++++++++ weed/worker/tasks/iceberg/compact.go | 12 +- weed/worker/tasks/iceberg/delete_rewrite.go | 13 +- weed/worker/tasks/iceberg/detection.go | 7 +- weed/worker/tasks/iceberg/exec_test.go | 6 +- .../tasks/iceberg/foreign_format_scan_test.go | 4 +- .../tasks/iceberg/foreign_format_test.go | 2 +- weed/worker/tasks/iceberg/handler_test.go | 29 +-- weed/worker/tasks/iceberg/operations.go | 20 +- weed/worker/tasks/iceberg/operations_test.go | 151 +++++++++++++ .../worker/tasks/iceberg/table_config_test.go | 2 +- 14 files changed, 645 insertions(+), 43 deletions(-) create mode 100644 weed/s3api/s3tables/iceberg_manifest.go create mode 100644 weed/s3api/s3tables/iceberg_manifest_test.go create mode 100644 weed/s3api/s3tables/s3tablestest/foreign_manifest.go create mode 100644 weed/worker/tasks/iceberg/operations_test.go diff --git a/go.mod b/go.mod index 944109e6b..f8a6bfae0 100644 --- a/go.mod +++ b/go.mod @@ -154,6 +154,7 @@ require ( github.com/tarantool/go-tarantool/v3 v3.0.1 github.com/testcontainers/testcontainers-go v0.43.0 github.com/tikv/client-go/v2 v2.0.7 + github.com/twmb/avro v1.7.2 github.com/xeipuuv/gojsonschema v1.2.0 github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.2 github.com/ydb-platform/ydb-go-sdk/v3 v3.151.1 @@ -269,7 +270,6 @@ require ( github.com/substrait-io/substrait v0.87.0 // indirect github.com/substrait-io/substrait-go/v8 v8.1.1 // indirect github.com/substrait-io/substrait-protobuf/go v0.85.0 // indirect - github.com/twmb/avro v1.7.2 // indirect github.com/twpayne/go-geom v1.6.1 // indirect github.com/twpayne/go-kml/v3 v3.2.1 // indirect github.com/tyler-smith/go-bip39 v1.1.0 // indirect diff --git a/weed/s3api/s3tables/iceberg_manifest.go b/weed/s3api/s3tables/iceberg_manifest.go new file mode 100644 index 000000000..ea7af4dee --- /dev/null +++ b/weed/s3api/s3tables/iceberg_manifest.go @@ -0,0 +1,91 @@ +package s3tables + +import ( + "bytes" + "time" + + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +const secondsPerDay = int64(24 * 60 * 60) + +// ReadManifest parses an Iceberg manifest and converts its partition values +// while the manifest's own schema still describes them. +// +// iceberg-go converts what the Avro decoder returned for a logical type - a +// time.Time for a date, a time.Duration for a time - to an Iceberg value +// lazily, on the first Partition() call, using the logical types it read from +// the manifest being parsed. ManifestWriter.addEntry rebinds those to the +// logical types of the manifest it is about to write before it makes that +// call, so an entry that is read and written again without anyone looking at +// its partition converts against the wrong schema. A day partition is where +// the two disagree: iceberg-go's day transform reports an int32 result type, +// so the manifest it writes has no date logical type for that field, nothing +// converts, and the time.Time reaches the encoder as "cannot use time.Time +// with Avro type int". +// +// What survives that is a value iceberg-go never recognized on read either: it +// takes a partition field's logical type from the last branch of its Avro +// union, so a writer that spells an optional partition [, null] rather +// than [null, ] hides it, and a time partition is then worse than a +// failed write - time.Duration converts to int64 nanoseconds and silently +// records the wrong value. normalizePartitionValue puts those back. +// +// specs maps partition spec ID to spec the way the table metadata records them. +func ReadManifest(m iceberg.ManifestFile, manifest []byte, discardDeleted bool, specs map[int]iceberg.PartitionSpec, schema *iceberg.Schema) ([]iceberg.ManifestEntry, error) { + entries, err := iceberg.ReadManifest(m, bytes.NewReader(manifest), discardDeleted) + if err != nil { + return nil, err + } + + var partitionFields []iceberg.NestedField + if spec, found := specs[int(m.PartitionSpecID())]; found && schema != nil { + if partitionType := spec.PartitionType(schema); partitionType != nil { + partitionFields = partitionType.FieldList + } + } + for _, entry := range entries { + partition := entry.DataFile().Partition() + for _, field := range partitionFields { + value, ok := partition[field.ID] + if !ok { + continue + } + if normalized, ok := normalizePartitionValue(value, field.Type); ok { + partition[field.ID] = normalized + } + } + } + return entries, nil +} + +// normalizePartitionValue converts one value the Avro decoder returned for a +// logical type to the Iceberg representation the manifest writer expects, +// mirroring what iceberg-go does itself for the unions it does recognize. +func normalizePartitionValue(value any, fieldType iceberg.Type) (any, bool) { + switch v := value.(type) { + case time.Time: + utc := v.UTC() + switch fieldType.(type) { + case iceberg.DateType, iceberg.Int32Type: + // A day transform reports an int32 result type and an identity + // transform on a date column reports date; both hold days. + midnight := time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC) + return iceberg.Date(midnight.Unix() / secondsPerDay), true + case iceberg.TimestampType, iceberg.TimestampTzType: + return iceberg.Timestamp(utc.UnixMicro()), true + case iceberg.TimestampNsType, iceberg.TimestampTzNsType: + return iceberg.TimestampNano(utc.UnixNano()), true + } + case time.Duration: + if _, ok := fieldType.(iceberg.TimeType); ok { + return iceberg.Time(v.Microseconds()), true + } + case [16]byte: + if _, ok := fieldType.(iceberg.UUIDType); ok { + return uuid.UUID(v), true + } + } + return nil, false +} diff --git a/weed/s3api/s3tables/iceberg_manifest_test.go b/weed/s3api/s3tables/iceberg_manifest_test.go new file mode 100644 index 000000000..594f24e96 --- /dev/null +++ b/weed/s3api/s3tables/iceberg_manifest_test.go @@ -0,0 +1,199 @@ +package s3tables + +import ( + "bytes" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +// A manifest whose partition union puts the value branch first hides the +// logical type from iceberg-go, which then hands back whatever the Avro decoder +// produced even once the partition is read. Rewriting such an entry fails when +// the writer builds its partition summaries, or encodes the wrong number for a +// time partition, so every logical type a partition field can carry has to come +// back as an Iceberg value. +func TestReadManifestNormalizesForeignPartitions(t *testing.T) { + day := time.Date(2026, time.October, 11, 0, 0, 0, 0, time.UTC) + + cases := []struct { + name string + sourceType iceberg.PrimitiveType + transform iceberg.Transform + logicalType string + // decoded is the value the foreign manifest carries, as the Avro + // decoder hands it back. + decoded any + want any + // nullFirst spells the union the way Java and iceberg-rust do, which + // iceberg-go converts on its own once the partition is read. + nullFirst bool + }{ + { + name: "day transform written null-first converts on read", + sourceType: iceberg.PrimitiveTypes.Timestamp, + transform: iceberg.DayTransform{}, + logicalType: "date", + decoded: day, + want: iceberg.Date(20737), + nullFirst: true, + }, + { + name: "day transform on a timestamp column", + sourceType: iceberg.PrimitiveTypes.Timestamp, + transform: iceberg.DayTransform{}, + logicalType: "date", + decoded: day, + want: iceberg.Date(20737), + }, + { + name: "identity transform on a date column", + sourceType: iceberg.PrimitiveTypes.Date, + transform: iceberg.IdentityTransform{}, + logicalType: "date", + decoded: day, + want: iceberg.Date(20737), + }, + { + name: "identity transform on a timestamp column", + sourceType: iceberg.PrimitiveTypes.Timestamp, + transform: iceberg.IdentityTransform{}, + logicalType: "timestamp-micros", + decoded: day.Add(3 * time.Hour), + want: iceberg.Timestamp(day.Add(3 * time.Hour).UnixMicro()), + }, + { + name: "identity transform on a time column", + sourceType: iceberg.PrimitiveTypes.Time, + transform: iceberg.IdentityTransform{}, + logicalType: "time-micros", + decoded: 3 * time.Hour, + want: iceberg.Time((3 * time.Hour).Microseconds()), + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "source", Type: c.sourceType, Required: true}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "part", Transform: c.transform, + }) + specs := map[int]iceberg.PartitionSpec{spec.ID(): spec} + + dfBuilder, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/file.parquet", + iceberg.ParquetFile, map[int]any{1000: c.want}, nil, nil, 1, 1) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapshotID := int64(1) + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, dfBuilder.Build()) + + foreignBytes, foreignManifest := s3tablestest.ForeignPartitionManifest(t, schema, spec, entry, + "metadata/foreign-manifest.avro", c.logicalType, c.decoded, !c.nullFirst) + + // iceberg-go reads the logical type off the union's last branch, so + // the ordering decides whether the entry arrives as an Iceberg value + // or as whatever the Avro decoder produced. + rawWant := any(c.decoded) + if c.nullFirst { + rawWant = c.want + } + raw, err := iceberg.ReadManifest(foreignManifest, bytes.NewReader(foreignBytes), true) + if err != nil { + t.Fatalf("read foreign manifest: %v", err) + } + if got := raw[0].DataFile().Partition()[1000]; got != rawWant { + t.Fatalf("unnormalized partition value = %#v (%T), want %#v (%T)", got, got, rawWant, rawWant) + } + + entries, err := ReadManifest(foreignManifest, foreignBytes, true, specs, schema) + if err != nil { + t.Fatalf("read foreign manifest: %v", err) + } + if got := entries[0].DataFile().Partition()[1000]; got != c.want { + t.Fatalf("normalized partition value = %#v (%T), want %#v (%T)", got, got, c.want, c.want) + } + + // The point of normalizing: the entry can be written out again. + var buf bytes.Buffer + if _, err := iceberg.WriteManifest("metadata/manifest.avro", &buf, 2, spec, schema, snapshotID, entries); err != nil { + t.Fatalf("write normalized manifest: %v", err) + } + }) + } +} + +// A manifest written under a spec the table metadata no longer records is +// returned as read rather than refused: the caller decides what to do with it. +func TestReadManifestUnknownSpec(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "source", Type: iceberg.PrimitiveTypes.Date, Required: true}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "part", Transform: iceberg.IdentityTransform{}, + }) + dfBuilder, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/file.parquet", + iceberg.ParquetFile, map[int]any{1000: iceberg.Date(20737)}, nil, nil, 1, 1) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapshotID := int64(1) + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, dfBuilder.Build()) + foreignBytes, foreignManifest := s3tablestest.ForeignPartitionManifest(t, schema, spec, entry, + "metadata/foreign-manifest.avro", "date", time.Date(2026, time.October, 11, 0, 0, 0, 0, time.UTC), true) + + entries, err := ReadManifest(foreignManifest, foreignBytes, true, nil, schema) + if err != nil { + t.Fatalf("read foreign manifest: %v", err) + } + if _, ok := entries[0].DataFile().Partition()[1000].(time.Time); !ok { + t.Fatalf("partition value = %T, want it returned as read", entries[0].DataFile().Partition()[1000]) + } +} + +// iceberg-go converts a partition value on the first Partition() call, with +// whatever logical types are installed then, and ManifestWriter.addEntry +// rebinds them to the manifest it is about to write before it makes that call. +// A day partition has no date logical type there, so an entry nobody looked at +// between reading and writing never converts at all -- the failure Doris and +// iceberg-rust tables hit, whose manifests spell the union null-first. +func TestReadManifestConvertsBeforeTheWriterRebindsLogicalTypes(t *testing.T) { + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "event_time", Type: iceberg.PrimitiveTypes.Timestamp, Required: true}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "event_time_day", Transform: iceberg.DayTransform{}, + }) + specs := map[int]iceberg.PartitionSpec{spec.ID(): spec} + dfBuilder, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/file.parquet", + iceberg.ParquetFile, map[int]any{1000: iceberg.Date(20737)}, nil, nil, 1, 1) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapshotID := int64(1) + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, dfBuilder.Build()) + foreignBytes, foreignManifest := s3tablestest.ForeignPartitionManifest(t, schema, spec, entry, + "metadata/foreign-manifest.avro", "date", time.Date(2026, time.October, 11, 0, 0, 0, 0, time.UTC), false) + + untouched, err := iceberg.ReadManifest(foreignManifest, bytes.NewReader(foreignBytes), true) + if err != nil { + t.Fatalf("read foreign manifest: %v", err) + } + var buf bytes.Buffer + if _, err := iceberg.WriteManifest("metadata/manifest.avro", &buf, 2, spec, schema, snapshotID, untouched); err == nil { + t.Fatal("writing an entry whose partition was never read should still hold a time.Time") + } + + entries, err := ReadManifest(foreignManifest, foreignBytes, true, specs, schema) + if err != nil { + t.Fatalf("read foreign manifest: %v", err) + } + buf.Reset() + if _, err := iceberg.WriteManifest("metadata/manifest.avro", &buf, 2, spec, schema, snapshotID, entries); err != nil { + t.Fatalf("write manifest read through the shim: %v", err) + } +} diff --git a/weed/s3api/s3tables/s3tablestest/foreign_manifest.go b/weed/s3api/s3tables/s3tablestest/foreign_manifest.go new file mode 100644 index 000000000..73a4ea112 --- /dev/null +++ b/weed/s3api/s3tables/s3tablestest/foreign_manifest.go @@ -0,0 +1,150 @@ +package s3tablestest + +import ( + "bytes" + "encoding/json" + "testing" + + "github.com/apache/iceberg-go" + "github.com/twmb/avro" + "github.com/twmb/avro/ocf" +) + +// ForeignPartitionManifest rewrites a manifest iceberg-go wrote so its +// partition field carries the Avro union a foreign writer emits: [, null] +// rather than [null, ]. Both are valid Avro, but iceberg-go reads a +// partition field's logical type from the union's last branch, so the ordering +// decides whether it hands back an Iceberg value or the rich Go value the Avro +// decoder produced. +// +// logicalType is stamped onto the value branch, since iceberg-go writes a day +// partition as a bare int, and decoded replaces the partition value in the +// record, standing in for what the foreign writer encoded. valueFirst orders +// the union [, null] rather than the [null, ] Java and iceberg-rust +// emit. +func ForeignPartitionManifest( + t *testing.T, + schema *iceberg.Schema, + spec iceberg.PartitionSpec, + entry iceberg.ManifestEntry, + manifestPath, logicalType string, + decoded any, + valueFirst bool, +) ([]byte, iceberg.ManifestFile) { + t.Helper() + + var original bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &original, 2, spec, schema, 1, []iceberg.ManifestEntry{entry}) + if err != nil { + t.Fatalf("write base manifest: %v", err) + } + + reader, err := ocf.NewReader(bytes.NewReader(original.Bytes())) + if err != nil { + t.Fatalf("open base manifest: %v", err) + } + metadata := reader.Metadata() + var record map[string]any + if err := reader.Decode(&record); err != nil { + t.Fatalf("decode base manifest: %v", err) + } + if err := reader.Close(); err != nil { + t.Fatalf("close base manifest: %v", err) + } + + fieldName := spec.Field(0).Name + + var schemaDoc map[string]any + if err := json.Unmarshal(metadata["avro.schema"], &schemaDoc); err != nil { + t.Fatalf("decode manifest schema: %v", err) + } + dataFile := findAvroField(t, schemaDoc, "data_file") + partition := findAvroField(t, dataFile, "partition") + partitionFields, ok := partition["fields"].([]any) + if !ok { + t.Fatalf("partition schema fields have type %T", partition["fields"]) + } + for _, rawField := range partitionFields { + field, ok := rawField.(map[string]any) + if !ok || field["name"] != fieldName { + continue + } + fieldType, ok := field["type"].([]any) + if !ok || len(fieldType) != 2 { + t.Fatalf("%s schema type = %#v, want nullable union", fieldName, field["type"]) + } + valueType, ok := fieldType[1].(map[string]any) + if !ok { + primitive, primitiveOK := fieldType[1].(string) + if !primitiveOK { + t.Fatalf("%s value branch = %T, want Avro type", fieldName, fieldType[1]) + } + valueType = map[string]any{"type": primitive} + } + valueType["logicalType"] = logicalType + if valueFirst { + field["type"] = []any{valueType, "null"} + } else { + field["type"] = []any{"null", valueType} + } + break + } + + dataFileRecord, ok := record["data_file"].(map[string]any) + if !ok { + t.Fatalf("decoded data_file = %T, want record", record["data_file"]) + } + partitionRecord, ok := dataFileRecord["partition"].(map[string]any) + if !ok { + t.Fatalf("decoded partition = %T, want record", dataFileRecord["partition"]) + } + partitionRecord[fieldName] = decoded + + foreignSchemaJSON, err := json.Marshal(schemaDoc) + if err != nil { + t.Fatalf("encode foreign manifest schema: %v", err) + } + foreignSchema, err := avro.Parse(string(foreignSchemaJSON)) + if err != nil { + t.Fatalf("parse foreign manifest schema: %v", err) + } + userMetadata := make(map[string][]byte) + for key, value := range metadata { + if key != "avro.schema" && key != "avro.codec" { + userMetadata[key] = value + } + } + var foreign bytes.Buffer + writer, err := ocf.NewWriter(&foreign, foreignSchema, ocf.WithSchema(string(foreignSchemaJSON)), ocf.WithMetadata(userMetadata)) + if err != nil { + t.Fatalf("create foreign manifest writer: %v", err) + } + if err := writer.Encode(record); err != nil { + t.Fatalf("encode foreign manifest: %v", err) + } + if err := writer.Close(); err != nil { + t.Fatalf("close foreign manifest writer: %v", err) + } + + return foreign.Bytes(), iceberg.NewManifestFile(2, manifest.FilePath(), int64(foreign.Len()), manifest.PartitionSpecID(), 1).AddedFiles(1).SequenceNum(1, 1).Build() +} + +func findAvroField(t *testing.T, record map[string]any, name string) map[string]any { + t.Helper() + fields, ok := record["fields"].([]any) + if !ok { + t.Fatalf("Avro record fields have type %T", record["fields"]) + } + for _, rawField := range fields { + field, ok := rawField.(map[string]any) + if ok && field["name"] == name { + fieldType, ok := field["type"].(map[string]any) + if ok { + return fieldType + } + t.Fatalf("Avro field %q type has type %T", name, field["type"]) + } + } + t.Fatalf("Avro field %q not found", name) + return nil +} diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index 2540e6eab..6200e315c 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -91,6 +91,9 @@ func (h *Handler) compactDataFiles( return "compaction skipped: delete manifests present and apply_deletes is disabled", nil, nil } + specsByID := specByID(meta) + schema := meta.CurrentSchema() + // Collect data file entries from data manifests var allEntries []iceberg.ManifestEntry for _, mf := range dataManifests { @@ -98,15 +101,13 @@ func (h *Handler) compactDataFiles( if err != nil { return "", nil, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err) } - entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema) if err != nil { return "", nil, fmt.Errorf("parse manifest %s: %w", mf.FilePath(), err) } allEntries = append(allEntries, entries...) } - specsByID := specByID(meta) - // Collect delete entries if we need to apply deletes var positionDeletes map[string][]int64 var eqDeleteGroups []equalityDeleteGroup @@ -117,7 +118,7 @@ func (h *Handler) compactDataFiles( if err != nil { return "", nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err) } - entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema) if err != nil { return "", nil, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), err) } @@ -157,7 +158,7 @@ func (h *Handler) compactDataFiles( } if len(eqDeleteEntries) > 0 { - eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, dataPath, eqDeleteEntries, meta.CurrentSchema()) + eqDeleteGroups, err = collectEqualityDeletes(ctx, filerClient, bucketName, dataPath, eqDeleteEntries, schema) if err != nil { return "", nil, fmt.Errorf("collect equality deletes: %w", err) } @@ -199,7 +200,6 @@ func (h *Handler) compactDataFiles( // Build a lookup from spec ID to PartitionSpec for per-bin manifest writing. specLookup := specsByID - schema := meta.CurrentSchema() version := meta.Version() snapshotID := currentSnap.SnapshotID diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go index 5d8d9e9a2..627835d40 100644 --- a/weed/worker/tasks/iceberg/delete_rewrite.go +++ b/weed/worker/tasks/iceberg/delete_rewrite.go @@ -46,7 +46,7 @@ func hasEligibleDeleteRewrite( meta table.Metadata, predicate *partitionPredicate, ) (bool, error) { - groups, _, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests) + groups, _, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests, specByID(meta), meta.CurrentSchema()) if err != nil { return false, err } @@ -76,6 +76,8 @@ func collectDeleteRewriteGroups( filerClient filer_pb.SeaweedFilerClient, bucketName, dataPath string, manifests []iceberg.ManifestFile, + specs map[int]iceberg.PartitionSpec, + schema *iceberg.Schema, ) (map[string]*deleteRewriteGroup, []iceberg.ManifestEntry, error) { groups := make(map[string]*deleteRewriteGroup) var allPositionEntries []iceberg.ManifestEntry @@ -89,7 +91,7 @@ func collectDeleteRewriteGroups( if err != nil { return nil, nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err) } - entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, err := s3tables.ReadManifest(mf, manifestData, true, specs, schema) if err != nil { return nil, nil, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), err) } @@ -289,6 +291,9 @@ func (h *Handler) rewritePositionDeleteFiles( return "", nil, fmt.Errorf("parse manifest list: %w", err) } + specs := specByID(meta) + schema := meta.CurrentSchema() + var dataManifests []iceberg.ManifestFile var allEqualityEntries []iceberg.ManifestEntry for _, mf := range manifests { @@ -300,7 +305,7 @@ func (h *Handler) rewritePositionDeleteFiles( if readErr != nil { return "", nil, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), readErr) } - entries, parseErr := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, parseErr := s3tables.ReadManifest(mf, manifestData, true, specs, schema) if parseErr != nil { return "", nil, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), parseErr) } @@ -312,7 +317,7 @@ func (h *Handler) rewritePositionDeleteFiles( } } - groupMap, allPositionEntries, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests) + groupMap, allPositionEntries, err := collectDeleteRewriteGroups(ctx, filerClient, bucketName, dataPath, manifests, specs, schema) if err != nil { return "", nil, err } diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index d352e204e..f32d3f642 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -1,7 +1,6 @@ package iceberg import ( - "bytes" "context" "errors" "fmt" @@ -438,13 +437,16 @@ func hasEligibleCompaction( return false, nil } + specsByID := specByID(meta) + schema := meta.CurrentSchema() + var allEntries []iceberg.ManifestEntry for _, mf := range dataManifests { manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath()) if err != nil { return false, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err) } - entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema) if err != nil { return false, fmt.Errorf("parse manifest %s: %w", mf.FilePath(), err) } @@ -453,7 +455,6 @@ func hasEligibleCompaction( candidateEntries := allEntries if predicate != nil { - specsByID := specByID(meta) candidateEntries = make([]iceberg.ManifestEntry, 0, len(allEntries)) for _, entry := range allEntries { spec, ok := specsByID[int(entry.DataFile().SpecID())] diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index 0e7e293c9..813873812 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -302,6 +302,10 @@ type tableSetup struct { // Age backdates the whole table so its snapshots can sit outside a // retention window. Age time.Duration + // Schema and Spec describe the table when the default unpartitioned test + // table does not fit. + Schema *iceberg.Schema + Spec *iceberg.PartitionSpec } func (ts tableSetup) tablePath() string { @@ -329,7 +333,7 @@ func (ts tableSetup) fileRef(elem ...string) string { func populateTable(t *testing.T, fs *fakeFilerServer, setup tableSetup) table.Metadata { t.Helper() - meta := buildTestMetadata(t, setup.Snapshots, setup.Refs, setup.Age, nil) + meta := buildTestMetadata(t, setup.Snapshots, setup.Refs, setup.Age, nil, setup.Schema, setup.Spec) fullMetadataJSON, err := json.Marshal(meta) if err != nil { t.Fatalf("marshal metadata: %v", err) diff --git a/weed/worker/tasks/iceberg/foreign_format_scan_test.go b/weed/worker/tasks/iceberg/foreign_format_scan_test.go index a9b103197..fb4cc27f1 100644 --- a/weed/worker/tasks/iceberg/foreign_format_scan_test.go +++ b/weed/worker/tasks/iceberg/foreign_format_scan_test.go @@ -29,7 +29,7 @@ func seedAdapterRegisteredLanceTable(t *testing.T, properties iceberg.Properties } filer.Put(s3tables.GetTableBucketPath(bucket), namespace, map[string][]byte{s3tables.ExtendedKeyMetadata: nsMeta}) - full, err := json.Marshal(buildTestMetadata(t, nil, nil, 0, properties)) + full, err := json.Marshal(buildTestMetadata(t, nil, nil, 0, properties, nil, nil)) if err != nil { t.Fatalf("marshal iceberg metadata: %v", err) } @@ -64,7 +64,7 @@ func seedAdapterRegisteredLanceTable(t *testing.T, properties iceberg.Properties func TestOrphanCleanupWouldDeleteAnAdapterRegisteredLanceDataset(t *testing.T) { filer, tablePath := seedAdapterRegisteredLanceTable(t, iceberg.Properties{tableTypeProperty: "lance"}) - meta := buildTestMetadata(t, nil, nil, 0, iceberg.Properties{tableTypeProperty: "lance"}) + meta := buildTestMetadata(t, nil, nil, 0, iceberg.Properties{tableTypeProperty: "lance"}, nil, nil) candidates, err := collectOrphanCandidates(context.Background(), filer.Client, "vectors", tablePath, meta, "v1.metadata.json", defaultOrphanOlderThanHours) if err != nil { diff --git a/weed/worker/tasks/iceberg/foreign_format_test.go b/weed/worker/tasks/iceberg/foreign_format_test.go index 65792d88b..bc337f390 100644 --- a/weed/worker/tasks/iceberg/foreign_format_test.go +++ b/weed/worker/tasks/iceberg/foreign_format_test.go @@ -48,7 +48,7 @@ func TestIsIcebergTableEntry(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { - meta := buildTestMetadata(t, nil, nil, 0, c.properties) + meta := buildTestMetadata(t, nil, nil, 0, c.properties, nil, nil) if got := isIcebergTableEntry(c.extended, meta); got != c.want { t.Fatalf("isIcebergTableEntry() = %v, want %v", got, c.want) } diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go index 598c65296..a2e68ddc3 100644 --- a/weed/worker/tasks/iceberg/handler_test.go +++ b/weed/worker/tasks/iceberg/handler_test.go @@ -123,7 +123,7 @@ func TestNeedsMaintenanceNoSnapshots(t *testing.T) { MaxSnapshotsToKeep: 2, } - meta := buildTestMetadata(t, nil, nil, 0, nil) + meta := buildTestMetadata(t, nil, nil, 0, nil, nil, nil) if needsMaintenance(meta, config) { t.Error("expected no maintenance for table with no snapshots") } @@ -141,7 +141,7 @@ func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) { {SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"}, {SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"}, } - meta := buildTestMetadata(t, snapshots, nil, 48*time.Hour, nil) + meta := buildTestMetadata(t, snapshots, nil, 48*time.Hour, nil, nil, nil) if !needsMaintenance(meta, config) { t.Error("expected maintenance for table exceeding max snapshots") } @@ -161,7 +161,7 @@ func TestNeedsMaintenanceExceedsMaxSnapshotsWithinRetention(t *testing.T) { {SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"}, {SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"}, } - if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil), config) { + if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil), config) { t.Error("expected no maintenance while every snapshot is inside the retention window") } } @@ -181,10 +181,10 @@ func TestNeedsMaintenanceSkipsRefPinnedSnapshots(t *testing.T) { refs := map[string]table.SnapshotRef{ "release": {SnapshotID: 1, SnapshotRefType: table.TagRef}, } - if needsMaintenance(buildTestMetadata(t, snapshots, refs, 0, nil), config) { + if needsMaintenance(buildTestMetadata(t, snapshots, refs, 0, nil, nil, nil), config) { t.Error("expected no maintenance when the only old snapshot is tagged") } - if !needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil), config) { + if !needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil), config) { t.Error("expected maintenance for the same table without the tag") } } @@ -199,7 +199,7 @@ func TestNeedsMaintenanceWithinLimits(t *testing.T) { snapshots := []table.Snapshot{ {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, } - meta := buildTestMetadata(t, snapshots, nil, 0, nil) + meta := buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil) if needsMaintenance(meta, config) { t.Error("expected no maintenance for table within limits") } @@ -217,7 +217,7 @@ func TestNeedsMaintenanceOldSnapshot(t *testing.T) { {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, {SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"}, } - meta := buildTestMetadata(t, snapshots, nil, 0, nil) + meta := buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil) if !needsMaintenance(meta, config) { t.Error("expected maintenance for table with expired snapshot") } @@ -235,7 +235,7 @@ func TestNeedsMaintenanceSingleSnapshot(t *testing.T) { snapshots := []table.Snapshot{ {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, } - if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil), config) { + if needsMaintenance(buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil), config) { t.Error("expected no maintenance for a table with only the current snapshot") } } @@ -280,7 +280,7 @@ func TestBuildMaintenanceProposal(t *testing.T) { {SnapshotID: 1, TimestampMs: now}, {SnapshotID: 2, TimestampMs: now + 1}, } - meta := buildTestMetadata(t, snapshots, nil, 0, nil) + meta := buildTestMetadata(t, snapshots, nil, 0, nil, nil, nil) info := tableInfo{ BucketName: "my-bucket", @@ -1549,11 +1549,16 @@ func TestExecuteNilRequest(t *testing.T) { // are genuinely past a retention window - iceberg-go refuses to add a snapshot // stamped more than a minute before the metadata's last-updated time, so the // shift has to happen after the build. -func buildTestMetadata(t *testing.T, snapshots []table.Snapshot, refs map[string]table.SnapshotRef, age time.Duration, properties iceberg.Properties) table.Metadata { +func buildTestMetadata(t *testing.T, snapshots []table.Snapshot, refs map[string]table.SnapshotRef, age time.Duration, properties iceberg.Properties, schema *iceberg.Schema, spec *iceberg.PartitionSpec) table.Metadata { t.Helper() - schema := newTestSchema() - meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "s3://test-bucket/test-table", properties) + if schema == nil { + schema = newTestSchema() + } + if spec == nil { + spec = iceberg.UnpartitionedSpec + } + meta, err := table.NewMetadata(schema, spec, table.UnsortedSortOrder, "s3://test-bucket/test-table", properties) if err != nil { t.Fatalf("failed to create test metadata: %v", err) } diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index d4355e219..eb476b2d8 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -442,6 +442,7 @@ func (h *Handler) rewriteManifests( // Build a lookup from spec ID to PartitionSpec specByID := specByID(meta) + schema := meta.CurrentSchema() var carriedDataManifests []iceberg.ManifestFile var manifestsRewritten int64 @@ -450,16 +451,17 @@ func (h *Handler) rewriteManifests( if err != nil { return "", nil, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err) } - entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + entries, err := s3tables.ReadManifest(mf, manifestData, true, specByID, schema) if err != nil { return "", nil, fmt.Errorf("parse manifest %s: %w", mf.FilePath(), err) } + sid := mf.PartitionSpecID() + spec, found := specByID[int(sid)] + if !found { + return "", nil, fmt.Errorf("partition spec %d not found in table metadata", sid) + } if predicate != nil { - spec, found := specByID[int(mf.PartitionSpecID())] - if !found { - return "", nil, fmt.Errorf("partition spec %d not found in table metadata", mf.PartitionSpecID()) - } allMatch := len(entries) > 0 for _, entry := range entries { match, err := predicate.Matches(spec, entry.DataFile().Partition()) @@ -477,14 +479,9 @@ func (h *Handler) rewriteManifests( } } - sid := mf.PartitionSpecID() se, ok := specMap[sid] if !ok { - ps, found := specByID[int(sid)] - if !found { - return "", nil, fmt.Errorf("partition spec %d not found in table metadata", sid) - } - se = &specEntries{specID: sid, spec: ps} + se = &specEntries{specID: sid, spec: spec} specMap[sid] = se } se.entries = append(se.entries, entries...) @@ -499,7 +496,6 @@ func (h *Handler) rewriteManifests( return "no data entries to rewrite", nil, nil } - schema := meta.CurrentSchema() version := meta.Version() snapshotID := currentSnap.SnapshotID newSnapshotID := time.Now().UnixMilli() diff --git a/weed/worker/tasks/iceberg/operations_test.go b/weed/worker/tasks/iceberg/operations_test.go new file mode 100644 index 000000000..972b16230 --- /dev/null +++ b/weed/worker/tasks/iceberg/operations_test.go @@ -0,0 +1,151 @@ +package iceberg + +import ( + "bytes" + "context" + "fmt" + "path" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +// Merging manifests a foreign producer wrote used to fail for good on a +// day-partitioned table, whichever way round the producer spelled the +// partition union: nothing between reading an entry and writing it looks at +// its partition, so the time.Time the Avro decoder produced is still there +// when the manifest writer, which has no date logical type for a day +// partition, tries to encode it. +func TestRewriteManifestsNormalizesForeignDayPartitions(t *testing.T) { + for _, valueFirst := range []bool{true, false} { + name := "null-first partition union" + if valueFirst { + name = "value-first partition union" + } + t.Run(name, func(t *testing.T) { + rewriteForeignDayPartitions(t, valueFirst) + }) + } +} + +func rewriteForeignDayPartitions(t *testing.T, valueFirst bool) { + t.Helper() + fs, client := startFakeFiler(t) + + schema := iceberg.NewSchema(0, + iceberg.NestedField{ID: 1, Name: "event_time", Type: iceberg.PrimitiveTypes.Timestamp, Required: true}, + ) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "event_time_day", Transform: iceberg.DayTransform{}, + }) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "analytics", + TableName: "events", + Schema: schema, + Spec: &spec, + Snapshots: []table.Snapshot{{ + SnapshotID: 1, + TimestampMs: time.Now().UnixMilli(), + ManifestList: "metadata/snap-1.avro", + }}, + } + meta := populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + wantDays := make(map[string]iceberg.Date) + var manifests []iceberg.ManifestFile + for i := 0; i < 3; i++ { + day := iceberg.Date(20737 + i) + filePath := setup.fileRef("data", fmt.Sprintf("foreign-%d.parquet", i)) + dfBuilder, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, filePath, iceberg.ParquetFile, + map[int]any{1000: day}, nil, nil, 1, 1) + if err != nil { + t.Fatalf("build data file %d: %v", i, err) + } + snapshotID := int64(1) + entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, nil, nil, dfBuilder.Build()) + + manifestName := fmt.Sprintf("foreign-manifest-%d.avro", i) + foreignBytes, manifest := s3tablestest.ForeignPartitionManifest(t, schema, spec, entry, + setup.fileRef("metadata", manifestName), "date", time.Unix(int64(day)*24*60*60, 0).UTC(), valueFirst) + fs.putEntry(metaDir, manifestName, &filer_pb.Entry{ + Name: manifestName, Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()}, Content: foreignBytes, + }) + manifests = append(manifests, manifest) + wantDays[filePath] = day + } + + var manifestList bytes.Buffer + seqNum := int64(1) + if err := iceberg.WriteManifestList(meta.Version(), &manifestList, 1, nil, &seqNum, 0, manifests); err != nil { + t.Fatalf("write manifest list: %v", err) + } + fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{ + Name: "snap-1.avro", Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix()}, Content: manifestList.Bytes(), + }) + + result, _, err := NewHandler(nil).rewriteManifests(context.Background(), client, setup.BucketName, setup.tablePath(), Config{ + MinManifestsToRewrite: 3, + MaxCommitRetries: 3, + }) + if err != nil { + t.Fatalf("rewriteManifests failed for foreign day partitions: %v", err) + } + if result != "rewrote 3 manifests into 1 (3 entries)" { + t.Fatalf("rewriteManifests result = %q, want 3 manifests merged", result) + } + + // The merged manifest has to carry the days the foreign manifests held. + // iceberg-go writes a day partition as a bare Avro int, without the date + // logical type, so it reads back as int32 rather than iceberg.Date. + got := make(map[string]any) + for _, mf := range currentManifests(t, client, setup) { + manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), mf.FilePath()) + if err != nil { + t.Fatalf("load merged manifest %s: %v", mf.FilePath(), err) + } + entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + if err != nil { + t.Fatalf("parse merged manifest %s: %v", mf.FilePath(), err) + } + for _, entry := range entries { + got[entry.DataFile().FilePath()] = entry.DataFile().Partition()[1000] + } + } + if len(got) != len(wantDays) { + t.Fatalf("merged manifests hold %d entries, want %d", len(got), len(wantDays)) + } + for filePath, day := range wantDays { + if got[filePath] != int32(day) { + t.Errorf("%s partition = %#v (%T), want %d", filePath, got[filePath], got[filePath], day) + } + } +} + +// currentManifests reads the manifests of the table's current snapshot. +func currentManifests(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) []iceberg.ManifestFile { + t.Helper() + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("reload metadata: %v", err) + } + snapshot := state.Metadata.CurrentSnapshot() + if snapshot == nil { + t.Fatal("table has no current snapshot") + } + manifestListData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, setup.tablePath(), snapshot.ManifestList) + if err != nil { + t.Fatalf("load manifest list: %v", err) + } + manifests, err := s3tables.ReadManifestList(manifestListData) + if err != nil { + t.Fatalf("parse manifest list: %v", err) + } + return manifests +} diff --git a/weed/worker/tasks/iceberg/table_config_test.go b/weed/worker/tasks/iceberg/table_config_test.go index a96e726be..42f6f87a3 100644 --- a/weed/worker/tasks/iceberg/table_config_test.go +++ b/weed/worker/tasks/iceberg/table_config_test.go @@ -388,7 +388,7 @@ func TestResolveCompactionRewritePlanAuto(t *testing.T) { cfg := baseTestConfig() cfg.RewriteStrategy = rewriteStrategyAuto - unsorted := buildTestMetadata(t, nil, nil, 0, nil) + unsorted := buildTestMetadata(t, nil, nil, 0, nil, nil, nil) plan, err := resolveCompactionRewritePlan(cfg, unsorted) if err != nil { t.Fatalf("auto must not fail on an unsorted table: %v", err)