zeroshade commented on code in PR #1970: URL: https://github.com/apache/iceberg-go/pull/1970#discussion_r3937951353
########## table/snapshot_manifest_cache.go: ########## @@ -0,0 +1,286 @@ +// 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" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + lru "github.com/hashicorp/golang-lru/v2" +) + +// snapshotManifestCacheSize bounds the number of decoded manifest lists held +// by a table. A table normally reuses its current snapshot, while a bounded +// history is enough to retain useful locality for repeated historical scans. +const snapshotManifestCacheSize = 64 + +// snapshotManifestCacheManifestLimit bounds the total number of manifest +// descriptors retained by the cache. This keeps a small number of unusually +// large snapshots from consuming an unbounded amount of memory. +const snapshotManifestCacheManifestLimit = 32 * 1024 + +// snapshotManifestSet keeps the complete manifest list and data partition +// together. Manifest descriptors are immutable for an Iceberg snapshot, so a +// table can safely share this decoded result across scans. +type snapshotManifestSet struct { + all []iceberg.ManifestFile + data []iceberg.ManifestFile +} + +func newSnapshotManifestSet(manifests []iceberg.ManifestFile) snapshotManifestSet { + set := snapshotManifestSet{all: manifests} + dataCount := 0 + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataCount++ + } + } + if dataCount == 0 { + return set + } + + set.data = make([]iceberg.ManifestFile, 0, dataCount) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + set.data = append(set.data, manifest) + } + } + + return set +} + +func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile { + return slices.Clone(s.all) +} + +func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile { + return slices.Clone(s.data) +} + +func snapshotManifestSetSize(set snapshotManifestSet) int { + return len(set.all) + len(set.data) +} + +type snapshotManifestCacheKey struct { + snapshotID int64 + manifestList string + hasEmbeddedSources bool + embeddedManifestSources string +} + +func snapshotManifestCacheKeyFor(snapshot Snapshot) snapshotManifestCacheKey { + var ( + hasEmbeddedSources bool + embeddedManifestSources string + ) + if snapshot.ManifestList == "" { + hasEmbeddedSources = snapshot.ManifestLocations != nil + // Valid file locations cannot contain a literal NUL, so this join is + // collision-free while preserving nil versus empty locations. + embeddedManifestSources = strings.Join(snapshot.ManifestLocations, "\x00") + } + + return snapshotManifestCacheKey{ + snapshotID: snapshot.SnapshotID, + manifestList: snapshot.ManifestList, + hasEmbeddedSources: hasEmbeddedSources, + embeddedManifestSources: embeddedManifestSources, + } +} + +type snapshotManifestCacheEntry struct { + ready chan struct{} + readyOnce sync.Once + manifests snapshotManifestSet + err error +} + +type snapshotManifestLoader func(context.Context) (snapshotManifestSet, error) + +// snapshotManifestCache memoizes successful manifest-list reads and shares an +// in-flight read with concurrent scans. Completed reads use a bounded LRU so a +// historical scan cannot retain every snapshot for the lifetime of a table. +// Failed reads are removed so a transient object-store error does not poison +// the cache. +type snapshotManifestCache struct { + mu sync.Mutex + entries map[snapshotManifestCacheKey]*snapshotManifestCacheEntry + complete *lru.Cache[snapshotManifestCacheKey, snapshotManifestSet] + completeManifestCount int +} + +func newSnapshotManifestCache() *snapshotManifestCache { + cache := &snapshotManifestCache{ + entries: make(map[snapshotManifestCacheKey]*snapshotManifestCacheEntry), + } + complete, err := lru.NewWithEvict( + snapshotManifestCacheSize, + func(_ snapshotManifestCacheKey, value snapshotManifestSet) { + cache.completeManifestCount -= snapshotManifestSetSize(value) + }, + ) + if err != nil { + panic(err) + } + cache.complete = complete + + return cache +} + +func (c *snapshotManifestCache) get( + ctx context.Context, + snapshot Snapshot, + load snapshotManifestLoader, +) (snapshotManifestSet, error) { + if c == nil { + return load(ctx) + } + if err := ctx.Err(); err != nil { + return snapshotManifestSet{}, err + } + + key := snapshotManifestCacheKeyFor(snapshot) + c.mu.Lock() + if manifests, ok := c.complete.Get(key); ok { + c.mu.Unlock() + + return manifests, nil + } + if entry, ok := c.entries[key]; ok { + c.mu.Unlock() + + select { + case <-entry.ready: + return entry.manifests, entry.err + default: + select { + case <-entry.ready: + return entry.manifests, entry.err + case <-ctx.Done(): + return snapshotManifestSet{}, ctx.Err() + } + } + } + + entry := &snapshotManifestCacheEntry{ready: make(chan struct{})} + c.entries[key] = entry + c.mu.Unlock() + + var ( + value snapshotManifestSet + err error + ) + defer func() { + if recovered := recover(); recovered != nil { + c.finish(key, entry, snapshotManifestSet{}, fmt.Errorf( + "panic while reading snapshot %d manifest list: %v", snapshot.SnapshotID, recovered)) + panic(recovered) + } + + c.finish(key, entry, value, err) + }() + + // The read is shared with callers whose contexts may outlive this one. Do + // not let the producer's cancellation abort a healthy waiter's read. + value, err = load(context.WithoutCancel(ctx)) + + return value, err +} + +func (c *snapshotManifestCache) finish( + key snapshotManifestCacheKey, + entry *snapshotManifestCacheEntry, + value snapshotManifestSet, + err error, +) { + c.mu.Lock() + defer c.mu.Unlock() + + if current, ok := c.entries[key]; ok && current == entry { + delete(c.entries, key) + if err == nil { + if c.complete.Contains(key) { + c.complete.Remove(key) + } + c.complete.Add(key, value) + c.completeManifestCount += snapshotManifestSetSize(value) + for c.completeManifestCount > snapshotManifestCacheManifestLimit { + if _, _, ok := c.complete.RemoveOldest(); !ok { + break + } + } + } + } + + entry.manifests = value + entry.err = err + entry.readyOnce.Do(func() { close(entry.ready) }) +} + +func readSnapshotManifestSet( + ctx context.Context, + snapshot Snapshot, + fsF FSysF, +) (snapshotManifestSet, error) { + if err := ctx.Err(); err != nil { + return snapshotManifestSet{}, err Review Comment: **minor** — Unreachable ctx.Err() guard in readSnapshotManifestSet, pinned only by a test of an impossible path Every production call reaches readSnapshotManifestSet through get(), which passes context.WithoutCancel(ctx); that context's Err() is always nil, and get() already checked ctx.Err() at :158. The guard can never fire outside tests. TestReadSnapshotManifestSetKeepsCompletedReadAfterCancellation exercises this by calling readSnapshotManifestSet directly with a raw cancellable ctx, which no caller does. Either drop the guard and the test, or make it meaningful by not detaching the factory call (see the major finding). ########## table/snapshot_manifest_cache.go: ########## @@ -0,0 +1,286 @@ +// 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" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + lru "github.com/hashicorp/golang-lru/v2" +) + +// snapshotManifestCacheSize bounds the number of decoded manifest lists held +// by a table. A table normally reuses its current snapshot, while a bounded +// history is enough to retain useful locality for repeated historical scans. +const snapshotManifestCacheSize = 64 + +// snapshotManifestCacheManifestLimit bounds the total number of manifest +// descriptors retained by the cache. This keeps a small number of unusually +// large snapshots from consuming an unbounded amount of memory. +const snapshotManifestCacheManifestLimit = 32 * 1024 + +// snapshotManifestSet keeps the complete manifest list and data partition +// together. Manifest descriptors are immutable for an Iceberg snapshot, so a +// table can safely share this decoded result across scans. +type snapshotManifestSet struct { + all []iceberg.ManifestFile + data []iceberg.ManifestFile +} + +func newSnapshotManifestSet(manifests []iceberg.ManifestFile) snapshotManifestSet { + set := snapshotManifestSet{all: manifests} + dataCount := 0 + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataCount++ + } + } + if dataCount == 0 { + return set + } + + set.data = make([]iceberg.ManifestFile, 0, dataCount) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + set.data = append(set.data, manifest) + } + } + + return set +} + +func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile { + return slices.Clone(s.all) +} + +func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile { + return slices.Clone(s.data) +} + +func snapshotManifestSetSize(set snapshotManifestSet) int { + return len(set.all) + len(set.data) +} + +type snapshotManifestCacheKey struct { + snapshotID int64 + manifestList string + hasEmbeddedSources bool + embeddedManifestSources string +} + +func snapshotManifestCacheKeyFor(snapshot Snapshot) snapshotManifestCacheKey { + var ( + hasEmbeddedSources bool + embeddedManifestSources string + ) + if snapshot.ManifestList == "" { + hasEmbeddedSources = snapshot.ManifestLocations != nil + // Valid file locations cannot contain a literal NUL, so this join is + // collision-free while preserving nil versus empty locations. + embeddedManifestSources = strings.Join(snapshot.ManifestLocations, "\x00") + } + + return snapshotManifestCacheKey{ + snapshotID: snapshot.SnapshotID, + manifestList: snapshot.ManifestList, + hasEmbeddedSources: hasEmbeddedSources, + embeddedManifestSources: embeddedManifestSources, + } +} + +type snapshotManifestCacheEntry struct { + ready chan struct{} + readyOnce sync.Once + manifests snapshotManifestSet + err error +} + +type snapshotManifestLoader func(context.Context) (snapshotManifestSet, error) + +// snapshotManifestCache memoizes successful manifest-list reads and shares an +// in-flight read with concurrent scans. Completed reads use a bounded LRU so a +// historical scan cannot retain every snapshot for the lifetime of a table. +// Failed reads are removed so a transient object-store error does not poison +// the cache. +type snapshotManifestCache struct { + mu sync.Mutex + entries map[snapshotManifestCacheKey]*snapshotManifestCacheEntry + complete *lru.Cache[snapshotManifestCacheKey, snapshotManifestSet] + completeManifestCount int +} + +func newSnapshotManifestCache() *snapshotManifestCache { + cache := &snapshotManifestCache{ + entries: make(map[snapshotManifestCacheKey]*snapshotManifestCacheEntry), + } + complete, err := lru.NewWithEvict( + snapshotManifestCacheSize, + func(_ snapshotManifestCacheKey, value snapshotManifestSet) { + cache.completeManifestCount -= snapshotManifestSetSize(value) + }, + ) + if err != nil { + panic(err) + } + cache.complete = complete + + return cache +} + +func (c *snapshotManifestCache) get( + ctx context.Context, + snapshot Snapshot, + load snapshotManifestLoader, +) (snapshotManifestSet, error) { + if c == nil { + return load(ctx) + } + if err := ctx.Err(); err != nil { + return snapshotManifestSet{}, err + } + + key := snapshotManifestCacheKeyFor(snapshot) + c.mu.Lock() + if manifests, ok := c.complete.Get(key); ok { + c.mu.Unlock() + + return manifests, nil + } + if entry, ok := c.entries[key]; ok { + c.mu.Unlock() + + select { + case <-entry.ready: + return entry.manifests, entry.err + default: + select { + case <-entry.ready: + return entry.manifests, entry.err + case <-ctx.Done(): + return snapshotManifestSet{}, ctx.Err() + } + } + } + + entry := &snapshotManifestCacheEntry{ready: make(chan struct{})} + c.entries[key] = entry + c.mu.Unlock() + + var ( + value snapshotManifestSet + err error + ) + defer func() { + if recovered := recover(); recovered != nil { + c.finish(key, entry, snapshotManifestSet{}, fmt.Errorf( + "panic while reading snapshot %d manifest list: %v", snapshot.SnapshotID, recovered)) + panic(recovered) + } + + c.finish(key, entry, value, err) + }() + + // The read is shared with callers whose contexts may outlive this one. Do + // not let the producer's cancellation abort a healthy waiter's read. + value, err = load(context.WithoutCancel(ctx)) + Review Comment: **major** — context.WithoutCancel makes the IO-factory resolution uncancellable, so scans can hang past their deadline The producer runs load(context.WithoutCancel(ctx)), and readSnapshotManifestSet:254 resolves fsF(ctx) inside that detached context. FSysF is not a cheap accessor: io.LoadFSFunc -> LoadFS -> the S3 backend calls config.LoadDefaultConfig(ctx) (io/gocloud/s3/s3.go:114), which performs ctx-bound IMDS/STS network I/O for credential resolution. Before this PR, planFilesLocal called scan.ioF(ctx) with the caller's context, so a deadline bounded that call. Now nothing can abort it: the caller's cancellation is stripped, no timeout is imposed, and the goroutine plus its IO are pinned indefinitely. Waiters can bail on their own ctx, but the producer cannot. Suggested fix: resolve the IO with the caller's context before entering the single-flight, and keep WithoutCancel only around the shared manifest-list read (the part whose result is memoized). No test pins the lost guarantee. <details><summary>Evidence</summary> ```text Probe (deleted after use) installing tbl.fsF = func(ctx){<-ctx.Done(); return nil, ctx.Err()} and calling tbl.Scan().PlanFiles(ctx) with a 200ms timeout. At merge base 52593109: 'PROBE1 PlanFiles returned within deadline: context deadline exceeded' --- PASS (0.20s). At head 2feda53: 'PROBE1 FAIL: PlanFiles hung past its context deadline; the IO factory call is uncancellable' --- FAIL (3.01s). ``` </details> ########## table/snapshot_manifest_cache.go: ########## @@ -0,0 +1,286 @@ +// 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" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + lru "github.com/hashicorp/golang-lru/v2" +) + +// snapshotManifestCacheSize bounds the number of decoded manifest lists held +// by a table. A table normally reuses its current snapshot, while a bounded +// history is enough to retain useful locality for repeated historical scans. +const snapshotManifestCacheSize = 64 + +// snapshotManifestCacheManifestLimit bounds the total number of manifest +// descriptors retained by the cache. This keeps a small number of unusually +// large snapshots from consuming an unbounded amount of memory. +const snapshotManifestCacheManifestLimit = 32 * 1024 + +// snapshotManifestSet keeps the complete manifest list and data partition +// together. Manifest descriptors are immutable for an Iceberg snapshot, so a +// table can safely share this decoded result across scans. +type snapshotManifestSet struct { + all []iceberg.ManifestFile + data []iceberg.ManifestFile +} + +func newSnapshotManifestSet(manifests []iceberg.ManifestFile) snapshotManifestSet { + set := snapshotManifestSet{all: manifests} + dataCount := 0 + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataCount++ + } + } + if dataCount == 0 { + return set + } + + set.data = make([]iceberg.ManifestFile, 0, dataCount) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + set.data = append(set.data, manifest) + } + } + + return set +} + +func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile { + return slices.Clone(s.all) +} + +func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile { + return slices.Clone(s.data) +} + +func snapshotManifestSetSize(set snapshotManifestSet) int { + return len(set.all) + len(set.data) +} + +type snapshotManifestCacheKey struct { + snapshotID int64 + manifestList string + hasEmbeddedSources bool + embeddedManifestSources string +} + +func snapshotManifestCacheKeyFor(snapshot Snapshot) snapshotManifestCacheKey { + var ( + hasEmbeddedSources bool + embeddedManifestSources string + ) + if snapshot.ManifestList == "" { + hasEmbeddedSources = snapshot.ManifestLocations != nil + // Valid file locations cannot contain a literal NUL, so this join is + // collision-free while preserving nil versus empty locations. + embeddedManifestSources = strings.Join(snapshot.ManifestLocations, "\x00") + } + + return snapshotManifestCacheKey{ + snapshotID: snapshot.SnapshotID, + manifestList: snapshot.ManifestList, + hasEmbeddedSources: hasEmbeddedSources, + embeddedManifestSources: embeddedManifestSources, + } +} + +type snapshotManifestCacheEntry struct { + ready chan struct{} + readyOnce sync.Once + manifests snapshotManifestSet + err error +} + +type snapshotManifestLoader func(context.Context) (snapshotManifestSet, error) + +// snapshotManifestCache memoizes successful manifest-list reads and shares an +// in-flight read with concurrent scans. Completed reads use a bounded LRU so a +// historical scan cannot retain every snapshot for the lifetime of a table. +// Failed reads are removed so a transient object-store error does not poison +// the cache. +type snapshotManifestCache struct { + mu sync.Mutex + entries map[snapshotManifestCacheKey]*snapshotManifestCacheEntry + complete *lru.Cache[snapshotManifestCacheKey, snapshotManifestSet] + completeManifestCount int +} + +func newSnapshotManifestCache() *snapshotManifestCache { + cache := &snapshotManifestCache{ + entries: make(map[snapshotManifestCacheKey]*snapshotManifestCacheEntry), + } + complete, err := lru.NewWithEvict( + snapshotManifestCacheSize, + func(_ snapshotManifestCacheKey, value snapshotManifestSet) { + cache.completeManifestCount -= snapshotManifestSetSize(value) + }, + ) + if err != nil { + panic(err) + } + cache.complete = complete + + return cache +} + +func (c *snapshotManifestCache) get( + ctx context.Context, + snapshot Snapshot, + load snapshotManifestLoader, +) (snapshotManifestSet, error) { + if c == nil { + return load(ctx) + } + if err := ctx.Err(); err != nil { + return snapshotManifestSet{}, err + } + + key := snapshotManifestCacheKeyFor(snapshot) + c.mu.Lock() + if manifests, ok := c.complete.Get(key); ok { + c.mu.Unlock() + + return manifests, nil + } + if entry, ok := c.entries[key]; ok { + c.mu.Unlock() + + select { + case <-entry.ready: + return entry.manifests, entry.err + default: + select { + case <-entry.ready: + return entry.manifests, entry.err + case <-ctx.Done(): + return snapshotManifestSet{}, ctx.Err() + } + } + } + + entry := &snapshotManifestCacheEntry{ready: make(chan struct{})} + c.entries[key] = entry + c.mu.Unlock() + + var ( + value snapshotManifestSet + err error + ) + defer func() { + if recovered := recover(); recovered != nil { + c.finish(key, entry, snapshotManifestSet{}, fmt.Errorf( + "panic while reading snapshot %d manifest list: %v", snapshot.SnapshotID, recovered)) + panic(recovered) + } + + c.finish(key, entry, value, err) + }() + + // The read is shared with callers whose contexts may outlive this one. Do + // not let the producer's cancellation abort a healthy waiter's read. + value, err = load(context.WithoutCancel(ctx)) + + return value, err +} + +func (c *snapshotManifestCache) finish( + key snapshotManifestCacheKey, + entry *snapshotManifestCacheEntry, + value snapshotManifestSet, + err error, +) { + c.mu.Lock() + defer c.mu.Unlock() + + if current, ok := c.entries[key]; ok && current == entry { + delete(c.entries, key) + if err == nil { + if c.complete.Contains(key) { + c.complete.Remove(key) Review Comment: **minor** — Unreachable defensive Contains/Remove before Add in finish A producer is only created on a complete-cache miss (get:164-168), and entries[key] serializes producers for a key, so no second finish for the same key can run concurrently. complete therefore cannot contain key when finish reaches :222. The branch is dead; if it is intended as insurance against a future refactor, that intent should be a comment, otherwise it should be removed along with the double-accounting it guards against. ########## table/table.go: ########## @@ -1398,6 +1413,7 @@ func New(ident Identifier, meta Metadata, metadataLocation string, fsF FSysF, ca metadata: meta, metadataLocation: metadataLocation, fsF: fsF, + manifestCache: newSnapshotManifestCache(), cat: cat, Review Comment: **minor** — Cross-scan manifest caching is always on with no way to disable or bound it by bytes New() unconditionally installs a cache, so every Table now retains up to 64 decoded manifest lists (bounded only by a descriptor count, not bytes) for its lifetime, with no table property or Option to turn it off. That is a meaningful change in memory posture for long-lived Table handles over wide snapshot histories, and it diverges from Java, where manifest caching is opt-in via io.manifest.cache-enabled (default false) with byte-based bounds (io.manifest.cache.max-total-bytes, io.manifest.cache.max-content-length). Worth at least a property to disable it. ########## table/inspect_files.go: ########## @@ -131,14 +131,15 @@ func (i InspectTable) manifestEntryReader( return nil, errors.New("table file IO is not configured") } - fs, err := i.tbl.fsF(ctx) + manifestSet, err := i.tbl.manifestSetWithFSF(ctx, *snapshot, sharedSnapshotManifestFSF(i.tbl.fsF)) if err != nil { Review Comment: **nit** — sharedSnapshotManifestFSF's sync.Once buys nothing on single-use paths, and inspect.go hand-rolls a fourth variant At inspect_files.go:134 the shared wrapper is constructed inline for a single manifestSetWithFSF call, so the memoization can never be exercised. Meanwhile inspect.go:455-461 wraps i.tbl.fsF in a bespoke closure that only adds a 'get file IO: %w' prefix instead of using the same helper. Four call sites, three different shapes, for one job. ########## table/snapshot_manifest_cache.go: ########## @@ -0,0 +1,286 @@ +// 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" + "slices" + "strings" + "sync" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + lru "github.com/hashicorp/golang-lru/v2" +) + +// snapshotManifestCacheSize bounds the number of decoded manifest lists held +// by a table. A table normally reuses its current snapshot, while a bounded +// history is enough to retain useful locality for repeated historical scans. +const snapshotManifestCacheSize = 64 + +// snapshotManifestCacheManifestLimit bounds the total number of manifest +// descriptors retained by the cache. This keeps a small number of unusually +// large snapshots from consuming an unbounded amount of memory. +const snapshotManifestCacheManifestLimit = 32 * 1024 + +// snapshotManifestSet keeps the complete manifest list and data partition +// together. Manifest descriptors are immutable for an Iceberg snapshot, so a +// table can safely share this decoded result across scans. +type snapshotManifestSet struct { + all []iceberg.ManifestFile + data []iceberg.ManifestFile +} + +func newSnapshotManifestSet(manifests []iceberg.ManifestFile) snapshotManifestSet { + set := snapshotManifestSet{all: manifests} + dataCount := 0 + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataCount++ + } + } + if dataCount == 0 { + return set + } + + set.data = make([]iceberg.ManifestFile, 0, dataCount) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + set.data = append(set.data, manifest) + } + } + + return set +} + +func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile { + return slices.Clone(s.all) +} + +func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile { + return slices.Clone(s.data) +} + +func snapshotManifestSetSize(set snapshotManifestSet) int { + return len(set.all) + len(set.data) Review Comment: **minor** — snapshotManifestSetSize double-counts data manifests, halving the documented descriptor bound size = len(all) + len(data), but data holds the same descriptors already counted in all. For an all-data snapshot (the common case) every manifest counts twice, so snapshotManifestCacheManifestLimit = 32*1024 admits only ~16384 distinct manifests, not the '32 * 1024 manifest descriptors' the comment at :37-40 claims. Either count len(all) only, or restate the constant's meaning. ########## table/table.go: ########## @@ -376,6 +388,7 @@ func (t Table) AllManifests(ctx context.Context) iter.Seq2[iceberg.ManifestFile, ch := make(chan list, allManifestsWorkerCount(n)) workers := allManifestsWorkerCount(n) g, groupCtx := errgroup.WithContext(workCtx) + manifestFS := sharedSnapshotManifestFSF(t.fsF) Review Comment: **nit** — AllManifests no longer surfaces an IO-factory error for a table with zero snapshots Previously fs, err := t.fsF(ctx) ran eagerly and a failure was yielded as an error even when there were no snapshots. sharedSnapshotManifestFSF is lazy and is never invoked when the jobs channel yields nothing, so an empty table with a broken IO factory now returns an empty sequence instead of an error. Probably the better behavior, but it is an undocumented, untested change. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
