iceberg maintenance: an unreadable parquet file fails the job, not the process (#12188)

* iceberg maintenance: an unreadable parquet file fails the job, not the process

parquet.NewReader panics on any file it cannot open (wrong magic, a
truncated footer, zero bytes), and parquet-go's row decoder panics on a
corrupt page under a valid footer. Maintenance runs in the worker's
goroutines with no recover, so one bad delete or data file in any table
ended the whole process.

- Every parquet open in compaction goes through newParquetReader, which
  opens with parquet.OpenFile and returns the error.
- A panic while decoding a delete file or a data file's rows is recovered
  into an error naming the file (recoverRead/guardRead).
- As a backstop, detection recovers per table and Execute per operation:
  the table is skipped or the operation fails, and the worker keeps
  running.

A failed delete file still fails the table's compaction (no commit), as a
read error did before; a failed data file still skips only its bin.

* iceberg maintenance: test a corrupt page inside a merge fails it naming the file

* iceberg maintenance: assert a skipped bin's files stay in the snapshot

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>

weed/worker/tasks/iceberg/compact.go

@@ -851,12 +851,16 @@ func readPositionDeleteFile(
 	ctx context.Context,
 	filerClient filer_pb.SeaweedFilerClient,
 	bucketName, dataPath, filePath string,
-) (map[string][]int64, error) {
+) (_ map[string][]int64, err error) {
+	defer recoverRead(&err, filePath)
 	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, filePath)
 	if err != nil {
 		return nil, err
 	}
-	reader := parquet.NewReader(bytes.NewReader(data))
+	reader, err := newParquetReader(data, filePath)
+	if err != nil {
+		return nil, err
+	}
 	defer reader.Close()
 
 	pqSchema := reader.Schema()
@@ -1000,12 +1004,16 @@ func readEqualityDeleteFile(
 	bucketName, dataPath, filePath string,
 	fieldIDs []int,
 	icebergSchema *iceberg.Schema,
-) (map[string]struct{}, error) {
+) (_ map[string]struct{}, err error) {
+	defer recoverRead(&err, filePath)
 	data, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, filePath)
 	if err != nil {
 		return nil, err
 	}
-	reader := parquet.NewReader(bytes.NewReader(data))
+	reader, err := newParquetReader(data, filePath)
+	if err != nil {
+		return nil, err
+	}
 	defer reader.Close()
 
 	colIndices, err := resolveEqualityColIndices(reader.Schema(), fieldIDs, icebergSchema, false)
@@ -1255,7 +1263,10 @@ func mergeParquetFilesInto(
 	if err != nil {
 		return nil, 0, fmt.Errorf("read parquet file %s: %w", entries[0].DataFile().FilePath(), err)
 	}
-	firstReader := parquet.NewReader(bytes.NewReader(firstData))
+	firstReader, err := newParquetReader(firstData, entries[0].DataFile().FilePath())
+	if err != nil {
+		return nil, 0, err
+	}
 	parquetSchema := firstReader.Schema()
 	if parquetSchema == nil {
 		firstReader.Close()
@@ -1307,8 +1318,9 @@ func mergeParquetFilesInto(
 	var outputBuf bytes.Buffer
 	writer := parquet.NewWriter(&outputBuf, outputSchema, parquet.MaxRowsPerRowGroup(rowGroupRows))
 
-	drainReader := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) {
+	drainReader := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (_ int64, err error) {
 		source := entry.DataFile().FilePath()
+		defer recoverRead(&err, source)
 		conv, groups, err := joins(reader, entry)
 		if err != nil {
 			reader.Close()
@@ -1350,7 +1362,11 @@ func mergeParquetFilesInto(
 			return nil, 0, fmt.Errorf("read parquet file %s: %w", entry.DataFile().FilePath(), err)
 		}
 
-		reader := parquet.NewReader(bytes.NewReader(data))
+		reader, err := newParquetReader(data, entry.DataFile().FilePath())
+		if err != nil {
+			writer.Close()
+			return nil, 0, err
+		}
 		rowsWritten, err := drainReader(reader, entry)
 		if err != nil {
 			writer.Close()
@@ -1506,7 +1522,10 @@ func mergeParquetFilesSorted(
 	if err != nil {
 		return nil, 0, fmt.Errorf("read parquet file %s: %w", entries[0].DataFile().FilePath(), err)
 	}
-	firstReader := parquet.NewReader(bytes.NewReader(firstData))
+	firstReader, err := newParquetReader(firstData, entries[0].DataFile().FilePath())
+	if err != nil {
+		return nil, 0, err
+	}
 	parquetSchema := firstReader.Schema()
 	if parquetSchema == nil {
 		firstReader.Close()
@@ -1568,7 +1587,8 @@ func mergeParquetFilesSorted(
 		}
 	}()
 
-	writeRows := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (int64, error) {
+	writeRows := func(reader *parquet.Reader, entry iceberg.ManifestEntry) (_ int64, err error) {
+		defer recoverRead(&err, entry.DataFile().FilePath())
 		return copyFileRows(ctx, reader, entry, layout, bucketName, dataPath, positionDeletes, resolvedEqGroups, eqDeleteGroups, icebergSchema, func(filtered []parquet.Row) error {
 			// The writer's row buffer copies the values it is handed, so these
 			// rows need no cloning out of the reader's own buffer first.
@@ -1594,7 +1614,10 @@ func mergeParquetFilesSorted(
 			return nil, 0, fmt.Errorf("read parquet file %s: %w", entry.DataFile().FilePath(), err)
 		}
 
-		reader := parquet.NewReader(bytes.NewReader(data))
+		reader, err := newParquetReader(data, entry.DataFile().FilePath())
+		if err != nil {
+			return nil, 0, err
+		}
 		if layout == nil && !schemasEqual(parquetSchema, reader.Schema()) {
 			reader.Close()
 			return nil, 0, fmt.Errorf("schema mismatch in %s: cannot merge files with different schemas", entry.DataFile().FilePath())

weed/worker/tasks/iceberg/detection.go

@@ -156,7 +156,11 @@ func (h *Handler) scanTablesForMaintenance(
 				}
 				effective := resolveTableConfig(config, state.Metadata.Properties(), maintenance)
 
-				needsWork, err := h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, effective, tableOps)
+				// A table whose files cannot be read is skipped for this pass;
+				// the other tables are still evaluated.
+				needsWork, err := guardRead(path.Join(bucketName, tablePath), func() (bool, error) {
+					return h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, effective, tableOps)
+				})
 				if err != nil {
 					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot evaluate maintenance need: %v", bucketName, nsName, tblName, err)
 					continue

weed/worker/tasks/iceberg/filer_io.go

@@ -325,6 +325,18 @@ func parquetSchemaByIcebergPath(ctx context.Context, client filer_pb.SeaweedFile
 	return f.Schema(), nil
 }
 
+// newParquetReader opens data as a parquet file. parquet.NewReader panics on
+// a file it cannot open (wrong magic, a truncated footer, zero bytes);
+// parquet.OpenFile returns the same failures as errors, and NewReader on the
+// opened file does not open it again.
+func newParquetReader(data []byte, name string) (*parquet.Reader, error) {
+	f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)))
+	if err != nil {
+		return nil, fmt.Errorf("open parquet file %s: %w", name, err)
+	}
+	return parquet.NewReader(f), nil
+}
+
 // normalizeIcebergPath converts an Iceberg path (an S3 URL, an absolute filer
 // path, or a path relative to the table root) into a path relative to the
 // bucket root, e.g. "ns/table/metadata/snap-1.avro". Absolute references are

weed/worker/tasks/iceberg/handler.go

@@ -2,6 +2,7 @@ package iceberg
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"path"
 	"strings"
@@ -602,28 +603,36 @@ func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequ
 		var opErr error
 		var opMetrics map[string]int64
 
-		switch op {
-		case "compact":
-			opResult, opMetrics, opErr = h.compactDataFiles(ctx, filerClient, bucketName, tablePath, workerConfig, func(binIdx, totalBins int) {
-				binProgress := progress + float64(binIdx+1)/float64(totalBins)*(100.0/float64(totalOps))
-				_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
-					JobId:           request.Job.JobId,
-					JobType:         canonicalJobType,
-					State:           plugin_pb.JobState_JOB_STATE_RUNNING,
-					ProgressPercent: binProgress,
-					Stage:           fmt.Sprintf("compact bin %d/%d", binIdx+1, totalBins),
-					Message:         fmt.Sprintf("compacting bin %d of %d", binIdx+1, totalBins),
+		// A panic in an operation (a row decoder on a corrupt page) fails that
+		// operation for this table instead of ending the worker process.
+		opResult, opMetrics, opErr = func() (res string, metrics map[string]int64, err error) {
+			defer recoverRead(&err, fmt.Sprintf("%s on %s/%s", op, bucketName, tablePath))
+			switch op {
+			case "compact":
+				return h.compactDataFiles(ctx, filerClient, bucketName, tablePath, workerConfig, func(binIdx, totalBins int) {
+					binProgress := progress + float64(binIdx+1)/float64(totalBins)*(100.0/float64(totalOps))
+					_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
+						JobId:           request.Job.JobId,
+						JobType:         canonicalJobType,
+						State:           plugin_pb.JobState_JOB_STATE_RUNNING,
+						ProgressPercent: binProgress,
+						Stage:           fmt.Sprintf("compact bin %d/%d", binIdx+1, totalBins),
+						Message:         fmt.Sprintf("compacting bin %d of %d", binIdx+1, totalBins),
+					})
 				})
-			})
-		case "rewrite_position_delete_files":
-			opResult, opMetrics, opErr = h.rewritePositionDeleteFiles(ctx, filerClient, bucketName, tablePath, workerConfig)
-		case "expire_snapshots":
-			opResult, opMetrics, opErr = h.expireSnapshots(ctx, filerClient, bucketName, tablePath, workerConfig)
-		case "remove_orphans":
-			opResult, opMetrics, opErr = h.removeOrphans(ctx, filerClient, bucketName, tablePath, workerConfig)
-		case "rewrite_manifests":
-			opResult, opMetrics, opErr = h.rewriteManifests(ctx, filerClient, bucketName, tablePath, workerConfig)
-		default:
+			case "rewrite_position_delete_files":
+				return h.rewritePositionDeleteFiles(ctx, filerClient, bucketName, tablePath, workerConfig)
+			case "expire_snapshots":
+				return h.expireSnapshots(ctx, filerClient, bucketName, tablePath, workerConfig)
+			case "remove_orphans":
+				return h.removeOrphans(ctx, filerClient, bucketName, tablePath, workerConfig)
+			case "rewrite_manifests":
+				return h.rewriteManifests(ctx, filerClient, bucketName, tablePath, workerConfig)
+			default:
+				return "", nil, errUnknownOperation
+			}
+		}()
+		if errors.Is(opErr, errUnknownOperation) {
 			glog.Warningf("unknown maintenance operation: %s", op)
 			continue
 		}
@@ -737,3 +746,6 @@ func (h *Handler) connectToFiler(ctx context.Context, grpcAddresses []string) (s
 
 // Ensure Handler implements JobHandler.
 var _ pluginworker.JobHandler = (*Handler)(nil)
+
+// errUnknownOperation is returned for a maintenance operation Execute does not know.
+var errUnknownOperation = errors.New("unknown maintenance operation")

weed/worker/tasks/iceberg/read_guard.go

@@ -0,0 +1,28 @@
+package iceberg
+
+import (
+	"fmt"
+	"runtime/debug"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// recoverRead turns a panic into *errp, naming what was being read (a file,
+// or a table operation).
+// parquet-go panics on more than a file it cannot open: a corrupt page under
+// a valid footer panics in its row decoder. Nothing above a maintenance job
+// recovers, so such a panic would end the process hosting the worker; with
+// this the job fails for that file instead. Use as
+// `defer recoverRead(&err, what)` in a function with a named error result.
+func recoverRead(errp *error, what string) {
+	if r := recover(); r != nil {
+		glog.Errorf("iceberg maintenance: recovered panic in %s: %v\n%s", what, r, debug.Stack())
+		*errp = fmt.Errorf("%s: panic: %v", what, r)
+	}
+}
+
+// guardRead runs fn, turning a panic into an error naming what.
+func guardRead[T any](what string, fn func() (T, error)) (v T, err error) {
+	defer recoverRead(&err, what)
+	return fn()
+}

weed/worker/tasks/iceberg/read_guard_test.go

@@ -0,0 +1,22 @@
+package iceberg
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestGuardReadTurnsAPanicIntoAnError(t *testing.T) {
+	v, err := guardRead("data/x.parquet", func() (int, error) {
+		var s []int
+		return s[3], nil // index out of range
+	})
+	if v != 0 || err == nil || !strings.Contains(err.Error(), "data/x.parquet") || !strings.Contains(err.Error(), "panic") {
+		t.Fatalf("v = %d, err = %v: want a zero value and an error naming the file", v, err)
+	}
+
+	want := errors.New("plain error")
+	if _, err := guardRead("data/y.parquet", func() (int, error) { return 0, want }); err != want {
+		t.Fatalf("err = %v, want the function's own error unchanged", err)
+	}
+}

weed/worker/tasks/iceberg/unreadable_file_test.go

@@ -0,0 +1,271 @@
+package iceberg
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/apache/iceberg-go"
+	"github.com/parquet-go/parquet-go"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+)
+
+type positionDeleteTestRow struct {
+	FilePath string `parquet:"file_path"`
+	Pos      int64  `parquet:"pos"`
+}
+
+func positionDeleteParquet(t *testing.T, rows []positionDeleteTestRow) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	w := parquet.NewGenericWriter[positionDeleteTestRow](&buf)
+	if _, err := w.Write(rows); err != nil {
+		t.Fatalf("write parquet: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close parquet: %v", err)
+	}
+	return buf.Bytes()
+}
+
+// noPanic runs fn and turns a panic into a test failure. A panic in a
+// maintenance goroutine is not recovered anywhere and ends the process
+// hosting the worker.
+func noPanic(t *testing.T, fn func()) {
+	t.Helper()
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("panicked: %v", r)
+		}
+	}()
+	fn()
+}
+
+// putInlineFile stores body as the file at an Iceberg path of the table.
+func putInlineFile(t *testing.T, fs *fakeFilerServer, bucket, dataPath, file string, body []byte) {
+	t.Helper()
+	full, err := icebergFilerPath(bucket, dataPath, file)
+	if err != nil {
+		t.Fatalf("path: %v", err)
+	}
+	fs.putEntry(path.Dir(full), path.Base(full), &filer_pb.Entry{
+		Name:       path.Base(full),
+		Content:    body,
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(body))},
+	})
+}
+
+// corruptPageFile is a position delete file whose footer is intact but some
+// bytes of its data pages are altered (the writer recorded no page CRC). It
+// opens; decoding its rows panics inside parquet-go.
+func corruptPageFile(t *testing.T) []byte {
+	t.Helper()
+	b, err := os.ReadFile("testdata/corrupt_page_deletes.parquet")
+	if err != nil {
+		t.Fatalf("read testdata: %v", err)
+	}
+	return b
+}
+
+func TestUnopenableParquetFileIsAnErrorNotAPanic(t *testing.T) {
+	good := positionDeleteParquet(t, []positionDeleteTestRow{{"s3://b/ns/t/data/f1.parquet", 1}})
+	for name, body := range map[string][]byte{
+		"wrong magic":      append([]byte("XXXX"), good[4:]...),
+		"truncated footer": good[:len(good)-6],
+		"not parquet":      []byte(`{"not": "parquet"}`),
+		"empty":            {},
+	} {
+		t.Run(name, func(t *testing.T) {
+			fs, client := startFakeFiler(t)
+			const bucket, dataPath, file = "b", "ns/t", "data/bad-deletes.parquet"
+			putInlineFile(t, fs, bucket, dataPath, file, body)
+
+			var err error
+			noPanic(t, func() {
+				_, err = readPositionDeleteFile(context.Background(), client, bucket, dataPath, file)
+			})
+			if err == nil || !strings.Contains(err.Error(), file) {
+				t.Fatalf("position delete read: err = %v, want an error naming %s", err, file)
+			}
+			noPanic(t, func() {
+				_, err = readEqualityDeleteFile(context.Background(), client, bucket, dataPath, file, []int{1}, newTestSchema())
+			})
+			if err == nil || !strings.Contains(err.Error(), file) {
+				t.Fatalf("equality delete read: err = %v, want an error naming %s", err, file)
+			}
+		})
+	}
+}
+
+func TestCorruptPageIsAnErrorNamingTheFileNotAPanic(t *testing.T) {
+	body := corruptPageFile(t)
+
+	// Control: parquet-go itself panics while decoding this file's rows.
+	panicked := func() (p bool) {
+		defer func() { p = recover() != nil }()
+		r := parquet.NewReader(bytes.NewReader(body))
+		defer r.Close()
+		rows := make([]parquet.Row, 64)
+		for {
+			if _, err := r.ReadRows(rows); err != nil {
+				return false
+			}
+		}
+	}()
+	if !panicked {
+		t.Fatal("control: decoding the corrupt page no longer panics parquet-go; the test file needs replacing")
+	}
+
+	fs, client := startFakeFiler(t)
+	const bucket, dataPath, file = "b", "ns/t", "data/corrupt-deletes.parquet"
+	putInlineFile(t, fs, bucket, dataPath, file, body)
+	var err error
+	noPanic(t, func() {
+		_, err = readPositionDeleteFile(context.Background(), client, bucket, dataPath, file)
+	})
+	if err == nil || !strings.Contains(err.Error(), file) || !strings.Contains(err.Error(), "panic") {
+		t.Fatalf("err = %v, want an error naming %s", err, file)
+	}
+}
+
+func TestCompactionWithACorruptDeleteFileFailsWithoutCommitting(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
+	populateTableWithDeleteFiles(t, fs, setup,
+		[]struct {
+			Name string
+			Rows []struct {
+				ID   int64
+				Name string
+			}
+		}{
+			{"d1.parquet", []struct {
+				ID   int64
+				Name string
+			}{{1, "alice"}, {2, "bob"}}},
+			{"d2.parquet", []struct {
+				ID   int64
+				Name string
+			}{{3, "carol"}}},
+		},
+		[]struct {
+			Name string
+			Rows []struct {
+				FilePath string
+				Pos      int64
+			}
+		}{
+			{"pd1.parquet", []struct {
+				FilePath string
+				Pos      int64
+			}{{"data/d1.parquet", 1}}},
+		},
+		nil,
+	)
+	dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
+	body := corruptPageFile(t)
+	fs.putEntry(dataDir, "pd1.parquet", &filer_pb.Entry{Name: "pd1.parquet", Content: body,
+		Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(len(body))}})
+	before, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("metadata: %v", err)
+	}
+
+	handler := NewHandler(nil)
+	config := Config{TargetFileSizeBytes: 256 * 1024 * 1024, MinInputFiles: 2, MaxCommitRetries: 3, ApplyDeletes: true}
+	var result string
+	noPanic(t, func() {
+		result, _, err = handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	})
+	if err == nil || !strings.Contains(err.Error(), "pd1.parquet") {
+		t.Fatalf("result %q, err = %v: want an error naming the delete file", result, err)
+	}
+	after, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath())
+	if err != nil {
+		t.Fatalf("metadata: %v", err)
+	}
+	if after.Metadata.CurrentSnapshot().SnapshotID != before.Metadata.CurrentSnapshot().SnapshotID {
+		t.Fatal("a snapshot was committed without the delete file applied")
+	}
+}
+
+func TestAnUnreadableDataFileSkipsItsBinAndTheOtherBinsCompact(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	partitionSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		SourceIDs: []int{2}, FieldID: 1000, Name: "name", Transform: iceberg.IdentityTransform{},
+	})
+	setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"}
+	row := func(id int64, name string) []struct {
+		ID   int64
+		Name string
+	} {
+		return []struct {
+			ID   int64
+			Name string
+		}{{id, name}}
+	}
+	populatePartitionedDataTable(t, fs, setup, partitionSpec, [][]partitionedTestFile{
+		{{Name: "us-1.parquet", Partition: map[int]any{1000: "us"}, Rows: row(1, "us")}},
+		{{Name: "us-2.parquet", Partition: map[int]any{1000: "us"}, Rows: row(2, "us")}},
+		{
+			{Name: "eu-1.parquet", Partition: map[int]any{1000: "eu"}, Rows: row(3, "eu")},
+			{Name: "eu-2.parquet", Partition: map[int]any{1000: "eu"}, Rows: row(4, "eu")},
+		},
+	})
+	dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data")
+	eu2 := fs.getEntry(dataDir, "eu-2.parquet")
+	broken := append([]byte("XXXX"), eu2.Content[4:]...) // wrong magic
+	fs.putEntry(dataDir, "eu-2.parquet", &filer_pb.Entry{Name: "eu-2.parquet", Content: broken, Attributes: eu2.Attributes})
+
+	handler := NewHandler(nil)
+	config := Config{TargetFileSizeBytes: 256 * 1024 * 1024, MinInputFiles: 2, MaxCommitRetries: 3}
+	var result string
+	var err error
+	noPanic(t, func() {
+		result, _, err = handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil)
+	})
+	if err != nil {
+		t.Fatalf("compactDataFiles: %v", err)
+	}
+	if !strings.Contains(result, "compacted 2 files into 1") {
+		t.Fatalf("result %q: want the readable bin compacted", result)
+	}
+	live := liveDataFiles(t, client, setup)
+	for _, name := range []string{"eu-1.parquet", "eu-2.parquet"} {
+		found := false
+		for _, df := range live {
+			if strings.HasSuffix(df.FilePath(), name) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Errorf("the skipped bin's %s is not in the current snapshot", name)
+		}
+	}
+}
+
+func TestMergeFailsOnACorruptPageNamingTheFile(t *testing.T) {
+	fs, client := startFakeFiler(t)
+	fs.putEntry("/buckets/test-bucket/ns/tbl/data", "corrupt.parquet", &filer_pb.Entry{
+		Name:    "corrupt.parquet",
+		Content: corruptPageFile(t),
+	})
+
+	// Its footer reads fine, so the panic comes from its pages inside the
+	// merge's row decoder.
+	var err error
+	noPanic(t, func() {
+		_, _, err = mergeParquetFilesInto(context.Background(), client, "test-bucket", "ns/tbl",
+			[]iceberg.ManifestEntry{schemaFileEntry(t, "data/corrupt.parquet", 1)},
+			nil, nil, newTestSchema(), 0, false, nil)
+	})
+	if err == nil || !strings.Contains(err.Error(), "corrupt.parquet") || !strings.Contains(err.Error(), "panic") {
+		t.Fatalf("err = %v, want an error naming corrupt.parquet", err)
+	}
+}