mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-24 15:00:43 +02:00
iceberg maintenance: match percent-decoded keys when checking file references (#11978)
* 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>
This commit is contained in:
1 parent
8541f959a9
commit
0d00e92178
3 files changed
+509
-25
No files matched your search
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user