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 [<type>, null]
rather than [null, <type>] 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 <chris.lu@gmail.com>

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 [<type>, null] rather
+// than [null, <type>] 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
+}

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

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: [<type>, null]
+// rather than [null, <type>]. 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 [<type>, null] rather than the [null, <type>] 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
+}

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
 

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
 	}

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())]

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)

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 {

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

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

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()

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

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)