You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "fallintoplace (via GitHub)" <gi...@apache.org> on 2026/10/07 11:20:08 UTC
[PR] perf(table): cache manifest content across scans [iceberg-go]
fallintoplace opened a new pull request, #2147:
URL: https://github.com/apache/iceberg-go/pull/2147
**What**
- Add opt-in raw manifest content caching across scans of a table handle.
**Why**
- Repeated local planning reopens immutable manifests even when the manifest list is cached. Remote FileIO adds latency for each read.
**Implementation**
- Byte-bounded LRU with idle expiration and a per-file size limit.
- Share concurrent misses. Keep waiter cancellation separate and fall back to normal FileIO on population errors.
- Read settings from saved catalog config. Disabled by default; limits are 100 MiB total, 8 MiB per manifest, 60s idle expiration.
- Integrate local scan planning and transaction scans. Keep Avro decoding, projection and filtering per scan.
**Benchmark**
Actual repeated `Scan.PlanFiles`, real Avro manifests, warm cache, concurrency 1. Apple M1 Pro, darwin/arm64, Go 1.26.3. Median of 3 runs, 300ms each, `-cpu=1`.
| Manifests | Simulated latency per open | Disabled ms/op | Warm ms/op | Manifest opens before → after |
| --- | --- | ---: | ---: | ---: |
| 8 | 0 | 2.21 | 2.26 | 8 → 0 |
| 32 | 0 | 8.69 | 9.51 | 32 → 0 |
| 8 | 1ms | 12.79 | 2.45 | 8 → 0 |
| 32 | 1ms | 48.55 | 8.85 | 32 → 0 |
Warm planning is about **5.2–5.5x faster** with simulated backend latency. Zero-latency controls are 2–9% slower. Each cached manifest adds one small allocation; Avro decoding remains unchanged.
```sh
go test ./table -run '^$' -bench '^BenchmarkPlanFilesManifestContentCache$' -benchmem -benchtime=300ms -count=3 -cpu=1
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): cache manifest content across scans [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #2147:
URL: https://github.com/apache/iceberg-go/pull/2147
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): cache manifest content across scans [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on PR #2147:
URL: https://github.com/apache/iceberg-go/pull/2147#issuecomment-6348031342
@fallintoplace - fix conflicts and i think we are good to merge
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): cache manifest content across scans [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #2147:
URL: https://github.com/apache/iceberg-go/pull/2147#discussion_r4328220249
##########
table/manifest_content_cache.go:
##########
@@ -0,0 +1,439 @@
+// 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"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "log"
+ "path"
+ "strconv"
+ "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
+ now func() time.Time
+ maxTotalBytes int64
+ maxContentLength int64
+ totalBytes int64
+
+ entries map[string]*manifestContentCacheEntry
+ lru list.List
+ loads map[string]*manifestContentCacheLoad
+}
+
+type manifestContentCacheEntry struct {
+ location string
+ content []byte
+ lastAccess time.Time
+ 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,
+ now: time.Now,
+ 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, ok := manifestContentCacheConfigInt64(config,
+ IOManifestCacheExpirationIntervalMsKey, IOManifestCacheExpirationIntervalMsDefault)
+ if !ok {
+ return nil
+ }
+ maxTotalBytes, ok := manifestContentCacheConfigInt64(config,
+ IOManifestCacheMaxTotalBytesKey, IOManifestCacheMaxTotalBytesDefault)
+ if !ok {
+ return nil
+ }
+ maxContentLength, ok := manifestContentCacheConfigInt64(config,
+ IOManifestCacheMaxContentLengthKey, IOManifestCacheMaxContentLengthDefault)
+ if !ok {
+ return nil
+ }
+ if expirationIntervalMs < 0 || maxTotalBytes <= 0 || maxContentLength <= 0 {
+ log.Printf("Warning: disabling manifest content cache: invalid limits (expiration=%d, total=%d, content=%d)",
+ expirationIntervalMs, maxTotalBytes, maxContentLength)
+
+ return nil
+ }
+
+ return newManifestContentCache(expirationIntervalMs, maxTotalBytes, maxContentLength)
+}
+
+func manifestContentCacheConfigInt64(config iceberg.Properties, key string, fallback int64) (int64, bool) {
+ raw, ok := config[key]
+ if !ok {
+ return fallback, true
+ }
+ value, err := strconv.ParseInt(raw, 10, 64)
+ if err != nil {
+ log.Printf("Warning: disabling manifest content cache: invalid %s=%q: %v", key, raw, err)
+
+ return 0, false
+ }
+
+ return value, true
+}
+
+// Reuse retained immutable manifests when a table is refreshed or committed.
+// Independent table handles still have independent byte budgets.
+func reuseManifestContentCache(previous, configured *manifestContentCache) *manifestContentCache {
+ if previous != nil && configured != nil &&
+ previous.expirationIntervalMs == configured.expirationIntervalMs &&
+ previous.maxTotalBytes == configured.maxTotalBytes &&
+ previous.maxContentLength == configured.maxContentLength {
+ return previous
+ }
+
+ return configured
+}
+
+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()
+ }
+ // Both channels can become ready together. Prefer cancellation
+ // even if the shared load also finished in the same instant.
+ if err := ctx.Err(); err != nil {
+ return nil, 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 {
Review Comment:
Handing `load.err` back to waiters was the first option I suggested last round, and it turned out to be the wrong one. That's on my suggestion, not your implementation. I'd have waiters take the same single plain `base.Open` the producer takes, the way the length-mismatch branch just below already does.
As it stands the outcome depends on timing. With a wrong `manifest_length`, an uncached scan succeeds and the producer succeeds through its fallback, but every waiter that raced it fails `PlanFiles` with `ErrUnexpectedEOF`. The same happens with a transient backend error. One open per waiter costs exactly what no cache costs, so there's no retry storm to guard against. On failure I'd wrap with the location, e.g. `fmt.Errorf("open manifest %s after failed cache population: %w", location, err)`, because the bare `load.err` doesn't say which manifest failed.
This flips `TestManifestContentCacheSharesFailedLoadWithoutRetryStorm` (followers expect `NoError`, plus one extra open per follower) and the "Existing waiters receive the shared error" sentence in configuration.md. While we're here, the producer drops the population error silently when its fallback succeeds. Since a mis-sized manifest never caches and gets read twice on every scan, a `log.Printf("Warning: ...")` there would at least make those double reads explainable.
##########
website/src/configuration.md:
##########
@@ -363,12 +363,25 @@ 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` | Idle expiration in milliseconds. Zero disables time-based expiration; negative values disable the cache. Expired entries are reclaimed on cache access/insertion, without a background cleanup task. |
+| `io.manifest.cache.max-total-bytes` | `104857600` | Retained raw bytes per independently loaded table. Handles returned by commit share their source table's cache and budget. Concurrent population buffers and bytes held by active readers are not counted. |
Review Comment:
This row and `max-content-length` below it don't say that non-positive or unparsable values warn and disable the cache, where Java rejects them. The same goes for the `IOManifestCacheMaxTotalBytesKey` / `IOManifestCacheMaxContentLengthKey` comments in properties.go. One "differs from Java" line under the table would cover all three keys, including the Java throw on negative expiration, which the properties.go comment mentions but this table doesn't.
##########
table/table.go:
##########
@@ -1482,6 +1488,7 @@ func New(ident Identifier, meta Metadata, metadataLocation string, fsF FSysF, ca
for _, opt := range opts {
opt(t)
}
+ t.manifestContentCache = newManifestContentCacheForConfig(t.savedConfig)
Review Comment:
This runs on every `New()`, so an invalid `io.manifest.cache.*` value logs its warning again on every `LoadTable`, `Refresh` and commit. On the reuse path we also build a cache only to throw it away. I'd parse and validate once, or only build when `reuseManifestContentCache` isn't going to keep the previous one. Not blocking.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): cache manifest content across scans [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] perf(table): cache manifest content across scans [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #2147:
URL: https://github.com/apache/iceberg-go/pull/2147#discussion_r4336958702
##########
table/properties.go:
##########
@@ -46,6 +46,33 @@ const (
ReadManifestListCacheEnabledKey = "read.manifest-list-cache.enabled"
ReadManifestListCacheEnabledDefault = true
+ // IOManifestCacheEnabledKey controls Java-compatible raw manifest-file
+ // content caching. These settings are FileIO/catalog configuration rather
Review Comment:
Could we read these from the catalog props the way Java's FileIO does, or document them as REST-only here and in configuration.md? `New()` builds the cache from `savedConfig` alone ([table.go](https://github.com/apache/iceberg-go/blob/7dd63959affa37186b5fc61bb64d2deb293f6791/table/table.go#L1491)), only REST sets it, and REST fills it from table properties plus the LoadTable `config` while leaving out `r.props` ([rest.go](https://github.com/apache/iceberg-go/blob/7dd63959affa37186b5fc61bb64d2deb293f6791/catalog/rest/rest.go#L1840-L1846)). So on REST this is effectively table metadata, and on Glue, SQL, Hive and Hadoop it's silently off. A test enabling it through a catalog `LoadTable` would pin whichever we pick.
##########
website/src/configuration.md:
##########
@@ -363,12 +363,28 @@ 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` | Idle expiration in milliseconds. Zero disables time-based expiration; negative values disable the cache. Expired entries are reclaimed on cache access/insertion, without a background cleanup task. |
Review Comment:
Could this row say unparsable values disable the cache too? A non-numeric expiration goes through the same parse branch as the size keys ([manifest_content_cache.go](https://github.com/apache/iceberg-go/blob/7dd63959affa37186b5fc61bb64d2deb293f6791/table/manifest_content_cache.go#L110-L123)), but `TestManifestContentCacheInvalidNumericConfig` only loops over the two size keys, so adding `IOManifestCacheExpirationIntervalMsKey` there would pin it.
```suggestion
| `io.manifest.cache.expiration-interval-ms` | `60000` | Idle expiration in milliseconds. Zero disables time-based expiration; negative or unparsable values warn and disable the cache. Expired entries are reclaimed on cache access/insertion, without a background cleanup task. |
```
##########
website/src/configuration.md:
##########
@@ -363,12 +363,28 @@ 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` | Idle expiration in milliseconds. Zero disables time-based expiration; negative values disable the cache. Expired entries are reclaimed on cache access/insertion, without a background cleanup task. |
+| `io.manifest.cache.max-total-bytes` | `104857600` | Retained raw bytes per independently loaded table. Handles returned by commit share their source table's cache and budget. Concurrent population buffers and bytes held by active readers are not counted. Non-positive or unparsable values warn and disable caching. |
+| `io.manifest.cache.max-content-length` | `8388608` | Maximum size of one manifest eligible for content caching. Compatible cache entries survive table refresh and commit. Non-positive or unparsable values warn and disable caching. |
Review Comment:
Could the survive-refresh sentence move to the paragraph below and say what "compatible" means? It describes the whole cache, not this key, and `reuseManifestContentCache` keeps the cache only when expiration, total bytes and content length are all unchanged ([manifest_content_cache.go](https://github.com/apache/iceberg-go/blob/7dd63959affa37186b5fc61bb64d2deb293f6791/table/manifest_content_cache.go#L125-L136)).
```suggestion
| `io.manifest.cache.max-content-length` | `8388608` | Maximum size of one manifest eligible for content caching. Non-positive or unparsable values warn and disable caching. |
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org