From edf2b0f5eeb81ac188d0dd9169b8538c1efdd56c Mon Sep 17 00:00:00 2001 From: Javier de la Torre Date: Thu, 29 Oct 2026 08:10:47 +0100 Subject: [PATCH] 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 --- weed/worker/tasks/iceberg/compact.go | 43 ++- weed/worker/tasks/iceberg/detection.go | 6 +- weed/worker/tasks/iceberg/filer_io.go | 12 + weed/worker/tasks/iceberg/handler.go | 54 ++-- weed/worker/tasks/iceberg/read_guard.go | 28 ++ weed/worker/tasks/iceberg/read_guard_test.go | 22 ++ .../testdata/corrupt_page_deletes.parquet | Bin 0 -> 17386 bytes .../tasks/iceberg/unreadable_file_test.go | 271 ++++++++++++++++++ 8 files changed, 404 insertions(+), 32 deletions(-) create mode 100644 weed/worker/tasks/iceberg/read_guard.go create mode 100644 weed/worker/tasks/iceberg/read_guard_test.go create mode 100644 weed/worker/tasks/iceberg/testdata/corrupt_page_deletes.parquet create mode 100644 weed/worker/tasks/iceberg/unreadable_file_test.go diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index bef7990d53..40bd84005b 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/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()) diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index 96b04bb326..3737311836 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/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 diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 440514c790..b2a447eeb7 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/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 diff --git a/weed/worker/tasks/iceberg/handler.go b/weed/worker/tasks/iceberg/handler.go index 1923a4abfe..9f8412bb9d 100644 --- a/weed/worker/tasks/iceberg/handler.go +++ b/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") diff --git a/weed/worker/tasks/iceberg/read_guard.go b/weed/worker/tasks/iceberg/read_guard.go new file mode 100644 index 0000000000..1c6d4a248e --- /dev/null +++ b/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() +} diff --git a/weed/worker/tasks/iceberg/read_guard_test.go b/weed/worker/tasks/iceberg/read_guard_test.go new file mode 100644 index 0000000000..ddeb26e1e2 --- /dev/null +++ b/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) + } +} diff --git a/weed/worker/tasks/iceberg/testdata/corrupt_page_deletes.parquet b/weed/worker/tasks/iceberg/testdata/corrupt_page_deletes.parquet new file mode 100644 index 0000000000000000000000000000000000000000..1468b8cf64e9a4044a6a5e08b98bd99b6dcfc068 GIT binary patch literal 17386 zcmcKCb+8o2z5w6_4(@V*gS)#s9NgXA-QC^Y-QC^Y-ARbM5hMZ;Ai_H*e7QI8zSR3G zw`-=p?cYvK|EhO-dTVEAYKPX%8ypn8F=E(KJvdHqAP|Tjw)V?kEO+jRO1q=LLu;AYX3;h`k4g{}`6}IFEUY$BD+bF?N0)a?jX|7Hk zAyVk%i^=XZPM@q>`F1lpZ;wBt;MzSE3*7$)gQ5S8o&I&7q5swN!&(ac$NS{3>4#+> z`rqsQ*DZzqvmfbyq#u@j=s$e$Wc|+r@*h4-|L-9aHb!Q?j1a#5F-ijeYm|hIl|M=Z zMr8OP3NJyg2}EYt`1+$nWi&=-48~+E#%3JGWjuzTM#4*i*8~zW5yQ_v;U)2F0!f&Z zAxy^POu>{)#nep0v`okJ%)pGy#LUdXtjxyj%)y+@#oWxpyv)b^EWm;+#KJ7XqAbSZ zEWwg2#nLRpP?lvmmS+W4WF=N+6;@?6R%Z>?WG&Wa9oA(%)@K7YWFt0a6E?yQj^_kUZs!i}!9`5Bn?&kp>vnSjLkTV%Xo~>1Wd?83}#{`VN!-L8Iv;wQ!*7(GY!)+9n&)dGcpr1GYhja z8?!S9b21lmGY|7JAM>*S3$hRkvj~f_7>lz6OR^M8vkXI7mgQKU6h>!V%-|$<0$M5-+&-ep>H>8XGBI~5F;}RqcR$!GX`Ta7GpCG<1!xOGXWDa5rdhSNtl!& zOvdC)!IVtJ)J(&)Ovm)hz>Lhq%*?{9%*O1@!JN#++|0wg%*XsJz=ABq!Ysm~EXLw2 z!ICV+(k#PJmSs7XX9ZSdC01q?R%JC-XARb5E!Jio)@41`X9G55BQ|CeHf1w5XA8Du zE4F4Ewq-lEX9sp4slgh;Qy2Sj8f z1~D?DFe;-lI%6;HnS@Ch!emU&6imrfOwBY*%XCc749v(( z%*-sz%52Qe9L&jF%*{N^%Y4kw0xZZvEX*P-%3>_e5-iD5EX^_u<)7`8Q!md7tjJ2N z%qpzPYOKy0tjSue%{r{hdaTa|Y{*7z%qDEgW^B$DY{^z^%{FYyc5KfM?8r{+%r5N8 zZtTt;?8#p2%|7hQe(cWy9LPZ&%pn}gVI0m89LZ4}%`qIyaU9PHoXAO>%qg78X`Id( zoXJ_7%{iRQd7RG$T*yUS%q3jPWn9h`T**~j%{5%hbzIL4+{jJb%q`r?ZQRZs+{sl%p*L?V}1ZXu6}|id5WibhG%(>=Xrq_d5M>Kg;#lv*Lj0Cd5gDs zhj)38Z}3gt=UaT6@9#ZySN_J|`HFvD;Exbw{%0fxF*2htDx)zvV=yLTF*f5cF5@vi z6EGnYF_?*&gh?5~WK7N!OvzMC%`{BQbWG0-%*ag4%q+~xY|PFa%*kBL%{5D)VRkMbCg z^8`=w6i@RE&+;74^8zpO5-;-#uksqN^9FD77H{(o@A4ks;G4Y9xA->S;k$f~@ACtG z$dC9jKjEkRjGyxhe#x)+fM4?=AMr7t@Ed;1@Ay5R@)>{NkNk;0^EqGeC4b?s{Eff! z761HWUxdi!e@0>uBQpx4G8&^Z24gZ7V>1rpG9KeI0TVJ2gPE8~n3N$*#^g-FluX6c zOvAKH$Mnp=jLgK$%)+e9#_Y_&oXo}C%)`9Q$NVh7f-JZ*O@i33@D39?tPw*s9@ifoyEYI;gFYqES z@iMRQDzEW6Z}28>@iy=9F7NRTzRCN1i*NHCzRUN*+z{4t@2h{n5BU*4<|q7=pYd~k z!7uq0AMk5Fv=x{BQg?$ z7@1KRmC+cTF&LAv7@Khzm+=^%37C+H7=B|XLAncavtY%0T*%+7jp@hav7I%1y^zvS91;5avj%)+c|75QQyc-+{`W9%5B`v z9o)%X+|51Q%YEF>13bt>Jj^3J%40mv6FkXNJk2va%X2)>3%tlnyv!@S%4@vN8@$O| zyv;kj%X@r-Z}L9h;@fCL`Gr|BQpx4G8&^Z24gZ7V>1rp zG9KeI0TVJ2gPE8~n3N$*#^g-FluX6cOvAKH$Mnp=jLgK$%)+e9#_Y_&oXo}C%)`9Q z$NVh7f-Ju>a_nJ(DF6h#qB1aKw-j1p`TAw;#~1Tl==H`nBuap+}qGpg?e}j1iMY z@7Se#hgQAY^zWQ85C{$mi$v<(t6#Sn(f;g|q|sha%Hg-WXZf3G*gnJFx;Q0bSnipkg}(@7$x=S7 zOUjo`Gq6M7eqDO?OfxWVu7bG=rpeNFK$q_A)3h(ru6>&}`SW*dpEV>$c!!7WpnBLX K0JX!O^6@{f;J`Bg literal 0 HcmV?d00001 diff --git a/weed/worker/tasks/iceberg/unreadable_file_test.go b/weed/worker/tasks/iceberg/unreadable_file_test.go new file mode 100644 index 0000000000..5ed64b08d9 --- /dev/null +++ b/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) + } +}