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 a5a06651b perf(table): prefetch historical snapshot manifest lists
(#2234)
a5a06651b is described below
commit a5a06651b76d8dfe378b8f9260d84c3d3358cd50
Author: Minh Vu <[email protected]>
AuthorDate: Tue Oct 20 19:29:08 2026 +0200
perf(table): prefetch historical snapshot manifest lists (#2234)
* perf(table): prefetch historical snapshot manifest lists
* perf(table): keep warm manifest scans synchronous
---
table/inspect_files.go | 24 +--
table/inspect_snapshot_prefetch.go | 225 ++++++++++++++++++++
table/inspect_snapshot_prefetch_test.go | 360 ++++++++++++++++++++++++++++++++
table/snapshot_manifest_cache.go | 19 ++
4 files changed, 605 insertions(+), 23 deletions(-)
diff --git a/table/inspect_files.go b/table/inspect_files.go
index 4352d34d0..42b7915fc 100644
--- a/table/inspect_files.go
+++ b/table/inspect_files.go
@@ -172,29 +172,7 @@ func (i InspectTable) allManifestEntryReader(
return i.manifestEntryReaderFromManifestSource(
ctx, arrowSchema, fs, discardDeleted, includeManifest,
appendEntry,
func(yield func(iceberg.ManifestFile, error) bool) {
- seen := make(map[string]struct{})
- for _, snapshot := range snapshots {
- if err := ctx.Err(); err != nil {
- yield(nil, err)
-
- return
- }
- manifests, err := provider.Manifests(ctx,
snapshot)
- if err != nil {
- yield(nil, fmt.Errorf("read snapshot %d
manifests: %w", snapshot.SnapshotID, err))
-
- return
- }
- for manifest := range manifests {
- if _, ok := seen[manifest.FilePath()];
ok {
- continue
- }
- seen[manifest.FilePath()] = struct{}{}
- if !yield(manifest, nil) {
- return
- }
- }
- }
+ streamSnapshotManifests(ctx, snapshots,
provider.cachedSet, provider.set, yield)
}), nil
}
diff --git a/table/inspect_snapshot_prefetch.go
b/table/inspect_snapshot_prefetch.go
new file mode 100644
index 000000000..10a2b8378
--- /dev/null
+++ b/table/inspect_snapshot_prefetch.go
@@ -0,0 +1,225 @@
+// 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 table
+
+import (
+ "context"
+ "fmt"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+)
+
+const inspectSnapshotManifestPrefetch = 8
+
+type inspectSnapshotManifestRead struct {
+ set snapshotManifestSet
+ err error
+}
+
+// streamSnapshotManifests loads historical manifest lists concurrently while
+// yielding them in snapshot order. Only a small window of lists can be held
+// by in-flight reads, even for tables with many snapshots.
+func streamSnapshotManifests(
+ ctx context.Context,
+ snapshots []Snapshot,
+ cached func(Snapshot) (snapshotManifestSet, bool),
+ load func(context.Context, Snapshot) (snapshotManifestSet, error),
+ yield func(iceberg.ManifestFile, error) bool,
+) {
+ if len(snapshots) == 0 {
+ return
+ }
+ if err := ctx.Err(); err != nil {
+ yield(nil, err)
+
+ return
+ }
+ seen := make(map[string]struct{})
+ if len(snapshots) == 1 {
+ set, ok := snapshotManifestSet{}, false
+ if cached != nil {
+ set, ok = cached(snapshots[0])
+ }
+ if !ok {
+ var err error
+ set, err = load(ctx, snapshots[0])
+ if err != nil {
+ yield(nil, fmt.Errorf("read snapshot %d
manifests: %w", snapshots[0].SnapshotID, err))
+
+ return
+ }
+ }
+ for _, manifest := range set.borrowAllManifests() {
+ path := manifest.FilePath()
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+ if !yield(manifest, nil) {
+ return
+ }
+ }
+
+ return
+ }
+ if cached != nil {
+ start := 0
+ for start < len(snapshots) {
+ count := min(inspectSnapshotManifestPrefetch,
len(snapshots)-start)
+ var sets
[inspectSnapshotManifestPrefetch]snapshotManifestSet
+ allCached := true
+ for i := range count {
+ set, ok := cached(snapshots[start+i])
+ if !ok {
+ allCached = false
+
+ break
+ }
+ sets[i] = set
+ }
+ if !allCached {
+ break
+ }
+
+ for i := range count {
+ if err := ctx.Err(); err != nil {
+ yield(nil, err)
+
+ return
+ }
+ for _, manifest := range
sets[i].borrowAllManifests() {
+ path := manifest.FilePath()
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+ if !yield(manifest, nil) {
+ return
+ }
+ }
+ }
+ start += count
+ }
+ if start == len(snapshots) {
+ return
+ }
+ // The first cache miss starts the concurrent path. Recheck
this small
+ // window there so cached and newly loaded results share one
ordered path.
+ snapshots = snapshots[start:]
+ }
+
+ readCtx := ctx
+ var cancel context.CancelFunc
+ var workers sync.WaitGroup
+ // Cancel first, then wait for all readers before returning. This also
+ // handles a consumer that stops iterating partway through a snapshot.
+ defer func() {
+ if cancel != nil {
+ cancel()
+ }
+ workers.Wait()
+ }()
+
+ window := min(len(snapshots), inspectSnapshotManifestPrefetch)
+ results := make([]chan inspectSnapshotManifestRead, window)
+ ready := make([]bool, window)
+ readyResults := make([]inspectSnapshotManifestRead, window)
+
+ startRead := func(index int) {
+ if cancel == nil {
+ readCtx, cancel = context.WithCancel(ctx)
+ }
+ slot := index % window
+ if results[slot] == nil {
+ // Buffer one result so a cancelled worker never blocks
sending it.
+ results[slot] = make(chan inspectSnapshotManifestRead,
1)
+ }
+ ready[slot] = false
+ workers.Go(func() {
+ set, err := load(readCtx, snapshots[index])
+ results[slot] <- inspectSnapshotManifestRead{set: set,
err: err}
+ })
+ }
+
+ for i := range window {
+ if cached != nil {
+ if set, ok := cached(snapshots[i]); ok {
+ ready[i] = true
+ readyResults[i] =
inspectSnapshotManifestRead{set: set}
+
+ continue
+ }
+ }
+ startRead(i)
+ }
+
+ for i, snapshot := range snapshots {
+ if err := readCtx.Err(); err != nil {
+ yield(nil, err)
+
+ return
+ }
+
+ slot := i % window
+ var result inspectSnapshotManifestRead
+ if ready[slot] {
+ result = readyResults[slot]
+ ready[slot] = false
+ } else {
+ select {
+ case <-readCtx.Done():
+ yield(nil, readCtx.Err())
+
+ return
+ case result = <-results[slot]:
+ }
+ }
+
+ if result.err != nil {
+ yield(nil, fmt.Errorf("read snapshot %d manifests: %w",
snapshot.SnapshotID, result.err))
+
+ return
+ }
+
+ // Reuse the completed slot before yielding. The consumer may
spend a
+ // long time decoding these manifests while the next list is
fetched.
+ if next := i + window; next < len(snapshots) {
+ if cached != nil {
+ if set, ok := cached(snapshots[next]); ok {
+ ready[slot] = true
+ readyResults[slot] =
inspectSnapshotManifestRead{set: set}
+ } else {
+ startRead(next)
+ }
+ } else {
+ startRead(next)
+ }
+ }
+
+ for _, manifest := range result.set.borrowAllManifests() {
+ path := manifest.FilePath()
+ if _, ok := seen[path]; ok {
+ continue
+ }
+ seen[path] = struct{}{}
+ if !yield(manifest, nil) {
+ return
+ }
+ }
+ }
+}
diff --git a/table/inspect_snapshot_prefetch_test.go
b/table/inspect_snapshot_prefetch_test.go
new file mode 100644
index 000000000..990a60493
--- /dev/null
+++ b/table/inspect_snapshot_prefetch_test.go
@@ -0,0 +1,360 @@
+// 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 table
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "slices"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/iceberg-go"
+ "github.com/stretchr/testify/require"
+)
+
+func TestInspectSnapshotManifestPrefetchPreservesOrderAndBoundsReads(t
*testing.T) {
+ snapshots := make([]Snapshot, 25)
+ for i := range snapshots {
+ snapshots[i].SnapshotID = int64(i + 1)
+ }
+
+ ctx := t.Context()
+ started := make(chan int64, inspectSnapshotManifestPrefetch)
+ release := make(chan struct{})
+ defer func() {
+ select {
+ case <-release:
+ default:
+ close(release)
+ }
+ }()
+
+ var active, maxActive atomic.Int32
+ resultCh := make(chan []string, 1)
+ go func() {
+ var paths []string
+ streamSnapshotManifests(ctx, snapshots, nil,
+ func(ctx context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ current := active.Add(1)
+ defer active.Add(-1)
+ for {
+ previous := maxActive.Load()
+ if current <= previous ||
maxActive.CompareAndSwap(previous, current) {
+ break
+ }
+ }
+
+ if snapshot.SnapshotID <=
inspectSnapshotManifestPrefetch {
+ started <- snapshot.SnapshotID
+ }
+ select {
+ case <-release:
+ case <-ctx.Done():
+ return snapshotManifestSet{}, ctx.Err()
+ }
+
+ path := fmt.Sprintf("manifest-%d.avro",
snapshot.SnapshotID)
+ manifest := iceberg.NewManifestFile(2, path,
10, 0, 1).Build()
+
+ return snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}, nil
+ },
+ func(manifest iceberg.ManifestFile, err error) bool {
+ if err != nil {
+ paths = append(paths, "ERROR:
"+err.Error())
+
+ return false
+ }
+ paths = append(paths, manifest.FilePath())
+
+ return true
+ })
+ resultCh <- paths
+ }()
+
+ var firstWindow []int64
+ for range inspectSnapshotManifestPrefetch {
+ select {
+ case id := <-started:
+ firstWindow = append(firstWindow, id)
+ case <-time.After(5 * time.Second):
+ t.Fatal("initial manifest reads did not start
concurrently")
+ }
+ }
+ slices.Sort(firstWindow)
+ for i, id := range firstWindow {
+ require.Equal(t, int64(i+1), id)
+ }
+ require.EqualValues(t, inspectSnapshotManifestPrefetch,
maxActive.Load())
+ close(release)
+
+ select {
+ case paths := <-resultCh:
+ require.Len(t, paths, len(snapshots))
+ for i, path := range paths {
+ require.Equal(t, fmt.Sprintf("manifest-%d.avro", i+1),
path)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("manifest prefetch failed to finish")
+ }
+ require.EqualValues(t, 0, active.Load())
+ require.LessOrEqual(t, maxActive.Load(),
int32(inspectSnapshotManifestPrefetch))
+}
+
+func TestInspectSnapshotManifestPrefetchUsesCachedSets(t *testing.T) {
+ snapshots := make([]Snapshot, 12)
+ cachedSets := make(map[int64]snapshotManifestSet)
+ for i := range snapshots {
+ snapshots[i].SnapshotID = int64(i + 1)
+ if i%2 == 0 {
+ path := fmt.Sprintf("manifest-%d.avro", i+1)
+ manifest := iceberg.NewManifestFile(2, path, 10, 0,
1).Build()
+ cachedSets[int64(i+1)] = snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}
+ }
+ }
+
+ var loaded atomic.Int32
+ var paths []string
+ streamSnapshotManifests(context.Background(), snapshots,
+ func(snapshot Snapshot) (snapshotManifestSet, bool) {
+ set, ok := cachedSets[snapshot.SnapshotID]
+
+ return set, ok
+ },
+ func(_ context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ loaded.Add(1)
+ path := fmt.Sprintf("manifest-%d.avro",
snapshot.SnapshotID)
+ manifest := iceberg.NewManifestFile(2, path, 10, 0,
1).Build()
+
+ return snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}, nil
+ },
+ func(manifest iceberg.ManifestFile, err error) bool {
+ require.NoError(t, err)
+ paths = append(paths, manifest.FilePath())
+
+ return true
+ })
+
+ require.EqualValues(t, len(snapshots)/2, loaded.Load())
+ require.Len(t, paths, len(snapshots))
+ for i, path := range paths {
+ require.Equal(t, fmt.Sprintf("manifest-%d.avro", i+1), path)
+ }
+}
+
+func TestInspectSnapshotManifestPrefetchDeduplicatesPaths(t *testing.T) {
+ snapshots := []Snapshot{{SnapshotID: 1}, {SnapshotID: 2}, {SnapshotID:
3}}
+ manifests := [][]string{{"a.avro", "shared.avro"}, {"shared.avro",
"b.avro"}, {"a.avro", "c.avro"}}
+ var paths []string
+ streamSnapshotManifests(context.Background(), snapshots, nil,
+ func(_ context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ set := snapshotManifestSet{}
+ for _, path := range manifests[snapshot.SnapshotID-1] {
+ set.all = append(set.all,
iceberg.NewManifestFile(2, path, 10, 0, 1).Build())
+ }
+
+ return set, nil
+ },
+ func(manifest iceberg.ManifestFile, err error) bool {
+ require.NoError(t, err)
+ paths = append(paths, manifest.FilePath())
+
+ return true
+ })
+
+ require.Equal(t, []string{"a.avro", "shared.avro", "b.avro", "c.avro"},
paths)
+}
+
+func TestInspectSnapshotManifestPrefetchStopsReadersOnEarlyExit(t *testing.T) {
+ snapshots := make([]Snapshot, 30)
+ for i := range snapshots {
+ snapshots[i].SnapshotID = int64(i + 1)
+ }
+
+ var active atomic.Int32
+ var yielded int
+ streamSnapshotManifests(context.Background(), snapshots, nil,
+ func(ctx context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ active.Add(1)
+ defer active.Add(-1)
+
+ if snapshot.SnapshotID == 1 {
+ manifest := iceberg.NewManifestFile(2,
"first.avro", 10, 0, 1).Build()
+
+ return snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}, nil
+ }
+ <-ctx.Done()
+
+ return snapshotManifestSet{}, ctx.Err()
+ },
+ func(_ iceberg.ManifestFile, _ error) bool {
+ yielded++
+
+ return false
+ })
+ require.Equal(t, 1, yielded)
+ require.EqualValues(t, 0, active.Load(), "all workers must finish
before return")
+}
+
+func TestInspectSnapshotManifestPrefetchReportsFirstOrderedError(t *testing.T)
{
+ failure := errors.New("manifest load failed")
+ snapshots := []Snapshot{{SnapshotID: 1}, {SnapshotID: 2}, {SnapshotID:
3}}
+ var paths []string
+ var gotErr error
+
+ streamSnapshotManifests(context.Background(), snapshots, nil,
+ func(_ context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ if snapshot.SnapshotID == 2 {
+ return snapshotManifestSet{}, failure
+ }
+ path := fmt.Sprintf("manifest-%d.avro",
snapshot.SnapshotID)
+ manifest := iceberg.NewManifestFile(2, path, 10, 0,
1).Build()
+
+ return snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}, nil
+ },
+ func(manifest iceberg.ManifestFile, err error) bool {
+ if err != nil {
+ gotErr = err
+
+ return false
+ }
+ paths = append(paths, manifest.FilePath())
+
+ return true
+ })
+
+ require.Equal(t, []string{"manifest-1.avro"}, paths)
+ require.ErrorIs(t, gotErr, failure)
+ require.ErrorContains(t, gotErr, "snapshot 2")
+}
+
+func BenchmarkInspectSnapshotManifestListPrefetch(b *testing.B) {
+ const snapshotCount = 64
+ snapshots := make([]Snapshot, snapshotCount)
+ for i := range snapshots {
+ snapshots[i].SnapshotID = int64(i + 1)
+ }
+
+ // Model an object-store manifest-list read with 1 ms of latency. This
+ // measures the I/O overlap, not end-to-end metadata-table throughput.
+ load := func(ctx context.Context, _ Snapshot) (snapshotManifestSet,
error) {
+ select {
+ case <-time.After(time.Millisecond):
+ return snapshotManifestSet{}, nil
+ case <-ctx.Done():
+ return snapshotManifestSet{}, ctx.Err()
+ }
+ }
+
+ b.Run("sequential", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ReportMetric(snapshotCount, "lists/op")
+ for b.Loop() {
+ for _, snapshot := range snapshots {
+ if _, err := load(b.Context(), snapshot); err
!= nil {
+ b.Fatal(err)
+ }
+ }
+ }
+ })
+ b.Run("prefetch8", func(b *testing.B) {
+ b.ReportAllocs()
+ b.ReportMetric(snapshotCount, "lists/op")
+ for b.Loop() {
+ streamSnapshotManifests(b.Context(), snapshots, nil,
load,
+ func(_ iceberg.ManifestFile, err error) bool {
+ if err != nil {
+ b.Fatal(err)
+ }
+
+ return true
+ })
+ }
+ })
+}
+
+func BenchmarkInspectSnapshotManifestListCacheHit(b *testing.B) {
+ cache := newSnapshotManifestCache()
+
+ for _, count := range []int{1, 64} {
+ b.Run(fmt.Sprintf("snapshots=%d", count), func(b *testing.B) {
+ snapshots := make([]Snapshot, count)
+ sets := make([]snapshotManifestSet, count)
+ for i := range snapshots {
+ snapshots[i].SnapshotID = int64(i + 1)
+ manifest := iceberg.NewManifestFile(2,
fmt.Sprintf("manifest-%d.avro", i+1), 10, 0, 1).Build()
+ sets[i] = snapshotManifestSet{all:
[]iceberg.ManifestFile{manifest}}
+ _, err := cache.get(context.Background(),
snapshots[i], func(context.Context) (snapshotManifestSet, error) {
+ return sets[i], nil
+ })
+ if err != nil {
+ b.Fatal(err)
+ }
+ }
+
+ load := func(_ context.Context, snapshot Snapshot)
(snapshotManifestSet, error) {
+ return sets[snapshot.SnapshotID-1], nil
+ }
+ b.Run("sequential", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ var yielded int
+ seen := make(map[string]struct{})
+ for _, snapshot := range snapshots {
+ value, err :=
cache.get(b.Context(), snapshot, func(context.Context) (snapshotManifestSet,
error) {
+ return
sets[snapshot.SnapshotID-1], nil
+ })
+ if err != nil {
+ b.Fatal(err)
+ }
+ for _, manifest := range
value.borrowAllManifests() {
+ path :=
manifest.FilePath()
+ if _, ok := seen[path];
ok {
+ continue
+ }
+ seen[path] = struct{}{}
+ yielded++
+ }
+ }
+ if yielded != count {
+ b.Fatal(yielded)
+ }
+ }
+ })
+ b.Run("prefetch", func(b *testing.B) {
+ b.ReportAllocs()
+ for b.Loop() {
+ var yielded int
+ streamSnapshotManifests(b.Context(),
snapshots, cache.tryGetComplete, load,
+ func(_ iceberg.ManifestFile,
err error) bool {
+ if err != nil {
+ b.Fatal(err)
+ }
+ yielded++
+
+ return true
+ })
+ if yielded != count {
+ b.Fatal(yielded)
+ }
+ }
+ })
+ })
+ }
+}
diff --git a/table/snapshot_manifest_cache.go b/table/snapshot_manifest_cache.go
index 8f8e56611..8b63f9736 100644
--- a/table/snapshot_manifest_cache.go
+++ b/table/snapshot_manifest_cache.go
@@ -218,6 +218,21 @@ func (c *snapshotManifestCache) get(
}
}
+// tryGetComplete returns a completed manifest list without starting or joining
+// a read. Inspect can use this to keep warm-cache traversal synchronous while
+// still prefetching cache misses concurrently.
+func (c *snapshotManifestCache) tryGetComplete(snapshot Snapshot)
(snapshotManifestSet, bool) {
+ if c == nil {
+ return snapshotManifestSet{}, false
+ }
+
+ c.mu.Lock()
+ manifests, ok := c.complete.Get(snapshotManifestCacheKeyFor(snapshot))
+ c.mu.Unlock()
+
+ return manifests, ok
+}
+
func (c *snapshotManifestCache) load(
ctx context.Context,
snapshot Snapshot,
@@ -352,6 +367,10 @@ func (p ManifestProvider) set(ctx context.Context,
snapshot Snapshot) (snapshotM
})
}
+func (p ManifestProvider) cachedSet(snapshot Snapshot) (snapshotManifestSet,
bool) {
+ return p.cache.tryGetComplete(snapshot)
+}
+
func sharedSnapshotManifestFSF(fsF FSysF) FSysF {
var (
once sync.Once
