This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 6d2913326 perf(maintenance): avoid orphan reference index for exact
matches (#2268)
6d2913326 is described below
commit 6d2913326d163ec7df6592fb162469d0d72b3082
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 18:28:07 2026 +0200
perf(maintenance): avoid orphan reference index for exact matches (#2268)
* perf(maintenance): build orphan reference index only on path misses
Signed-off-by: Minh Vu <[email protected]>
* test(maintenance): cover native path aliases and legacy parity
Signed-off-by: Minh Vu <[email protected]>
* style(maintenance): satisfy lint
Signed-off-by: Minh Vu <[email protected]>
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/maintenance/orphan_cleanup.go | 66 +++++----
.../orphan_cleanup_lazy_index_bench_test.go | 135 +++++++++++++++++++
.../maintenance/orphan_cleanup_lazy_index_test.go | 150 +++++++++++++++++++++
3 files changed, 328 insertions(+), 23 deletions(-)
diff --git a/table/maintenance/orphan_cleanup.go
b/table/maintenance/orphan_cleanup.go
index bfae0048c..b4a964b95 100644
--- a/table/maintenance/orphan_cleanup.go
+++ b/table/maintenance/orphan_cleanup.go
@@ -422,33 +422,53 @@ func planOrphanFiles(ctx context.Context, tbl
*table.Table, cfg *orphanCleanupCo
return OrphanCleanupPlan{}, err
}
- // Identify orphans.
- referencedIndex := newReferencedFileIndex(referencedFiles, cfg)
+ // Most listed paths already match their metadata references exactly.
+ // Avoid normalizing the entire reference set unless a raw lookup
misses.
+ plan, err := findOrphanFiles(scannedFiles, referencedFiles, cfg)
+ if err != nil {
+ return OrphanCleanupPlan{}, err
+ }
+ plan.cutoff = cutoff
+
+ return plan, nil
+}
+
+// findOrphanFiles defers building the URI alias index until it is needed.
+// Exact matches do not need URI parsing or prefix-mismatch handling.
+func findOrphanFiles(
+ scannedFiles []scannedFile, referencedFiles map[string]bool, cfg
*orphanCleanupConfig,
+) (OrphanCleanupPlan, error) {
+ plan := OrphanCleanupPlan{orphanFiles: make([]OrphanFile, 0)}
+ var referencedIndex referencedFileIndex
+ indexBuilt := false
- var orphanFiles []string
- orphanFileEntries := make([]OrphanFile, 0)
- var totalOrphanSize int64
for _, f := range scannedFiles {
+ if _, referenced := referencedFiles[f.path]; referenced {
+ continue
+ }
+
+ if !indexBuilt {
+ referencedIndex =
newReferencedFileIndex(referencedFiles, cfg)
+ indexBuilt = true
+ }
+
isOrphan, err := isFileOrphan(f.path, referencedFiles,
referencedIndex, cfg)
if err != nil {
return OrphanCleanupPlan{}, fmt.Errorf("failed to
identify orphan %s: %w", f.path, err)
}
- if isOrphan {
- orphanFiles = append(orphanFiles, f.path)
- orphanFileEntries = append(orphanFileEntries,
OrphanFile{
- Path: f.path,
- SizeBytes: f.size,
- })
- totalOrphanSize += f.size
+ if !isOrphan {
+ continue
}
+
+ plan.orphanFileLocations = append(plan.orphanFileLocations,
f.path)
+ plan.orphanFiles = append(plan.orphanFiles, OrphanFile{
+ Path: f.path,
+ SizeBytes: f.size,
+ })
+ plan.totalSizeBytes += f.size
}
- return OrphanCleanupPlan{
- orphanFileLocations: orphanFiles,
- orphanFiles: orphanFileEntries,
- totalSizeBytes: totalOrphanSize,
- cutoff: cutoff,
- }, nil
+ return plan, nil
}
func executeOrphanCleanup(ctx context.Context, tbl *table.Table, plan
OrphanCleanupPlan, cfg *orphanCleanupConfig) (OrphanCleanupResult, error) {
@@ -655,14 +675,14 @@ func isFileOrphan(
referencedIndex referencedFileIndex,
cfg *orphanCleanupConfig,
) (bool, error) {
- normalizedFiles := normalizedFilePathAliases(file, cfg)
- normalizedFile := normalizedFiles[0]
-
- // Any presence in referencedFiles means referenced;
- // the bool distinguishes data vs metadata for gc.enabled, not
membership"
+ // Any presence in referencedFiles means referenced. Its bool value
+ // distinguishes data and metadata, not membership.
if _, ok := referencedFiles[file]; ok {
return false, nil
}
+
+ normalizedFiles := normalizedFilePathAliases(file, cfg)
+ normalizedFile := normalizedFiles[0]
if _, ok := referencedFiles[normalizedFile]; ok {
return false, nil
}
diff --git a/table/maintenance/orphan_cleanup_lazy_index_bench_test.go
b/table/maintenance/orphan_cleanup_lazy_index_bench_test.go
new file mode 100644
index 000000000..8f70dc6a4
--- /dev/null
+++ b/table/maintenance/orphan_cleanup_lazy_index_bench_test.go
@@ -0,0 +1,135 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package maintenance
+
+import (
+ "fmt"
+ "strconv"
+ "testing"
+)
+
+// This is the pre-optimization orphan discovery path, kept in a benchmark
+// only so before/after measurements can use the same binary and inputs.
+func findOrphanFilesEagerForBenchmark(
+ scannedFiles []scannedFile, referencedFiles map[string]bool, cfg
*orphanCleanupConfig,
+) (OrphanCleanupPlan, error) {
+ referencedIndex := newReferencedFileIndex(referencedFiles, cfg)
+ plan := OrphanCleanupPlan{orphanFiles: make([]OrphanFile, 0)}
+
+ for _, f := range scannedFiles {
+ // Before: every file was normalized before the raw reference
lookup.
+ aliases := normalizedFilePathAliases(f.path, cfg)
+ normalized := aliases[0]
+ if _, ok := referencedFiles[f.path]; ok {
+ continue
+ }
+ if _, ok := referencedFiles[normalized]; ok {
+ continue
+ }
+
+ matched := false
+ for _, alias := range aliases {
+ if _, ok := referencedIndex.normalized[alias]; ok {
+ matched = true
+
+ break
+ }
+ }
+ if matched {
+ continue
+ }
+
+ for _, reference := range
referencedIndex.byPath[filePathKey(normalized)] {
+ decision, err := checkPrefixMismatch(reference, f.path,
cfg)
+ if err != nil {
+ return OrphanCleanupPlan{}, fmt.Errorf("failed
to identify orphan %s: %w", f.path, err)
+ }
+ if decision == prefixMismatchKeep {
+ matched = true
+
+ break
+ }
+ }
+ if matched {
+ continue
+ }
+
+ plan.orphanFileLocations = append(plan.orphanFileLocations,
f.path)
+ plan.orphanFiles = append(plan.orphanFiles, OrphanFile{
+ Path: f.path, SizeBytes: f.size,
+ })
+ plan.totalSizeBytes += f.size
+ }
+
+ return plan, nil
+}
+
+func BenchmarkFindOrphanFilesLazyIndex(b *testing.B) {
+ for _, count := range []int{1_000, 10_000, 100_000} {
+ for _, scenario := range []string{"all-exact",
"one-percent-orphans", "one-percent-scheme-aliases"} {
+ b.Run(fmt.Sprintf("files=%d/%s", count, scenario),
func(b *testing.B) {
+ cfg := newOrphanCleanupConfig()
+ if scenario == "one-percent-scheme-aliases" {
+ cfg =
newOrphanCleanupConfig(WithEqualSchemes(map[string]string{"s3,s3a": "s3"}))
+ }
+
+ references := make(map[string]bool, count)
+ scanned := make([]scannedFile, count)
+ wantOrphans := 0
+ for i := range scanned {
+ path :=
fmt.Sprintf("s3://bucket/warehouse/table/data/part-%06d.parquet", i)
+ if scenario == "one-percent-orphans" &&
i%100 == 0 {
+ wantOrphans++
+ } else {
+ references[path] = true
+ }
+ if scenario ==
"one-percent-scheme-aliases" && i%100 == 0 {
+ path = "s3a" + path[2:]
+ }
+ scanned[i] = scannedFile{path: path,
size: int64(i + 1)}
+ }
+
+ for _, implementation := range []struct {
+ name string
+ find func([]scannedFile,
map[string]bool, *orphanCleanupConfig) (OrphanCleanupPlan, error)
+ }{
+ {"before",
findOrphanFilesEagerForBenchmark},
+ {"after", findOrphanFiles},
+ } {
+ b.Run(implementation.name, func(b
*testing.B) {
+ b.ReportAllocs()
+ var plan OrphanCleanupPlan
+ b.ResetTimer()
+ for range b.N {
+ var err error
+ plan, err =
implementation.find(scanned, references, cfg)
+ if err != nil {
+ b.Fatal(err)
+ }
+ }
+ b.StopTimer()
+ if len(plan.orphanFiles) !=
wantOrphans {
+ b.Fatalf("got %d
orphans, want %d", len(plan.orphanFiles), wantOrphans)
+ }
+ orphanCleanupBenchmarkSink =
strconv.FormatInt(plan.totalSizeBytes, 10)
+ })
+ }
+ })
+ }
+ }
+}
diff --git a/table/maintenance/orphan_cleanup_lazy_index_test.go
b/table/maintenance/orphan_cleanup_lazy_index_test.go
new file mode 100644
index 000000000..1d902ed25
--- /dev/null
+++ b/table/maintenance/orphan_cleanup_lazy_index_test.go
@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package maintenance
+
+import (
+ "slices"
+ "strings"
+ "testing"
+)
+
+func TestFindOrphanFilesLazyIndex(t *testing.T) {
+ tests := []struct {
+ name string
+ options []OrphanCleanupOption
+ references map[string]bool
+ scanned []scannedFile
+ want []OrphanFile
+ wantError string
+ }{
+ {
+ name: "all exact references",
+ references: map[string]bool{
+ "s3://bucket/table/a.parquet": false,
+ "s3://bucket/table/b.parquet": true,
+ },
+ scanned: []scannedFile{
+ {path: "s3://bucket/table/a.parquet", size: 10},
+ {path: "s3://bucket/table/b.parquet", size: 20},
+ },
+ },
+ {
+ name: "scheme alias and real orphan",
+ options: []OrphanCleanupOption{
+ WithEqualSchemes(map[string]string{"s3,s3a":
"s3"}),
+ },
+ references: map[string]bool{
+ "s3://bucket/table/a.parquet": true,
+ "s3://bucket/table/b.parquet": true,
+ },
+ scanned: []scannedFile{
+ {path: "s3://bucket/table/a.parquet", size: 10},
+ {path: "s3a://bucket/table/b.parquet", size:
20},
+ {path: "s3://bucket/table/c.parquet", size: 30},
+ },
+ want: []OrphanFile{{Path:
"s3://bucket/table/c.parquet", SizeBytes: 30}},
+ },
+ {
+ name: "prefix mismatch ignored",
+ options: []OrphanCleanupOption{
+ WithPrefixMismatchMode(PrefixMismatchIgnore),
+ },
+ references: map[string]bool{
+ "s3://bucket/table/a.parquet": true,
+ },
+ scanned: []scannedFile{{path:
"s3://other/table/a.parquet", size: 50}},
+ },
+ {
+ name: "prefix mismatch deleted",
+ options: []OrphanCleanupOption{
+ WithPrefixMismatchMode(PrefixMismatchDelete),
+ },
+ references: map[string]bool{
+ "s3://bucket/table/a.parquet": true,
+ },
+ scanned: []scannedFile{{path:
"s3://other/table/a.parquet", size: 50}},
+ want: []OrphanFile{{Path:
"s3://other/table/a.parquet", SizeBytes: 50}},
+ },
+ {
+ name: "prefix mismatch error",
+ references: map[string]bool{
+ "s3://bucket/table/a.parquet": true,
+ },
+ scanned: []scannedFile{{path:
"s3://other/table/a.parquet", size: 50}},
+ wantError: "prefix mismatch detected",
+ },
+ {
+ name: "local file URI alias",
+ references: map[string]bool{
+ "file:///tmp/iceberg/table/a.parquet": true,
+ },
+ scanned: []scannedFile{{path:
"/tmp/iceberg/table/a.parquet", size: 5}},
+ },
+ {
+ name: "Windows drive case alias",
+ references: map[string]bool{
+ "C:\\Warehouse\\Iceberg\\A.parquet": true,
+ },
+ scanned: []scannedFile{{path:
"c:/warehouse/iceberg/a.parquet", size: 5}},
+ },
+ {
+ name: "no references",
+ scanned: []scannedFile{{path:
"s3://bucket/table/new.parquet", size: 75}},
+ want: []OrphanFile{{Path:
"s3://bucket/table/new.parquet", SizeBytes: 75}},
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ cfg := newOrphanCleanupConfig(tt.options...)
+ got, err := findOrphanFiles(tt.scanned, tt.references,
cfg)
+ before, beforeErr :=
findOrphanFilesEagerForBenchmark(tt.scanned, tt.references, cfg)
+ if (err == nil) != (beforeErr == nil) {
+ t.Fatalf("optimized error = %v, legacy error =
%v", err, beforeErr)
+ }
+ if tt.wantError != "" {
+ if err == nil || !strings.Contains(err.Error(),
tt.wantError) {
+ t.Fatalf("expected error containing %q,
got %v", tt.wantError, err)
+ }
+
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !slices.Equal(got.OrphanFiles(), tt.want) {
+ t.Fatalf("orphan files = %v, want %v",
got.OrphanFiles(), tt.want)
+ }
+ if !slices.Equal(got.OrphanFiles(),
before.OrphanFiles()) {
+ t.Fatalf("orphan files = %v, legacy = %v",
got.OrphanFiles(), before.OrphanFiles())
+ }
+ var wantSize int64
+ wantPaths := make([]string, len(tt.want))
+ for i, file := range tt.want {
+ wantPaths[i] = file.Path
+ wantSize += file.SizeBytes
+ }
+ if !slices.Equal(got.Files(), wantPaths) {
+ t.Fatalf("orphan paths = %v, want %v",
got.Files(), wantPaths)
+ }
+ if got.TotalSizeBytes() != wantSize {
+ t.Fatalf("total size = %d, want %d",
got.TotalSizeBytes(), wantSize)
+ }
+ })
+ }
+}
