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)