diff --git a/weed/admin/dash/iceberg_data_preview.go b/weed/admin/dash/iceberg_data_preview.go index 872140097..adb7a9e3b 100644 --- a/weed/admin/dash/iceberg_data_preview.go +++ b/weed/admin/dash/iceberg_data_preview.go @@ -213,7 +213,7 @@ func (s *AdminServer) listIcebergDataFiles(ctx context.Context, bucketName, name if err != nil { return nil, false, nil, fmt.Errorf("read manifest list %s: %w", manifestListLocation, err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(listBytes)) + manifests, err := s3tables.ReadManifestList(listBytes) if err != nil { return nil, false, nil, fmt.Errorf("parse manifest list: %w", err) } diff --git a/weed/s3api/s3tables/iceberg_manifest_list.go b/weed/s3api/s3tables/iceberg_manifest_list.go new file mode 100644 index 000000000..9d5f56a88 --- /dev/null +++ b/weed/s3api/s3tables/iceberg_manifest_list.go @@ -0,0 +1,168 @@ +package s3tables + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "strconv" + + "github.com/apache/iceberg-go" +) + +// avroFileMagic prefixes every Avro object container file. +var avroFileMagic = []byte{'O', 'b', 'j', 1} + +// formatVersionKey is the Avro header entry naming the Iceberg format version. +const formatVersionKey = "format-version" + +// ReadManifestList parses an Iceberg manifest list, tolerating writers that +// leave the format version out of the Avro header. +// +// The Iceberg spec pins the header metadata of manifest *files* but says +// nothing about manifest *lists*, so writers disagree: Java and PyIceberg +// record "format-version", DuckDB writes no header metadata at all. iceberg-go +// reads a missing entry as v1, which makes every v2 manifest reachable from +// such a list fail with "manifest file's 'format-version' metadata indicates +// version 2, but entry from manifest list indicates version 1", and silently +// decodes delete manifests as data manifests because v1 has no "content" +// field. Both leave DuckDB-written tables unmaintainable. +// +// When the entry is missing we derive the version from the record schema the +// writer embedded — the fields it carries say which version wrote it — and add +// it to the header before handing the bytes to iceberg-go. +func ReadManifestList(manifestList []byte) ([]iceberg.ManifestFile, error) { + if patched, ok := withFormatVersion(manifestList); ok { + manifestList = patched + } + return iceberg.ReadManifestList(bytes.NewReader(manifestList)) +} + +// withFormatVersion returns manifestList with a "format-version" header entry +// spliced in. It reports false when the header already carries one, or when it +// cannot be parsed — the untouched bytes then go to iceberg-go, which reports +// the underlying problem. +func withFormatVersion(manifestList []byte) ([]byte, bool) { + metadata, terminator, err := readAvroFileMetadata(manifestList) + if err != nil { + return nil, false + } + if _, ok := metadata[formatVersionKey]; ok { + return nil, false + } + + // The metadata map is a sequence of blocks closed by a zero count, so a + // new single-entry block can be spliced in ahead of that closing count + // without re-encoding what the writer already wrote. + block := appendAvroLong(nil, 1) + block = appendAvroBytes(block, []byte(formatVersionKey)) + block = appendAvroBytes(block, []byte(strconv.Itoa(manifestListFormatVersion(metadata["avro.schema"])))) + + patched := make([]byte, 0, len(manifestList)+len(block)) + patched = append(patched, manifestList[:terminator]...) + patched = append(patched, block...) + return append(patched, manifestList[terminator:]...), true +} + +// manifestListFormatVersion infers the format version a manifest list was +// written at from the fields of its manifest_file record: v2 added "content", +// "sequence_number" and "min_sequence_number", v3 added "first_row_id". +func manifestListFormatVersion(schema []byte) int { + var record struct { + Fields []struct { + Name string `json:"name"` + } `json:"fields"` + } + if err := json.Unmarshal(schema, &record); err != nil { + return 1 + } + version := 1 + for _, field := range record.Fields { + switch field.Name { + case "content", "sequence_number", "min_sequence_number": + if version < 2 { + version = 2 + } + case "first_row_id": + version = 3 + } + } + return version +} + +// readAvroFileMetadata decodes the header metadata map of an Avro object +// container file, returning the map alongside the offset of the zero block +// count that closes it. +func readAvroFileMetadata(data []byte) (map[string][]byte, int, error) { + if !bytes.HasPrefix(data, avroFileMagic) { + return nil, 0, errors.New("not an avro object container file") + } + metadata := make(map[string][]byte) + pos := len(avroFileMagic) + for { + blockStart := pos + count, next, err := readAvroLong(data, pos) + if err != nil { + return nil, 0, err + } + pos = next + if count == 0 { + return metadata, blockStart, nil + } + if count < 0 { + // A negative count is followed by the block size in bytes. + count = -count + if _, pos, err = readAvroLong(data, pos); err != nil { + return nil, 0, err + } + } + for i := int64(0); i < count; i++ { + key, next, err := readAvroBytes(data, pos) + if err != nil { + return nil, 0, err + } + value, next, err := readAvroBytes(data, next) + if err != nil { + return nil, 0, err + } + metadata[string(key)] = value + pos = next + } + } +} + +// readAvroLong decodes the zig-zag varint at pos, returning it with the offset +// just past it. +func readAvroLong(data []byte, pos int) (int64, int, error) { + if pos < 0 || pos >= len(data) { + return 0, 0, fmt.Errorf("avro long at offset %d is out of range", pos) + } + value, n := binary.Varint(data[pos:]) + if n <= 0 { + return 0, 0, fmt.Errorf("avro long at offset %d is truncated", pos) + } + return value, pos + n, nil +} + +// readAvroBytes decodes the length-prefixed byte sequence at pos, returning it +// with the offset just past it. The result aliases data. +func readAvroBytes(data []byte, pos int) ([]byte, int, error) { + length, pos, err := readAvroLong(data, pos) + if err != nil { + return nil, 0, err + } + if length < 0 || int64(len(data)-pos) < length { + return nil, 0, fmt.Errorf("avro bytes at offset %d are truncated", pos) + } + return data[pos : pos+int(length)], pos + int(length), nil +} + +func appendAvroLong(dst []byte, value int64) []byte { + var scratch [binary.MaxVarintLen64]byte + return append(dst, scratch[:binary.PutVarint(scratch[:], value)]...) +} + +func appendAvroBytes(dst []byte, value []byte) []byte { + return append(appendAvroLong(dst, int64(len(value))), value...) +} diff --git a/weed/s3api/s3tables/iceberg_manifest_list_test.go b/weed/s3api/s3tables/iceberg_manifest_list_test.go new file mode 100644 index 000000000..820cc4d93 --- /dev/null +++ b/weed/s3api/s3tables/iceberg_manifest_list_test.go @@ -0,0 +1,173 @@ +package s3tables + +import ( + "bytes" + "testing" + + "github.com/apache/iceberg-go" +) + +// writeManifestList builds a manifest list holding one data and one delete +// manifest, the shape DuckDB produces for a v2 table that has been updated. +func writeManifestList(t *testing.T, version int) []byte { + t.Helper() + + dataManifest := iceberg.NewManifestFile(version, "s3://bucket/ns/tbl/metadata/data-m0.avro", 1024, 0, 7). + SequenceNum(3, 3). + Content(iceberg.ManifestContentData). + AddedFiles(1). + AddedRows(10). + Build() + files := []iceberg.ManifestFile{dataManifest} + if version > 1 { + // Delete manifests only exist from v2 onwards. + files = append(files, iceberg.NewManifestFile(version, "s3://bucket/ns/tbl/metadata/deletes-m0.avro", 512, 0, 7). + SequenceNum(3, 3). + Content(iceberg.ManifestContentDeletes). + AddedFiles(1). + AddedRows(1). + Build()) + } + + var buf bytes.Buffer + seqNum := int64(3) + if err := iceberg.WriteManifestList(version, &buf, 7, nil, &seqNum, 0, files); err != nil { + t.Fatalf("write v%d manifest list: %v", version, err) + } + return buf.Bytes() +} + +// stripFormatVersion rewrites an Avro header without its "format-version" +// entry, reproducing what DuckDB writes: a manifest list carrying no Iceberg +// header metadata at all. +func stripFormatVersion(t *testing.T, data []byte) []byte { + t.Helper() + + metadata, terminator, err := readAvroFileMetadata(data) + if err != nil { + t.Fatalf("read avro header: %v", err) + } + if _, ok := metadata[formatVersionKey]; !ok { + t.Fatal("fixture has no format-version to strip") + } + + var entries []byte + count := 0 + for key, value := range metadata { + if key == formatVersionKey { + continue + } + entries = appendAvroBytes(appendAvroBytes(entries, []byte(key)), value) + count++ + } + + stripped := append([]byte{}, avroFileMagic...) + stripped = appendAvroLong(stripped, int64(count)) + stripped = append(stripped, entries...) + // Everything from the closing zero count on — sync marker and data + // blocks included — is unaffected by the header rewrite. + return append(stripped, data[terminator:]...) +} + +// DuckDB writes manifest lists with no Avro header metadata, so iceberg-go +// falls back to v1 and every v2 manifest below it fails to parse. +func TestReadManifestListWithoutFormatVersion(t *testing.T) { + stripped := stripFormatVersion(t, writeManifestList(t, 2)) + + // Baseline: iceberg-go alone reads the list as v1, which both mislabels + // the delete manifest and makes ReadManifest reject the v2 manifests. + unpatched, err := iceberg.ReadManifestList(bytes.NewReader(stripped)) + if err != nil { + t.Fatalf("iceberg.ReadManifestList: %v", err) + } + if got := unpatched[0].Version(); got != 1 { + t.Fatalf("expected iceberg-go to default to v1, got v%d", got) + } + + manifests, err := ReadManifestList(stripped) + if err != nil { + t.Fatalf("ReadManifestList: %v", err) + } + if len(manifests) != 2 { + t.Fatalf("expected 2 manifests, got %d", len(manifests)) + } + for _, mf := range manifests { + if mf.Version() != 2 { + t.Errorf("manifest %s: version = %d, want 2", mf.FilePath(), mf.Version()) + } + if mf.SequenceNum() != 3 || mf.MinSequenceNum() != 3 { + t.Errorf("manifest %s: sequence numbers = %d/%d, want 3/3", mf.FilePath(), mf.SequenceNum(), mf.MinSequenceNum()) + } + } + if got := manifests[0].ManifestContent(); got != iceberg.ManifestContentData { + t.Errorf("first manifest content = %s, want data", got) + } + if got := manifests[1].ManifestContent(); got != iceberg.ManifestContentDeletes { + t.Errorf("second manifest content = %s, want deletes", got) + } +} + +func TestReadManifestListKeepsWrittenFormatVersion(t *testing.T) { + for _, version := range []int{1, 2} { + manifests, err := ReadManifestList(writeManifestList(t, version)) + if err != nil { + t.Fatalf("v%d: ReadManifestList: %v", version, err) + } + if got := manifests[0].Version(); got != version { + t.Errorf("v%d: version = %d", version, got) + } + } +} + +// A v1 manifest list without the header entry must stay v1: its record schema +// has none of the fields v2 added. +func TestReadManifestListWithoutFormatVersionStaysV1(t *testing.T) { + manifests, err := ReadManifestList(stripFormatVersion(t, writeManifestList(t, 1))) + if err != nil { + t.Fatalf("ReadManifestList: %v", err) + } + if got := manifests[0].Version(); got != 1 { + t.Errorf("version = %d, want 1", got) + } +} + +// Unparseable input is handed to iceberg-go untouched so it keeps reporting +// the underlying problem rather than a header-rewriting one. +func TestReadManifestListRejectsNonAvro(t *testing.T) { + for name, data := range map[string][]byte{ + "empty": nil, + "not avro": []byte("this is not an avro file"), + "truncated magic": avroFileMagic[:3], + "truncated header": func() []byte { + list := writeManifestList(t, 2) + return list[:len(avroFileMagic)+2] + }(), + } { + if _, err := ReadManifestList(data); err == nil { + t.Errorf("%s: expected an error", name) + } + } +} + +func TestManifestListFormatVersion(t *testing.T) { + tests := []struct { + name string + schema string + want int + }{ + {"v1 fields", `{"fields":[{"name":"manifest_path"},{"name":"added_snapshot_id"}]}`, 1}, + {"v2 adds content", `{"fields":[{"name":"manifest_path"},{"name":"content"}]}`, 2}, + {"v2 adds sequence numbers", `{"fields":[{"name":"sequence_number"},{"name":"min_sequence_number"}]}`, 2}, + {"v3 adds first_row_id", `{"fields":[{"name":"content"},{"name":"first_row_id"}]}`, 3}, + {"v3 field ordering", `{"fields":[{"name":"first_row_id"},{"name":"content"}]}`, 3}, + {"unparseable schema", `not json`, 1}, + {"missing schema", ``, 1}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := manifestListFormatVersion([]byte(tc.schema)); got != tc.want { + t.Errorf("manifestListFormatVersion() = %d, want %d", got, tc.want) + } + }) + } +} diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index 7d6ef3dec..a5a17b470 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -69,7 +69,7 @@ func (h *Handler) compactDataFiles( if err != nil { return "", nil, fmt.Errorf("read manifest list: %w", err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(manifestListData)) + manifests, err := s3tables.ReadManifestList(manifestListData) if err != nil { return "", nil, fmt.Errorf("parse manifest list: %w", err) } diff --git a/weed/worker/tasks/iceberg/delete_rewrite.go b/weed/worker/tasks/iceberg/delete_rewrite.go index 9c4ef0721..5d8d9e9a2 100644 --- a/weed/worker/tasks/iceberg/delete_rewrite.go +++ b/weed/worker/tasks/iceberg/delete_rewrite.go @@ -284,7 +284,7 @@ func (h *Handler) rewritePositionDeleteFiles( if err != nil { return "", nil, fmt.Errorf("read manifest list: %w", err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(manifestListData)) + manifests, err := s3tables.ReadManifestList(manifestListData) if err != nil { return "", nil, fmt.Errorf("parse manifest list: %w", err) } diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index bbf704e4c..c06d36156 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -371,7 +371,7 @@ func loadCurrentManifests( if err != nil { return nil, fmt.Errorf("read manifest list: %w", err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(manifestListData)) + manifests, err := s3tables.ReadManifestList(manifestListData) if err != nil { return nil, fmt.Errorf("parse manifest list: %w", err) } diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index 6495bd716..dbe4994b7 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -871,6 +871,75 @@ func TestRemoveOrphansPreservesReferencedFiles(t *testing.T) { } } +// DuckDB writes manifest lists carrying no Iceberg header metadata, which +// iceberg-go reads as v1 and then rejects every v2 manifest listed in them, +// failing all three maintenance operations before they touch anything. +func TestMaintenanceOnManifestListWithoutFormatVersion(t *testing.T) { + fs, client := startFakeFiler(t) + + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "test", + TableName: "foo", + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: time.Now().UnixMilli(), ManifestList: "metadata/snap-1.avro"}, + }, + } + populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dropManifestListFormatVersion(t, fs, metaDir, "snap-1.avro") + + handler := NewHandler(nil) + config := Config{ + OrphanOlderThanHours: 0, // no safety window — every unreferenced file is an orphan + MaxCommitRetries: 3, + } + + result, _, err := handler.removeOrphans(context.Background(), client, setup.BucketName, setup.tablePath(), config) + if err != nil { + t.Fatalf("removeOrphans failed: %v", err) + } + if !strings.Contains(result, "removed 0 orphan") { + t.Errorf("expected every file to still be referenced, got %q", result) + } + if fs.getEntry(metaDir, "manifest-1.avro") == nil { + t.Error("manifest-1.avro (referenced manifest) should not have been deleted") + } +} + +// dropManifestListFormatVersion rewrites a manifest list's Avro header the way +// DuckDB writes it, without a "format-version" entry. The key is renamed rather +// than removed so every Avro length prefix stays valid — iceberg-go ignores +// header entries it does not recognise. +func dropManifestListFormatVersion(t *testing.T, fs *fakeFilerServer, dir, name string) { + t.Helper() + + entry := fs.getEntry(dir, name) + if entry == nil { + t.Fatalf("manifest list %s/%s not found", dir, name) + } + patched := bytes.Replace(entry.Content, []byte("format-version"), []byte("ignored-header"), 1) + if bytes.Equal(patched, entry.Content) { + t.Fatalf("manifest list %s carries no format-version to drop", name) + } + fs.putEntry(dir, name, &filer_pb.Entry{ + Name: entry.Name, + Attributes: entry.Attributes, + Content: patched, + }) + + // Confirm the fixture reproduces the report: with the entry gone, + // iceberg-go alone falls back to v1 and mislabels the v2 manifests. + manifests, err := iceberg.ReadManifestList(bytes.NewReader(patched)) + if err != nil { + t.Fatalf("read patched manifest list: %v", err) + } + if got := manifests[0].Version(); got != 1 { + t.Fatalf("patched manifest list still reports v%d", got) + } +} + func TestRewriteManifestsExecution(t *testing.T) { fs, client := startFakeFiler(t) diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index fc1733974..fcfa698f6 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -189,7 +189,7 @@ func collectSnapshotFiles( if err != nil { return nil, fmt.Errorf("read manifest list %s: %w", snap.ManifestList, err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(manifestListData)) + manifests, err := s3tables.ReadManifestList(manifestListData) if err != nil { return nil, fmt.Errorf("parse manifest list %s: %w", snap.ManifestList, err) } @@ -342,7 +342,7 @@ func (h *Handler) rewriteManifests( return "", nil, fmt.Errorf("read manifest list: %w", err) } - manifests, err := iceberg.ReadManifestList(bytes.NewReader(manifestListData)) + manifests, err := s3tables.ReadManifestList(manifestListData) if err != nil { return "", nil, fmt.Errorf("parse manifest list: %w", err) }