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 346019c17 perf(maintenance): bound orphan reference buffers (#2379)
346019c17 is described below
commit 346019c1795621675a6c04d77eea2a8e794f7028
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 18:41:48 2026 +0100
perf(maintenance): bound orphan reference buffers (#2379)
* perf(maintenance): bound orphan reference buffers
Signed-off-by: Minh Vu <[email protected]>
* test(maintenance): consolidate orphan reference tests
Move batching tests into existing cleanup tests and keep the
production-path benchmark next to other cleanup benchmarks.
Signed-off-by: Minh Vu <[email protected]>
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/maintenance/orphan_cleanup.go | 51 +++++++++++++++++---------
table/maintenance/orphan_cleanup_bench_test.go | 45 +++++++++++++++++++++++
table/maintenance/orphan_cleanup_test.go | 50 +++++++++++++++++++++++++
3 files changed, 128 insertions(+), 18 deletions(-)
diff --git a/table/maintenance/orphan_cleanup.go
b/table/maintenance/orphan_cleanup.go
index b170eae98..d91f91898 100644
--- a/table/maintenance/orphan_cleanup.go
+++ b/table/maintenance/orphan_cleanup.go
@@ -585,16 +585,12 @@ func getReferencedFiles(ctx context.Context, tbl
*table.Table, fs io.IO, maxConc
}
// Pass 2: Read entries from each unique manifest in parallel.
- type refEntry struct {
- path string
- isData bool
- }
var mu sync.Mutex
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(max(min(maxConcurrency, len(uniqueManifests)), 1))
for _, m := range uniqueManifests {
g.Go(func() error {
- var entries []refEntry
+ batch := referencePathBatch{referenced: referenced, mu:
&mu}
// discardDeleted=true: skip DELETED-status entries when
// computing the reachable file set. A DELETED entry is
// not live in this snapshot and should not pin the file
@@ -610,26 +606,16 @@ func getReferencedFiles(ctx context.Context, tbl
*table.Table, fs io.IO, maxConc
}
// All files tracked within a manifest (data
files, equality deletes, position deletes)
// are considered "data files" for the purposes
of gc.enabled.
- entries = append(entries, refEntry{
- path: entry.DataFile().FilePath(),
- isData: true,
- })
+ batch.add(entry.DataFile().FilePath())
if ref :=
iceberginternal.BorrowedDataFileReferencedDataFile(entry.DataFile()); ref !=
nil {
// This is a deletion vector entry
referencing a data file.
// Its FilePath() is the deletion
vector (.dv) file itself (added above).
// We must also mark the referenced
data file as referenced.
- entries = append(entries, refEntry{
- path: *ref,
- isData: true,
- })
+ batch.add(*ref)
}
}
- mu.Lock()
- for _, e := range entries {
- referenced[e.path] = e.isData
- }
- mu.Unlock()
+ batch.flush()
return nil
})
@@ -642,6 +628,35 @@ func getReferencedFiles(ctx context.Context, tbl
*table.Table, fs io.IO, maxConc
return referenced, nil
}
+// Bound the temporary path buffer for each concurrently read manifest.
+const referencePathBatchSize = 512
+
+type referencePathBatch struct {
+ paths []string
+ referenced map[string]bool
+ mu *sync.Mutex
+}
+
+func (b *referencePathBatch) add(path string) {
+ b.paths = append(b.paths, path)
+ if len(b.paths) == referencePathBatchSize {
+ b.flush()
+ }
+}
+
+func (b *referencePathBatch) flush() {
+ if len(b.paths) == 0 {
+ return
+ }
+ b.mu.Lock()
+ for _, path := range b.paths {
+ b.referenced[path] = true
+ }
+ b.mu.Unlock()
+ clear(b.paths)
+ b.paths = b.paths[:0]
+}
+
func walkDirectory(ctx context.Context, fsys io.IO, root string, fn func(path
string, info stdfs.FileInfo) error) error {
if listable, ok := fsys.(io.ListableIO); ok {
return listable.WalkDir(ctx, root, func(path string, d
stdfs.DirEntry, err error) error {
diff --git a/table/maintenance/orphan_cleanup_bench_test.go
b/table/maintenance/orphan_cleanup_bench_test.go
index 3ccd4ac31..143c484d7 100644
--- a/table/maintenance/orphan_cleanup_bench_test.go
+++ b/table/maintenance/orphan_cleanup_bench_test.go
@@ -18,14 +18,17 @@
package maintenance
import (
+ "bytes"
"context"
"fmt"
+ "strconv"
"strings"
"testing"
"time"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
)
var orphanCleanupBenchmarkSink string
@@ -70,6 +73,48 @@ func BenchmarkApplyURIEquivalence(b *testing.B) {
}
}
+func referenceManifestFixture(t testing.TB, n int) (*table.Table, *trackingIO)
{
+ t.Helper()
+ fs := newTrackingIO()
+ schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "x",
Type: iceberg.PrimitiveTypes.Int64, Required: true})
+ spec := iceberg.NewPartitionSpec()
+ snapshotID := int64(1)
+ entries := make([]iceberg.ManifestEntry, n)
+ for i := range entries {
+ df, err := iceberg.NewDataFileBuilder(spec,
iceberg.EntryContentData, fmt.Sprintf("s3://bucket/data/%d.parquet", i),
iceberg.ParquetFile, nil, nil, nil, 1, 1024)
+ require.NoError(t, err)
+ entries[i] =
iceberg.NewManifestEntryBuilder(iceberg.EntryStatusADDED, &snapshotID,
df.Build()).DataSequenceNumber(1).Build()
+ }
+ manifestPath := "s3://bucket/meta/manifest.avro"
+ listPath := "s3://bucket/meta/snap.avro"
+ var buf bytes.Buffer
+ _, err := iceberg.WriteManifest(manifestPath, &buf, 2, spec, schema,
snapshotID, entries)
+ require.NoError(t, err)
+ require.NoError(t, fs.WriteFile(context.Background(), manifestPath,
buf.Bytes()))
+ mf := iceberg.NewManifestBuilder(2, manifestPath, int64(buf.Len()), 0,
snapshotID).SequenceNumber(1,
1).AddedFiles(int32(n)).AddedRows(int64(n)).Build()
+ writeManifestList(t, fs, snapshotID, listPath,
[]iceberg.ManifestFile{mf})
+ meta, err :=
table.ParseMetadataString(buildMetaJSON(metaJSONOpts{snapshots:
fmt.Sprintf(`{"snapshot-id":1,"timestamp-ms":1000,"manifest-list":%q}`,
listPath)}))
+ require.NoError(t, err)
+
+ return table.New(table.Identifier{"ns", "tbl"}, meta, "metadata.json",
testFSF(fs), nil), fs
+}
+
+func BenchmarkGetReferencedFilesManifestEntries(b *testing.B) {
+ for _, n := range []int{1, 16384} {
+ b.Run(strconv.Itoa(n), func(b *testing.B) {
+ tbl, fs := referenceManifestFixture(b, n)
+ b.ReportAllocs()
+ for b.Loop() {
+ refs, err :=
getReferencedFiles(context.Background(), tbl, fs, 1, true)
+ // Four non-data references: metadata.json,
version hint, manifest list, and manifest.
+ if err != nil || len(refs) != n+4 {
+ b.Fatalf("references=%d error=%v",
len(refs), err)
+ }
+ }
+ })
+ }
+}
+
func BenchmarkGetReferencedFilesManifestLists(b *testing.B) {
for _, snapshotCount := range []int{1, 16, 128, 1024} {
for _, maxWorkers := range []int{1, 4, 16} {
diff --git a/table/maintenance/orphan_cleanup_test.go
b/table/maintenance/orphan_cleanup_test.go
index e1747ba25..1a88a6374 100644
--- a/table/maintenance/orphan_cleanup_test.go
+++ b/table/maintenance/orphan_cleanup_test.go
@@ -25,6 +25,7 @@ import (
"os"
"path/filepath"
"runtime"
+ "strconv"
"strings"
"sync"
"sync/atomic"
@@ -2606,3 +2607,52 @@ func
TestGetReferencedFilesFetchesManifestListsConcurrently(t *testing.T) {
assert.Greater(t, maxOpen, 1, "manifest lists should be fetched
concurrently")
assert.LessOrEqual(t, maxOpen, maxWorkers, "manifest-list reads must
respect the configured limit")
}
+
+func TestReferencePathBatch(t *testing.T) {
+ for _, n := range []int{0, 1, 511, 512, 513, 1024, 1031} {
+ t.Run(strconv.Itoa(n), func(t *testing.T) {
+ refs := map[string]bool{"existing": false}
+ var mu sync.Mutex
+ batch := referencePathBatch{referenced: refs, mu: &mu}
+ for i := range n {
+ batch.add(strconv.Itoa(i))
+ }
+ batch.flush()
+ batch.flush()
+ require.Len(t, refs, n+1)
+ require.False(t, refs["existing"])
+ for i := range n {
+ require.True(t, refs[strconv.Itoa(i)])
+ }
+ require.Empty(t, batch.paths)
+ })
+ }
+}
+
+func TestReferencePathBatchConcurrent(t *testing.T) {
+ const workers = 4
+ const perWorker = referencePathBatchSize + 1
+ refs := make(map[string]bool)
+ var mu sync.Mutex
+ var wg sync.WaitGroup
+ for worker := range workers {
+ wg.Go(func() {
+ batch := referencePathBatch{referenced: refs, mu: &mu}
+ for i := range perWorker {
+ batch.add(fmt.Sprintf("%d-%d", worker, i))
+ }
+ batch.flush()
+ })
+ }
+ wg.Wait()
+ require.Len(t, refs, workers*perWorker)
+}
+
+func TestGetReferencedFilesAcrossBatchBoundary(t *testing.T) {
+ tbl, fs := referenceManifestFixture(t, 1025)
+ refs, err := getReferencedFiles(context.Background(), tbl, fs, 2, true)
+ require.NoError(t, err)
+ for i := range 1025 {
+ require.True(t, refs[fmt.Sprintf("s3://bucket/data/%d.parquet",
i)])
+ }
+}
