iceberg: read manifest lists that omit the Avro format version (#10475)

* s3tables: read Iceberg manifest lists that omit the Avro format version The Iceberg spec pins the Avro 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, so every v2 manifest listed in a DuckDB-written list is rejected with manifest file's 'format-version' metadata indicates version 2, but entry from manifest list indicates version 1 and, because v1 has no "content" field, delete manifests silently decode as data manifests. ReadManifestList derives the version from the record schema the writer embedded - v2 added "content" and the sequence numbers, v3 added "first_row_id" - and splices it into the header before handing the bytes to iceberg-go. Lists that already carry the entry, and input that is not a parseable Avro container, go through untouched. * iceberg: parse DuckDB-written manifest lists in maintenance and data preview Every manifest list read - the four maintenance operations and the admin table data preview - went straight to iceberg-go, so tables written by DuckDB failed detection and all of compact, remove_orphans, rewrite_manifests and expire_snapshots before they touched anything. Route them through s3tables.ReadManifestList, which recovers the format version the writer left out of the Avro header. This also restores the manifest content type on those tables: with the list read as v1 every delete manifest looked like a data manifest, which hid deletes from the compaction guard and made the preview report a table with position deletes as having none.
2026-08-16 12:16:36 +00:00 · 2026-07-28 16:42:17 -07:00
parent c8cafd8a1a
commit 4b0d09683a
8 changed files with 416 additions and 6 deletions
@@ -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)
 	}
@@ -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...)
+}
@@ -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)
+			}
+		})
+	}
+}
@@ -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)
 	}
@@ -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)
 	}
@@ -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)
 	}
@@ -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)

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