laskoviymishka commented on code in PR #2147: URL: https://github.com/apache/iceberg-go/pull/2147#discussion_r4234101778
########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) Review Comment: `finishLoad` isn't deferred, so if `base.Open` or the read panics, `loads[location]` stays set and `ready` never closes. Every later open of that path then blocks in the select, and a caller with `context.Background()` hangs for the life of the handle. I'd wrap the producer section in a deferred finish that completes the load with an error on panic (re-panicking afterward if we want to preserve the crash). ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { Review Comment: On a failed shared load, every waiter here plus the producer at the fallback each re-open the path independently, so one failure fans out into N+1 backend reads right when the backend is likely already struggling, and the original error gets dropped with no log. I'd hand the shared `load.err` back to waiters, or let one waiter become the new producer. If the Java-style fallback is intentional, keep it but preserve and surface the original error rather than silently re-reading. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) + if err != nil { + if canceled { + return nil, ctx.Err() + } + + // Match Java's manifest-content cache behavior: a cache population + // failure falls back to an ordinary file open instead of changing + // the manifest read's error semantics. + return base.Open(location) + } + + return newCachedManifestFile(location, content), nil + } +} + +func (c *manifestContentCache) getLocked(location string, expectedLength int64) ([]byte, bool) { + entry, ok := c.entries[location] + if !ok { + return nil, false + } + if int64(len(entry.content)) != expectedLength { + c.removeEntryLocked(entry) + + return nil, false + } + + if c.expirationIntervalMs > 0 { Review Comment: Expired entries only get removed when that key is read again or LRU pressure evicts, so an idle handle can hold up to the full `max-total-bytes` resident indefinitely even though the docs say entries "expire". I'd sweep expired tail entries in `addEntryLocked` (access order is monotonic from the back), or reword the docs so "expire" clearly means "not reused on next access" rather than "reclaimed". ########## table/table.go: ########## @@ -248,6 +249,7 @@ func (t *Table) Refresh(ctx context.Context) error { t.fsF = fresh.fsF t.metadataLocation = fresh.metadataLocation t.manifestCache = newSnapshotManifestCacheForMetadata(fresh.metadata) + t.manifestContentCache = fresh.manifestContentCache Review Comment: This drops the whole content cache on every `Refresh`, since `fresh` comes from a fresh `New()` with an empty cache (the same thing happens on commit and the copy-constructor paths). Manifests are immutable and keyed by path, so after a new snapshot most of the previous scan's manifests are still valid, and we throw them away in exactly the refresh-then-rescan loop this cache is meant to speed up. I'd keep `t.manifestContentCache` across `Refresh` when the new config yields the same limits, and only rebuild when the settings actually change. If dropping on refresh is deliberate, let's say so here and in `configuration.md`, because the docs currently read as if only the manifest-list cache is replaced. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) + if err != nil { + if canceled { + return nil, ctx.Err() + } + + // Match Java's manifest-content cache behavior: a cache population + // failure falls back to an ordinary file open instead of changing + // the manifest read's error semantics. + return base.Open(location) + } + + return newCachedManifestFile(location, content), nil + } +} + +func (c *manifestContentCache) getLocked(location string, expectedLength int64) ([]byte, bool) { + entry, ok := c.entries[location] + if !ok { + return nil, false + } + if int64(len(entry.content)) != expectedLength { + c.removeEntryLocked(entry) + + return nil, false + } + + if c.expirationIntervalMs > 0 { + now := time.Now().UnixMilli() + if now-entry.lastAccessMs >= c.expirationIntervalMs { + c.removeEntryLocked(entry) + + return nil, false + } + entry.lastAccessMs = now + } + c.lru.MoveToFront(entry.element) + + return entry.content, true +} + +func (c *manifestContentCache) finishLoad( + location string, + expectedLength int64, + load *manifestContentCacheLoad, + content []byte, + err error, + canceled bool, +) { + c.mu.Lock() + defer c.mu.Unlock() + + if current, ok := c.loads[location]; ok && current == load { + delete(c.loads, location) + if err == nil && int64(len(content)) == expectedLength { + c.addEntryLocked(location, content) + } + } + + load.content = content + load.err = err + load.canceled = canceled + close(load.ready) +} + +func (c *manifestContentCache) addEntryLocked(location string, content []byte) { + size := int64(len(content)) + if size > c.maxContentLength || size > c.maxTotalBytes { + return + } + if previous, ok := c.entries[location]; ok { + c.removeEntryLocked(previous) + } + for c.totalBytes > c.maxTotalBytes-size && c.lru.Len() > 0 { + oldest := c.lru.Back().Value.(*manifestContentCacheEntry) Review Comment: The single-value assertion `c.lru.Back().Value.(*manifestContentCacheEntry)` panics if the element type is ever wrong, and it panics while holding `mu`, which combined with the non-deferred finish above would wedge the cache. Use the two-value form and bail safely. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) + if err != nil { + if canceled { + return nil, ctx.Err() + } + + // Match Java's manifest-content cache behavior: a cache population + // failure falls back to an ordinary file open instead of changing + // the manifest read's error semantics. + return base.Open(location) + } + + return newCachedManifestFile(location, content), nil + } +} + +func (c *manifestContentCache) getLocked(location string, expectedLength int64) ([]byte, bool) { + entry, ok := c.entries[location] + if !ok { + return nil, false + } + if int64(len(entry.content)) != expectedLength { + c.removeEntryLocked(entry) + + return nil, false + } + + if c.expirationIntervalMs > 0 { + now := time.Now().UnixMilli() Review Comment: `time.Now().UnixMilli()` drops the monotonic reading, so an NTP step or a VM pause can expire a fresh entry or keep a stale one alive, and it's also why the test has to reach in and rewrite `lastAccessMs`. I'd store a `time.Time` and inject a `now func() time.Time` defaulting to `time.Now`, so expiry uses a monotonic delta and tests can drive a fake clock. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) + if err != nil { + if canceled { + return nil, ctx.Err() + } + + // Match Java's manifest-content cache behavior: a cache population + // failure falls back to an ordinary file open instead of changing + // the manifest read's error semantics. + return base.Open(location) + } + + return newCachedManifestFile(location, content), nil + } +} + +func (c *manifestContentCache) getLocked(location string, expectedLength int64) ([]byte, bool) { + entry, ok := c.entries[location] + if !ok { + return nil, false + } + if int64(len(entry.content)) != expectedLength { + c.removeEntryLocked(entry) + + return nil, false + } + + if c.expirationIntervalMs > 0 { + now := time.Now().UnixMilli() + if now-entry.lastAccessMs >= c.expirationIntervalMs { + c.removeEntryLocked(entry) + + return nil, false + } + entry.lastAccessMs = now + } + c.lru.MoveToFront(entry.element) + + return entry.content, true +} + +func (c *manifestContentCache) finishLoad( + location string, + expectedLength int64, + load *manifestContentCacheLoad, + content []byte, + err error, + canceled bool, +) { + c.mu.Lock() + defer c.mu.Unlock() + + if current, ok := c.loads[location]; ok && current == load { + delete(c.loads, location) + if err == nil && int64(len(content)) == expectedLength { + c.addEntryLocked(location, content) + } + } + + load.content = content + load.err = err + load.canceled = canceled + close(load.ready) +} + +func (c *manifestContentCache) addEntryLocked(location string, content []byte) { + size := int64(len(content)) + if size > c.maxContentLength || size > c.maxTotalBytes { + return + } + if previous, ok := c.entries[location]; ok { + c.removeEntryLocked(previous) + } + for c.totalBytes > c.maxTotalBytes-size && c.lru.Len() > 0 { + oldest := c.lru.Back().Value.(*manifestContentCacheEntry) + c.removeEntryLocked(oldest) + } + + entry := &manifestContentCacheEntry{ + location: location, + content: content, + } + if c.expirationIntervalMs > 0 { + entry.lastAccessMs = time.Now().UnixMilli() + } + entry.element = c.lru.PushFront(entry) + c.entries[location] = entry + c.totalBytes += size +} + +func (c *manifestContentCache) removeEntryLocked(entry *manifestContentCacheEntry) { + delete(c.entries, entry.location) + c.lru.Remove(entry.element) + c.totalBytes -= int64(len(entry.content)) +} + +func readManifestContent(base iceio.IO, location string, expectedLength int64) (content []byte, err error) { + file, err := base.Open(location) + if err != nil { + return nil, err + } + defer func() { + if closeErr := file.Close(); err == nil { + err = closeErr + } + }() + + // The advertised length has already passed the cache's size limits. Read + // into an exact-sized buffer so retained capacity matches byte accounting + // and a larger actual file cannot grow the population buffer without bound. + content = make([]byte, expectedLength) + if _, err := io.ReadFull(file, content); err != nil { + return nil, err + } + + var extra [1]byte + if _, err := io.ReadFull(file, extra[:]); err != io.EOF { Review Comment: `err != io.EOF` should be `errors.Is(err, io.EOF)` (errorlint flags the direct sentinel compare), and the inner `if _, err := io.ReadFull(...)` shadows the named return, which is easy to misread here. Both are quick to tidy while we're in this file. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { Review Comment: These are advertised as Java-compatible keys, but I don't think the zero and negative `expiration-interval-ms` semantics were verified against Java's `ContentCache`/`CatalogProperties`. Here `0` means never expire and a negative value disables the cache; if Java instead uses `-1` for no-expiry and `0` for immediate expiry, then someone sharing Java config and setting `-1` silently gets no cache. Can you confirm against Java and either match it or document the divergence in `properties.go` and the config docs? Separately, zero or negative `max-total-bytes`/`max-content-length` (and a non-numeric typo, which falls back to the default via `GetInt64`) silently returns nil and disables the cache with no signal, so I'd at least log a warning when a bad value turns it off. ########## table/manifest_content_cache.go: ########## @@ -0,0 +1,337 @@ +// 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 ( + "bytes" + "container/list" + "context" + "fmt" + "io" + "io/fs" + "path" + "sync" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" +) + +// manifestContentCache stores immutable manifest bytes so repeated scans can +// reuse the same object-store read while still decoding and filtering entries +// independently for each scan. +type manifestContentCache struct { + mu sync.Mutex + + expirationIntervalMs int64 + maxTotalBytes int64 + maxContentLength int64 + totalBytes int64 + + entries map[string]*manifestContentCacheEntry + lru list.List + loads map[string]*manifestContentCacheLoad +} + +type manifestContentCacheEntry struct { + location string + content []byte + lastAccessMs int64 + element *list.Element +} + +type manifestContentCacheLoad struct { + ready chan struct{} + content []byte + err error + canceled bool +} + +func newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength int64) *manifestContentCache { + return &manifestContentCache{ + expirationIntervalMs: expirationIntervalMs, + maxTotalBytes: maxTotalBytes, + maxContentLength: maxContentLength, + entries: make(map[string]*manifestContentCacheEntry), + loads: make(map[string]*manifestContentCacheLoad), + } +} + +func newManifestContentCacheForConfig(config iceberg.Properties) *manifestContentCache { + if !config.GetBool(IOManifestCacheEnabledKey, IOManifestCacheEnabledDefault) { + return nil + } + + expirationIntervalMs := config.GetInt64( + IOManifestCacheExpirationIntervalMsKey, + IOManifestCacheExpirationIntervalMsDefault, + ) + maxTotalBytes := config.GetInt64(IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault) + maxContentLength := config.GetInt64( + IOManifestCacheMaxContentLengthKey, + IOManifestCacheMaxContentLengthDefault, + ) + if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 { + return nil + } + + return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength) +} + +func (c *manifestContentCache) wrap( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) iceio.IO { + if c == nil { + return base + } + + return &manifestContentCacheIO{ + ctx: ctx, + base: base, + cache: c, + manifest: manifest, + } +} + +func (c *manifestContentCache) open( + ctx context.Context, + base iceio.IO, + manifest iceberg.ManifestFile, +) (iceio.File, error) { + location := manifest.FilePath() + expectedLength := manifest.Length() + if expectedLength <= 0 || expectedLength > c.maxContentLength || expectedLength > c.maxTotalBytes { + return base.Open(location) + } + + for { + if err := ctx.Err(); err != nil { + return nil, err + } + + c.mu.Lock() + if content, ok := c.getLocked(location, expectedLength); ok { + c.mu.Unlock() + + return newCachedManifestFile(location, content), nil + } + if load, ok := c.loads[location]; ok { + c.mu.Unlock() + select { + case <-load.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + if load.canceled { + continue + } + // Concurrent callers may describe the same path with different lengths. + // Never reuse in-flight bytes that do not match this caller's descriptor. + if load.err != nil || int64(len(load.content)) != expectedLength { + return base.Open(location) + } + + return newCachedManifestFile(location, load.content), nil + } + + load := &manifestContentCacheLoad{ready: make(chan struct{})} + c.loads[location] = load + c.mu.Unlock() + + content, err := readManifestContent(base, location, expectedLength) + canceled := err != nil && ctx.Err() != nil + c.finishLoad(location, expectedLength, load, content, err, canceled) + if err != nil { + if canceled { + return nil, ctx.Err() + } + + // Match Java's manifest-content cache behavior: a cache population + // failure falls back to an ordinary file open instead of changing + // the manifest read's error semantics. + return base.Open(location) + } + + return newCachedManifestFile(location, content), nil + } +} + +func (c *manifestContentCache) getLocked(location string, expectedLength int64) ([]byte, bool) { + entry, ok := c.entries[location] + if !ok { + return nil, false + } + if int64(len(entry.content)) != expectedLength { + c.removeEntryLocked(entry) + + return nil, false + } + + if c.expirationIntervalMs > 0 { + now := time.Now().UnixMilli() + if now-entry.lastAccessMs >= c.expirationIntervalMs { + c.removeEntryLocked(entry) + + return nil, false + } + entry.lastAccessMs = now + } + c.lru.MoveToFront(entry.element) + + return entry.content, true +} + +func (c *manifestContentCache) finishLoad( + location string, + expectedLength int64, + load *manifestContentCacheLoad, + content []byte, + err error, + canceled bool, +) { + c.mu.Lock() + defer c.mu.Unlock() + + if current, ok := c.loads[location]; ok && current == load { + delete(c.loads, location) + if err == nil && int64(len(content)) == expectedLength { + c.addEntryLocked(location, content) + } + } + + load.content = content + load.err = err + load.canceled = canceled + close(load.ready) +} + +func (c *manifestContentCache) addEntryLocked(location string, content []byte) { + size := int64(len(content)) + if size > c.maxContentLength || size > c.maxTotalBytes { + return + } + if previous, ok := c.entries[location]; ok { + c.removeEntryLocked(previous) + } + for c.totalBytes > c.maxTotalBytes-size && c.lru.Len() > 0 { + oldest := c.lru.Back().Value.(*manifestContentCacheEntry) + c.removeEntryLocked(oldest) + } + + entry := &manifestContentCacheEntry{ + location: location, + content: content, + } + if c.expirationIntervalMs > 0 { + entry.lastAccessMs = time.Now().UnixMilli() + } + entry.element = c.lru.PushFront(entry) + c.entries[location] = entry + c.totalBytes += size +} + +func (c *manifestContentCache) removeEntryLocked(entry *manifestContentCacheEntry) { + delete(c.entries, entry.location) + c.lru.Remove(entry.element) + c.totalBytes -= int64(len(entry.content)) +} + +func readManifestContent(base iceio.IO, location string, expectedLength int64) (content []byte, err error) { + file, err := base.Open(location) + if err != nil { + return nil, err + } + defer func() { + if closeErr := file.Close(); err == nil { + err = closeErr + } + }() + + // The advertised length has already passed the cache's size limits. Read + // into an exact-sized buffer so retained capacity matches byte accounting + // and a larger actual file cannot grow the population buffer without bound. + content = make([]byte, expectedLength) + if _, err := io.ReadFull(file, content); err != nil { + return nil, err + } + + var extra [1]byte + if _, err := io.ReadFull(file, extra[:]); err != io.EOF { + if err == nil { + err = fmt.Errorf("manifest content exceeds expected length %d", expectedLength) + } + + return nil, err + } + + return content, nil +} + +type manifestContentCacheIO struct { + ctx context.Context Review Comment: Two small things on this wrapper. It stores the errgroup `gctx`, so it's only safe as long as it never outlives the scan goroutine; a one-line comment to that effect would help. And it forwards only `Open`/`Remove`, so any optional interface the base IO exposes (probed by type assertion elsewhere) goes invisible once wrapped. That's fine as long as the manifest read path only calls `Open`, which it does today, but worth a comment pinning the assumption. ########## website/src/configuration.md: ########## @@ -355,12 +355,22 @@ Property key constants are in [`table/properties.go`](https://github.com/apache/ |---|---|---| | `read.split.target-size` | `134217728` | Target size for coalescing safe row-group ranges when splitting large local Parquet files. A range may exceed the target when no supplied offset can divide it safely. | | `read.manifest-list-cache.enabled` | `true` | Reuse decoded snapshot manifest lists across scans of a table handle. Set to `false` to disable retention and shared in-flight reads. Applied when loading or refreshing a table, and when creating a transaction scan. | +| `io.manifest.cache-enabled` | `false` | Reuse raw manifest file content across local scans. This mirrors Java's FileIO/catalog setting and is read from the table's saved catalog config. | +| `io.manifest.cache.expiration-interval-ms` | `60000` | Expire manifest content after this many milliseconds without access. Zero disables time-based expiration. | +| `io.manifest.cache.max-total-bytes` | `104857600` | Maximum total raw manifest bytes retained by the content cache. | Review Comment: The "100 MiB total" here reads as a global ceiling, but it's optimistic two ways: the cache is per table handle, so N handles over one FileIO can retain up to N times this (Java shares one budget per FileIO), and in-flight population buffers plus entries a reader still holds aren't counted against the cap. I'd note both in this row so the number isn't read as a hard global bound. ########## table/manifest_content_cache_internal_test.go: ########## @@ -0,0 +1,370 @@ +// 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 ( + "bytes" + "context" + "io" + "sync" + "testing" + "time" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTableScansReuseManifestContent(t *testing.T) { + fs := newTrackingCallsIO() + tbl, manifestListPath, manifestPath := newManifestContentCacheTestTable(t, fs, + iceberg.Properties{IOManifestCacheEnabledKey: "true"}) + + for range 2 { + tasks, err := tbl.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context()) + require.NoError(t, err) + require.Len(t, tasks, 1) + } + + assert.Equal(t, 1, fs.openCount[manifestListPath], "manifest list should use the decoded list cache") + assert.Equal(t, 1, fs.openCount[manifestPath], "manifest bytes should be read once across scans") +} + +func TestManifestContentCacheSkipsOversizedManifest(t *testing.T) { + fs := newTrackingCallsIO() + tbl, _, manifestPath := newManifestContentCacheTestTable(t, fs, iceberg.Properties{ + IOManifestCacheEnabledKey: "true", + IOManifestCacheMaxContentLengthKey: "1", + }) + + for range 2 { + tasks, err := tbl.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context()) + require.NoError(t, err) + require.Len(t, tasks, 1) + } + + assert.Equal(t, 2, fs.openCount[manifestPath], "oversized manifests must bypass the content cache") +} + +func TestManifestContentCacheBoundsPopulationAndPreservesFullReads(t *testing.T) { + const location = "mem://manifest-content-cache/length.avro" + content := bytes.Repeat([]byte("x"), 32) + for _, tt := range []struct { + name string + expectedLength int64 + wantOpens int + firstReadBytes int + }{ + {name: "matching length", expectedLength: 32, wantOpens: 1, firstReadBytes: 32}, + {name: "longer content", expectedLength: 8, wantOpens: 2, firstReadBytes: 9}, + {name: "shorter content", expectedLength: 64, wantOpens: 2, firstReadBytes: 32}, + {name: "unknown length", expectedLength: 0, wantOpens: 1, firstReadBytes: 32}, + } { + t.Run(tt.name, func(t *testing.T) { + base := iceio.NewMemFS() + require.NoError(t, base.WriteFile(location, content)) + fs := &manifestContentCountingIO{IO: base} + cache := newManifestContentCache(0, 128, 64) + manifest := iceberg.NewManifestFile(2, location, tt.expectedLength, 0, 1).Build() + + file, err := cache.open(t.Context(), fs, manifest) + require.NoError(t, err) + got, err := io.ReadAll(file) + require.NoError(t, err) + require.NoError(t, file.Close()) + assert.Equal(t, content, got, "cache population must not truncate the ordinary read") + require.Len(t, fs.files, tt.wantOpens) + assert.Equal(t, tt.firstReadBytes, fs.files[0].bytesRead) + if tt.expectedLength == int64(len(content)) { + require.Contains(t, cache.entries, location) + assert.Equal(t, len(content), cap(cache.entries[location].content)) + assert.Equal(t, int64(len(content)), cache.totalBytes) + } else { + assert.Empty(t, cache.entries) + assert.Zero(t, cache.totalBytes) + } + }) + } +} + +type manifestContentCountingIO struct { + iceio.IO + files []*manifestContentCountingFile +} + +func (f *manifestContentCountingIO) Open(name string) (iceio.File, error) { + file, err := f.IO.Open(name) + if err != nil { + return nil, err + } + counted := &manifestContentCountingFile{File: file} + f.files = append(f.files, counted) + + return counted, nil +} + +type manifestContentCountingFile struct { + iceio.File + bytesRead int +} + +func (f *manifestContentCountingFile) Read(p []byte) (int, error) { + n, err := f.File.Read(p) + f.bytesRead += n + + return n, err +} + +func TestManifestContentCacheEvictsLeastRecentlyUsed(t *testing.T) { + fs := newTrackingCallsIO() + const ( + firstPath = "mem://manifest-content-cache/first.avro" + secondPath = "mem://manifest-content-cache/second.avro" + thirdPath = "mem://manifest-content-cache/third.avro" + ) + firstBytes := []byte("aaaa") + secondBytes := []byte("bbbb") + require.NoError(t, fs.WriteFile(firstPath, firstBytes)) + require.NoError(t, fs.WriteFile(secondPath, secondBytes)) + require.NoError(t, fs.WriteFile(thirdPath, []byte("cccc"))) + + first := iceberg.NewManifestFile(2, firstPath, int64(len(firstBytes)), 0, 1).Build() + second := iceberg.NewManifestFile(2, secondPath, int64(len(secondBytes)), 0, 1).Build() + third := iceberg.NewManifestFile(2, thirdPath, 4, 0, 1).Build() + cache := newManifestContentCache(0, 8, 4) + + readCachedManifest(t, cache, fs, first) + readCachedManifest(t, cache, fs, second) + readCachedManifest(t, cache, fs, first) + readCachedManifest(t, cache, fs, third) + + assert.Contains(t, cache.entries, firstPath, "access must refresh LRU order") + assert.NotContains(t, cache.entries, secondPath) + assert.Equal(t, 1, fs.openCount[firstPath]) + assert.Equal(t, 1, fs.openCount[secondPath]) + assert.Equal(t, int64(8), cache.totalBytes) +} + +func TestManifestContentCacheExpiresIdleEntries(t *testing.T) { + const location = "mem://manifest-content-cache/expired.avro" + fs := newTrackingCallsIO() + require.NoError(t, fs.WriteFile(location, []byte("aaaa"))) + manifest := iceberg.NewManifestFile(2, location, 4, 0, 1).Build() + cache := newManifestContentCache(60_000, 8, 4) + + readCachedManifest(t, cache, fs, manifest) + cache.entries[location].lastAccessMs = time.Now().Add(-time.Hour).UnixMilli() + readCachedManifest(t, cache, fs, manifest) + + assert.Equal(t, 2, fs.openCount[location]) + assert.Equal(t, int64(4), cache.totalBytes) + assert.Len(t, cache.entries, 1) +} + +func TestManifestContentCacheSharesInFlightRead(t *testing.T) { Review Comment: This in-flight test can pass even if single-flight were partly broken: callers that start after `release` just hit the cache instead of joining the load, so the single-open assertion doesn't guarantee the shared path ran. I'd gate on all 16 callers entering before releasing. Two subtle branches also aren't covered: the failure-fallback (first open or read fails, second succeeds, assert nothing was cached) and the producer-cancel path (producer's open fails under a canceled ctx while a live waiter still succeeds with exactly one more read). Those are the ones worth pinning. -- 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]
