From 0d00e921789731777281f765eb6790eb3349f5e4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 20:49:49 -0700 Subject: [PATCH] iceberg maintenance: match percent-decoded keys when checking file references (#11978) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * iceberg maintenance: match percent-decoded keys when checking file references Writers may percent-escape partition directory names in the file paths they record in manifests (e.g. "part%3D1"), while the S3 layer decodes the key before storing the object. remove_orphans compared the escaped reference to the decoded filer path, counted the live file as an orphan, and deleted it once it passed the age window — readers then got NoSuchKey on the next scan. Match both spellings of every referenced path in remove_orphans, apply the same coverage to the kept-file set in expire_snapshots, and delete expired files at their decoded key so the object the S3 layer stored is the one removed. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: retry filer lookups at the percent-decoded key A manifest reference carrying an escaped segment resolves to a filer name that does not exist, because the S3 layer decoded the key on write. When a lookup misses and the reference contains escapes, retry once at the decoded spelling so maintenance can still read the file. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: resolve referenced files verbatim-first, decoded fallback filer_pb.LookupEntry reports misses as ErrNotFound rather than a gRPC NotFound, so the escaped-reference retry never fired against a real filer. Both maintenance reads and expired-snapshot deletes now resolve through one verbatim-first/decoded-fallback lookup, and expiry refuses to delete a resolved path outside the table's data directory. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: test expiry of a file stored at a literal percent key * iceberg maintenance: resolve referenced files in one filer lookup lookupFileByIcebergPath re-fetched the entry resolveFilerPath had just looked up, doubling the RPC per referenced file during compaction and expiration. The resolver now returns the entry alongside the path so a hit costs one lookup and a decoded fallback two. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: keep every decoded spelling of a referenced path The kept-reference sets recorded a path plus a single decode of it, but lookupResolvedEntry decodes recursively until the spelling is stable. A doubly-escaped reference could resolve to a spelling absent from the set and be deleted while still referenced. addFilerPathSpellings now walks the same decode chain the resolver does. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * iceberg maintenance: keep every decoded spelling of a referenced path * iceberg maintenance: test expiry keeps a doubly-escaped kept reference * iceberg maintenance: clean decoded spellings in the referenced-path set * iceberg maintenance: test expiry keeps a kept reference resolved via decoded parent segment * iceberg maintenance: decode only the reference in the spelling set Decoding the composed filer path also decoded the table directory, so a table at a literal percent-escaped path produced a kept set missing the spelling the resolver returns, and expiry could delete a live file. * test: gofmt the s3tables cluster helpers Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/worker/tasks/iceberg/exec_test.go | 435 ++++++++++++++++++++++++ weed/worker/tasks/iceberg/filer_io.go | 75 +++- weed/worker/tasks/iceberg/operations.go | 24 +- 3 files changed, 509 insertions(+), 25 deletions(-) diff --git a/weed/worker/tasks/iceberg/exec_test.go b/weed/worker/tasks/iceberg/exec_test.go index 18cb661ca6..172e896a1e 100644 --- a/weed/worker/tasks/iceberg/exec_test.go +++ b/weed/worker/tasks/iceberg/exec_test.go @@ -4435,3 +4435,438 @@ func TestResolveCompactionRewritePlanFallsBackForUnsupportedSortTransform(t *tes t.Fatalf("expected fallback to %q for unsupported transform, got %+v", defaultRewriteStrategy, plan) } } + +// A writer may percent-escape a partition directory name ("part=1" becomes +// "part%3D1") in the file paths it records in manifests, while the S3 layer +// decodes the key before storing it. Orphan cleanup must match the decoded +// spelling or it deletes live data files once they pass the age window. +func TestRemoveOrphansKeepsPercentEscapedReferences(t *testing.T) { + fs, client := startFakeFiler(t) + + now := time.Now().UnixMilli() + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "ns", + TableName: "tbl", + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"}, + }, + } + meta := populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + oldTime := time.Now().Add(-200 * time.Hour).Unix() + + // Rewrite the manifest so its entries carry the escaped spelling the + // writer recorded. + escapedPath := fmt.Sprintf("s3://%s/%s/data/month_collected_at_1%%3D681/f.parquet", setup.BucketName, setup.tablePath()) + dfBuilder, err := iceberg.NewDataFileBuilder( + meta.PartitionSpec(), + iceberg.EntryContentData, + escapedPath, + iceberg.ParquetFile, + map[int]any{}, + nil, nil, + 10, + 4096, + ) + if err != nil { + t.Fatalf("build escaped data file: %v", err) + } + snapID := int64(1) + var manifestBuf bytes.Buffer + if _, err := iceberg.WriteManifest( + setup.fileRef("metadata", "manifest-1.avro"), &manifestBuf, + meta.Version(), meta.PartitionSpec(), meta.CurrentSchema(), snapID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())}, + ); err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, "manifest-1.avro", &filer_pb.Entry{ + Name: "manifest-1.avro", + Attributes: &filer_pb.FuseAttributes{Mtime: oldTime, FileSize: uint64(manifestBuf.Len())}, + Content: manifestBuf.Bytes(), + }) + + // The object lives at the decoded key, the same place the S3 PUT landed it. + fs.putEntry(dataDir, "month_collected_at_1=681", &filer_pb.Entry{ + Name: "month_collected_at_1=681", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{Mtime: oldTime}, + }) + partDir := path.Join(dataDir, "month_collected_at_1=681") + fs.putEntry(partDir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", + Attributes: &filer_pb.FuseAttributes{Mtime: oldTime, FileSize: 4096}, + Content: []byte("fake-parquet-data"), + }) + + handler := NewHandler(nil) + config := Config{OrphanOlderThanHours: 0, 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 1 orphan") { + t.Errorf("expected only the unreferenced snap-1-data.parquet removed, got %q", result) + } + if fs.getEntry(partDir, "f.parquet") == nil { + t.Error("data file referenced through an escaped manifest path was deleted") + } +} + +// The sibling of the orphan-sweep case on the read side: a manifest reference +// carrying a percent-escaped segment must find the object the S3 layer stored +// at the decoded key. +func TestLookupFileByIcebergPathDecodesEscapedReference(t *testing.T) { + fs, client := startFakeFiler(t) + + dir := path.Join(s3tables.TablesPath, "bkt", "ns/tbl/data/part=1") + fs.putEntry(dir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", + Content: []byte("data"), + }) + literalDir := path.Join(s3tables.TablesPath, "bkt", "ns/tbl/data/part%3D2") + fs.putEntry(literalDir, "g.parquet", &filer_pb.Entry{ + Name: "g.parquet", + Content: []byte("data"), + }) + + entry, err := lookupFileByIcebergPath(context.Background(), client, "bkt", "ns/tbl", "s3://bkt/ns/tbl/data/part%3D1/f.parquet") + if err != nil { + t.Fatalf("escaped reference should resolve to the decoded key: %v", err) + } + if entry == nil || entry.Name != "f.parquet" { + t.Fatalf("resolved to wrong entry: %+v", entry) + } + + // A file genuinely stored under an escaped name still resolves verbatim. + entry, err = lookupFileByIcebergPath(context.Background(), client, "bkt", "ns/tbl", "s3://bkt/ns/tbl/data/part%3D2/g.parquet") + if err != nil { + t.Fatalf("literal escaped key should resolve: %v", err) + } + if entry == nil || entry.Name != "g.parquet" { + t.Fatalf("resolved to wrong entry: %+v", entry) + } + + // A real filer reports misses as filer_pb.ErrNotFound, not a gRPC NotFound + // status; the decoded retry must trigger on that shape too. + entry, err = lookupFileByIcebergPath(context.Background(), errNotFoundClient{client}, "bkt", "ns/tbl", "s3://bkt/ns/tbl/data/part%3D1/f.parquet") + if err != nil { + t.Fatalf("escaped reference should resolve under ErrNotFound misses: %v", err) + } + if entry == nil || entry.Name != "f.parquet" { + t.Fatalf("resolved to wrong entry: %+v", entry) + } +} + +// errNotFoundClient converts gRPC NotFound misses into filer_pb.ErrNotFound, +// the shape a real filer produces through filer_pb.LookupEntry. +type errNotFoundClient struct { + filer_pb.SeaweedFilerClient +} + +func (c errNotFoundClient) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + resp, err := c.SeaweedFilerClient.LookupDirectoryEntry(ctx, req, opts...) + if status.Code(err) == codes.NotFound { + return nil, filer_pb.ErrNotFound + } + return resp, err +} + +func TestExpireSnapshotsDeletesEscapedReferenceAtExistingKey(t *testing.T) { + fs, client := startFakeFiler(t) + + now := time.Now() + old := now.Add(-10 * 24 * time.Hour) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "ns", + TableName: "tbl", + Age: 10 * 24 * time.Hour, + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now.UnixMilli(), ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 2, TimestampMs: now.UnixMilli() + 1, ManifestList: "metadata/snap-2.avro"}, + }, + } + meta := populateTable(t, fs, setup) + + // Snapshot 1 (expired) references a file the S3 layer stored at the decoded + // key; snapshot 2 (kept) does not reference it. + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + + escapedPath := fmt.Sprintf("s3://%s/%s/data/month%%3D1/f.parquet", setup.BucketName, setup.tablePath()) + dfBuilder, err := iceberg.NewDataFileBuilder( + meta.PartitionSpec(), iceberg.EntryContentData, escapedPath, + iceberg.ParquetFile, map[int]any{}, nil, nil, 10, 4096, + ) + if err != nil { + t.Fatalf("build escaped data file: %v", err) + } + snapID := int64(1) + var manifestBuf bytes.Buffer + if _, err := iceberg.WriteManifest( + setup.fileRef("metadata", "manifest-1.avro"), &manifestBuf, + meta.Version(), meta.PartitionSpec(), meta.CurrentSchema(), snapID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())}, + ); err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, "manifest-1.avro", &filer_pb.Entry{ + Name: "manifest-1.avro", + Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: uint64(manifestBuf.Len())}, + Content: manifestBuf.Bytes(), + }) + + partDir := path.Join(dataDir, "month=1") + fs.putEntry(dataDir, "month=1", &filer_pb.Entry{ + Name: "month=1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix()}, + }) + fs.putEntry(partDir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 4096}, + Content: []byte("fake-parquet-data"), + }) + + handler := NewHandler(nil) + config := Config{ + SnapshotRetentionMs: hoursToMs(0), + MaxSnapshotsToKeep: 1, + MaxCommitRetries: 3, + Operations: "expire_snapshots", + } + + result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config) + if err != nil { + t.Fatalf("expireSnapshots failed: %v", err) + } + t.Logf("expireSnapshots result: %s", result) + + if fs.getEntry(partDir, "f.parquet") != nil { + t.Error("expired file at the decoded key should have been deleted") + } +} + +// A client that escapes its own percent sign on upload leaves an object under +// a literal %3D-containing key while the manifest spells the same reference. +// Expiration must delete the entry that actually exists, not the decoded +// spelling. +func TestExpireSnapshotsDeletesLiteralPercentKey(t *testing.T) { + fs, client := startFakeFiler(t) + + now := time.Now() + old := now.Add(-10 * 24 * time.Hour) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "ns", + TableName: "tbl", + Age: 10 * 24 * time.Hour, + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now.UnixMilli(), ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 2, TimestampMs: now.UnixMilli() + 1, ManifestList: "metadata/snap-2.avro"}, + }, + } + meta := populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + + escapedPath := fmt.Sprintf("s3://%s/%s/data/month%%3D1/f.parquet", setup.BucketName, setup.tablePath()) + dfBuilder, err := iceberg.NewDataFileBuilder( + meta.PartitionSpec(), iceberg.EntryContentData, escapedPath, + iceberg.ParquetFile, map[int]any{}, nil, nil, 10, 4096, + ) + if err != nil { + t.Fatalf("build escaped data file: %v", err) + } + snapID := int64(1) + var manifestBuf bytes.Buffer + if _, err := iceberg.WriteManifest( + setup.fileRef("metadata", "manifest-1.avro"), &manifestBuf, + meta.Version(), meta.PartitionSpec(), meta.CurrentSchema(), snapID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())}, + ); err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, "manifest-1.avro", &filer_pb.Entry{ + Name: "manifest-1.avro", + Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: uint64(manifestBuf.Len())}, + Content: manifestBuf.Bytes(), + }) + + literalDir := path.Join(dataDir, "month%3D1") + fs.putEntry(dataDir, "month%3D1", &filer_pb.Entry{ + Name: "month%3D1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix()}, + }) + fs.putEntry(literalDir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 4096}, + Content: []byte("fake-parquet-data"), + }) + + handler := NewHandler(nil) + config := Config{ + SnapshotRetentionMs: hoursToMs(0), + MaxSnapshotsToKeep: 1, + MaxCommitRetries: 3, + Operations: "expire_snapshots", + } + + result, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config) + if err != nil { + t.Fatalf("expireSnapshots failed: %v", err) + } + t.Logf("expireSnapshots result: %s", result) + + if fs.getEntry(literalDir, "f.parquet") != nil { + t.Error("expired file at the literal percent key should have been deleted") + } +} + +// A client that escapes an already-escaped name writes a manifest reference +// like month%253D1 while the object sits at month=1. The kept set must hold +// every decoding in the chain, or the expiry pass deletes a file a live +// snapshot still references. +func TestExpireSnapshotsKeepsDoublyEscapedKeptReference(t *testing.T) { + fs, client := startFakeFiler(t) + + now := time.Now() + old := now.Add(-10 * 24 * time.Hour) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "ns", + TableName: "tbl", + Age: 10 * 24 * time.Hour, + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now.UnixMilli(), ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 2, TimestampMs: now.UnixMilli() + 1, ManifestList: "metadata/snap-2.avro"}, + }, + } + meta := populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + + escapedPath := fmt.Sprintf("s3://%s/%s/data/month%%253D1/f.parquet", setup.BucketName, setup.tablePath()) + writeRefManifest := func(fileName string, snapID int64) { + dfBuilder, err := iceberg.NewDataFileBuilder( + meta.PartitionSpec(), iceberg.EntryContentData, escapedPath, + iceberg.ParquetFile, map[int]any{}, nil, nil, 10, 4096, + ) + if err != nil { + t.Fatalf("build escaped data file: %v", err) + } + var manifestBuf bytes.Buffer + if _, err := iceberg.WriteManifest( + setup.fileRef("metadata", fileName), &manifestBuf, + meta.Version(), meta.PartitionSpec(), meta.CurrentSchema(), snapID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())}, + ); err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, fileName, &filer_pb.Entry{ + Name: fileName, + Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: uint64(manifestBuf.Len())}, + Content: manifestBuf.Bytes(), + }) + } + writeRefManifest("manifest-1.avro", 1) + writeRefManifest("manifest-2.avro", 2) + + partDir := path.Join(dataDir, "month=1") + fs.putEntry(dataDir, "month=1", &filer_pb.Entry{ + Name: "month=1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix()}, + }) + fs.putEntry(partDir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 4096}, + Content: []byte("fake-parquet-data"), + }) + + handler := NewHandler(nil) + config := Config{ + SnapshotRetentionMs: hoursToMs(0), + MaxSnapshotsToKeep: 1, + MaxCommitRetries: 3, + Operations: "expire_snapshots", + } + + if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil { + t.Fatalf("expireSnapshots failed: %v", err) + } + + if fs.getEntry(partDir, "f.parquet") == nil { + t.Error("file referenced by the kept snapshot through a doubly-escaped path was deleted") + } +} + +// A kept reference whose decoded spelling contains a parent segment resolves +// to the cleaned path. The kept set must record the cleaned spelling, or the +// expiry pass misses the protection and deletes a file a live snapshot still +// references. +func TestExpireSnapshotsKeepsDecodedParentReference(t *testing.T) { + fs, client := startFakeFiler(t) + + now := time.Now() + old := now.Add(-10 * 24 * time.Hour) + setup := tableSetup{ + BucketName: "test-bucket", + Namespace: "ns", + TableName: "tbl", + Age: 10 * 24 * time.Hour, + Snapshots: []table.Snapshot{ + {SnapshotID: 1, TimestampMs: now.UnixMilli(), ManifestList: "metadata/snap-1.avro"}, + {SnapshotID: 2, TimestampMs: now.UnixMilli() + 1, ManifestList: "metadata/snap-2.avro"}, + }, + } + meta := populateTable(t, fs, setup) + + metaDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "metadata") + dataDir := path.Join(s3tables.TablesPath, setup.BucketName, setup.tablePath(), "data") + + writeRefManifest := func(fileName string, snapID int64, filePath string) { + dfBuilder, err := iceberg.NewDataFileBuilder( + meta.PartitionSpec(), iceberg.EntryContentData, filePath, + iceberg.ParquetFile, map[int]any{}, nil, nil, 10, 4096, + ) + if err != nil { + t.Fatalf("build data file: %v", err) + } + var manifestBuf bytes.Buffer + if _, err := iceberg.WriteManifest( + setup.fileRef("metadata", fileName), &manifestBuf, + meta.Version(), meta.PartitionSpec(), meta.CurrentSchema(), snapID, + []iceberg.ManifestEntry{iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())}, + ); err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, fileName, &filer_pb.Entry{ + Name: fileName, + Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: uint64(manifestBuf.Len())}, + Content: manifestBuf.Bytes(), + }) + } + writeRefManifest("manifest-1.avro", 1, fmt.Sprintf("s3://%s/%s/data/f.parquet", setup.BucketName, setup.tablePath())) + writeRefManifest("manifest-2.avro", 2, fmt.Sprintf("s3://%s/%s/data/tmp/%%2e%%2e/f.parquet", setup.BucketName, setup.tablePath())) + + fs.putEntry(dataDir, "f.parquet", &filer_pb.Entry{ + Name: "f.parquet", Attributes: &filer_pb.FuseAttributes{Mtime: old.Unix(), FileSize: 4096}, + Content: []byte("fake-parquet-data"), + }) + + handler := NewHandler(nil) + config := Config{ + SnapshotRetentionMs: hoursToMs(0), + MaxSnapshotsToKeep: 1, + MaxCommitRetries: 3, + Operations: "expire_snapshots", + } + + if _, _, err := handler.expireSnapshots(context.Background(), client, setup.BucketName, setup.tablePath(), config); err != nil { + t.Fatalf("expireSnapshots failed: %v", err) + } + + if fs.getEntry(dataDir, "f.parquet") == nil { + t.Error("file referenced by the kept snapshot through a decoded parent segment was deleted") + } +} diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index 482c8dc80a..9c05e010d2 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "io" + "net/url" "path" "strconv" "strings" @@ -162,26 +163,66 @@ func loadCurrentMetadata(ctx context.Context, client filer_pb.SeaweedFilerClient return state, nil } +func isFilerNotFound(err error) bool { + return errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound +} + +// resolveFilerPath returns the filer path that actually holds a referenced +// file: the verbatim spelling first, then the percent-decoded one. Writers +// may record a path with escaped segments while the S3 layer decoded the key +// before storing the object. +func resolveFilerPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (string, error) { + _, fullPath, err := lookupResolvedEntry(ctx, client, bucketName, dataPath, icebergPath) + return fullPath, err +} + +// lookupResolvedEntry returns the entry and filer path holding an +// Iceberg-style reference under the verbatim-first rule. +func lookupResolvedEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*filer_pb.Entry, string, error) { + fullPath, err := icebergFilerPath(bucketName, dataPath, icebergPath) + if err != nil { + return nil, "", err + } + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ + Directory: path.Dir(fullPath), + Name: path.Base(fullPath), + }) + if err == nil { + return resp.Entry, fullPath, nil + } + if !isFilerNotFound(err) { + return nil, "", err + } + if decoded, decErr := url.PathUnescape(icebergPath); decErr == nil && decoded != icebergPath { + return lookupResolvedEntry(ctx, client, bucketName, dataPath, decoded) + } + return nil, "", err +} + +// addFilerPathSpellings records every filer path lookupResolvedEntry can +// return for a reference: each successive decoding of the reference, with the +// rest of the path left verbatim. An extra spelling only keeps a file; a +// missing one can drop a referenced file. +func addFilerPathSpellings(set map[string]struct{}, bucketName, dataPath, icebergPath string) { + for spelling := icebergPath; ; { + filerPath, err := icebergFilerPath(bucketName, dataPath, spelling) + if err != nil { + return + } + set[filerPath] = struct{}{} + decoded, err := url.PathUnescape(spelling) + if err != nil || decoded == spelling { + return + } + spelling = decoded + } +} + // lookupFileByIcebergPath returns the filer entry holding an Iceberg-style // path. dataPath is the bucket-relative directory the table's files live in. func lookupFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*filer_pb.Entry, error) { - fullPath, err := icebergFilerPath(bucketName, dataPath, icebergPath) - if err != nil { - return nil, err - } - dir, fileName := path.Dir(fullPath), path.Base(fullPath) - - resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ - Directory: dir, - Name: fileName, - }) - if err != nil { - return nil, fmt.Errorf("lookup %s/%s: %w", dir, fileName, err) - } - if resp == nil || resp.Entry == nil { - return nil, fmt.Errorf("file not found: %s/%s", dir, fileName) - } - return resp.Entry, nil + entry, _, err := lookupResolvedEntry(ctx, client, bucketName, dataPath, icebergPath) + return entry, err } // loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. diff --git a/weed/worker/tasks/iceberg/operations.go b/weed/worker/tasks/iceberg/operations.go index 2266ad0ae3..8b7bb22f24 100644 --- a/weed/worker/tasks/iceberg/operations.go +++ b/weed/worker/tasks/iceberg/operations.go @@ -9,6 +9,7 @@ import ( "math/rand/v2" "path" "sort" + "strings" "time" "github.com/apache/iceberg-go" @@ -90,9 +91,7 @@ func (h *Handler) expireSnapshots( // Resolve kept file paths for consistent comparison keptFilerPaths := make(map[string]struct{}, len(keptFiles)) for f := range keptFiles { - if filerPath, err := icebergFilerPath(bucketName, dataPath, f); err == nil { - keptFilerPaths[filerPath] = struct{}{} - } + addFilerPathSpellings(keptFilerPaths, bucketName, dataPath, f) } // Use MetadataBuilder to remove snapshots and create new metadata @@ -118,11 +117,18 @@ func (h *Handler) expireSnapshots( } // Delete files exclusively referenced by expired snapshots (best-effort) + tableBase := path.Join(s3tables.TablesPath, bucketName, dataPath) + "/" deletedCount := 0 for filePath := range expiredFiles { - resolved, err := icebergFilerPath(bucketName, dataPath, filePath) + resolved, err := resolveFilerPath(ctx, filerClient, bucketName, dataPath, filePath) if err != nil { - glog.Warningf("iceberg maintenance: cannot resolve expired file %s: %v", filePath, err) + if !isFilerNotFound(err) { + glog.Warningf("iceberg maintenance: cannot resolve expired file %s: %v", filePath, err) + } + continue + } + if !strings.HasPrefix(resolved, tableBase) { + glog.Warningf("iceberg maintenance: refusing to delete %s outside table path", resolved) continue } if _, stillReferenced := keptFilerPaths[resolved]; stillReferenced { @@ -357,9 +363,11 @@ func collectOrphanCandidates( referencedFilerPaths := make(map[string]struct{}, len(referencedFiles)) for ref := range referencedFiles { - if filerPath, err := icebergFilerPath(bucketName, dataPath, ref); err == nil { - referencedFilerPaths[filerPath] = struct{}{} - } + // Writers percent-escape partition directory names (e.g. "a=1" stored + // as "a%3D1" in the manifest path), but the S3 layer decodes the key + // before it lands on the filer. Match every spelling the resolver can + // return, or a live file is counted as an orphan and deleted. + addFilerPathSpellings(referencedFilerPaths, bucketName, dataPath, ref) } tableBasePath := path.Join(s3tables.TablesPath, bucketName, dataPath)