This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 1d953ba4d perf(table): cache manifest content across scans (#2147)
1d953ba4d is described below
commit 1d953ba4d09256c0f715186692f08c49a5d6de29
Author: Minh Vu <[email protected]>
AuthorDate: Fri Oct 23 18:16:08 2026 +0200
perf(table): cache manifest content across scans (#2147)
* perf(table): cache manifest content across scans
* fix(table): bound manifest cache population buffers
Signed-off-by: Minh Vu <[email protected]>
* fix(table): verify manifest length for in-flight cache readers
* style(table): gofmt manifest cache regression test
* fix(table): retain manifest cache and handle load failures safely
* style(table): satisfy manifest cache lint checks
* style(table): complete manifest cache formatting fixes
* test(table): cover cache refresh, single-flight failures and canceled
producers
* style(table): format refreshed manifest cache tests and commit path
* fix(table): preserve manifest cache load errors
Keep both population and fallback failures discoverable through errors.Is
and errors.As. Exercise warm cache reuse across transaction scans and commit,
and clarify the shared budget and failure behavior.
Signed-off-by: Codex <[email protected]>
Signed-off-by: Minh Vu <[email protected]>
* fix(table): honor canceled manifest-cache waiters when loads finish
* test(table): pin Java manifest cache expiration semantics
* fix(table): fall back for every failed shared manifest load
* fix(table): preserve shared load errors and cover panic waiters
* fix(table): read manifest cache settings from catalog FileIO props
* style(table): gofmt Scan literal in manifest cache PR
---------
Signed-off-by: Codex <[email protected]>
Signed-off-by: Minh Vu <[email protected]>
---
catalog/glue/glue.go | 2 +
catalog/hadoop/hadoop.go | 7 +-
catalog/hadoop/hadoop_test.go | 59 ++
catalog/hive/hive.go | 7 +-
catalog/sql/sql.go | 1 +
table/manifest_content_cache.go | 447 ++++++++++++
table/manifest_content_cache_bench_test.go | 114 +++
table/manifest_content_cache_internal_test.go | 952 ++++++++++++++++++++++++++
table/properties.go | 27 +
table/scanner.go | 19 +-
table/table.go | 71 +-
table/transaction.go | 11 +-
website/src/configuration.md | 28 +-
13 files changed, 1699 insertions(+), 46 deletions(-)
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index 7824a601f..9e554c9c8 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -518,6 +518,7 @@ func (c *Catalog) RegisterTable(ctx context.Context,
identifier table.Identifier
metadataLocation,
io.LoadFSFunc(c.props, metadataLocation),
c,
+ table.WithScanPlanningIOProperties(c.props),
)
if err != nil {
return nil, fmt.Errorf("failed to read table metadata from %s:
%w", metadataLocation, err)
@@ -1120,6 +1121,7 @@ func (c *Catalog) convertGlueToIceberg(ctx
context.Context, glueTable *types.Tab
io.LoadFSFunc(c.props, metadataLocation),
c,
table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
)
if err != nil {
return nil, fmt.Errorf("failed to create iceberg table from
location %s: %w", metadataLocation, err)
diff --git a/catalog/hadoop/hadoop.go b/catalog/hadoop/hadoop.go
index 38ab76bdc..b59ecd2e4 100644
--- a/catalog/hadoop/hadoop.go
+++ b/catalog/hadoop/hadoop.go
@@ -594,6 +594,7 @@ func (c *Catalog) CreateTable(ctx context.Context, ident
table.Identifier, sc *i
io.LoadFSFunc(c.props, metaPath),
c,
table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
)
return tbl, nil
@@ -614,7 +615,11 @@ func (c *Catalog) loadTable(ctx context.Context, ident
table.Identifier) (*table
return nil, 0, fmt.Errorf("failed to initialize metrics
reporter: %w", err)
}
- tbl, err := table.NewFromLocation(ctx, ident, metaPath,
io.LoadFSFunc(c.props, metaPath), c, table.WithMetricsReporter(reporter))
+ tbl, err := table.NewFromLocation(
+ ctx, ident, metaPath, io.LoadFSFunc(c.props, metaPath), c,
+ table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
+ )
if err != nil {
return nil, 0, err
}
diff --git a/catalog/hadoop/hadoop_test.go b/catalog/hadoop/hadoop_test.go
index 3baa3ede0..f3761237c 100644
--- a/catalog/hadoop/hadoop_test.go
+++ b/catalog/hadoop/hadoop_test.go
@@ -2147,6 +2147,65 @@ func (s *HadoopCatalogTestSuite)
TestLoadTableNotExists() {
s.ErrorIs(err, catalog.ErrNoSuchTable)
}
+func (s *HadoopCatalogTestSuite)
TestLoadTableEnablesCatalogManifestContentCache() {
+ ctx := context.Background()
+ cat, err := NewCatalog("test", s.warehouse, iceberg.Properties{
+ table.IOManifestCacheEnabledKey: "true",
+ })
+ s.Require().NoError(err)
+
+ identifier := table.Identifier{"cache-ns", "cache-table"}
+ s.Require().NoError(cat.CreateNamespace(ctx, identifier[:1], nil))
+ created, err := cat.CreateTable(ctx, identifier, s.testSchema())
+ s.Require().NoError(err)
+
+ builder, err := iceberg.NewDataFileBuilder(
+ *iceberg.UnpartitionedSpec, iceberg.EntryContentData,
+ filepath.Join(created.Location(), "sample.parquet"),
iceberg.ParquetFile,
+ nil, nil, nil, 1, 1,
+ )
+ s.Require().NoError(err)
+ txn := created.NewTransaction()
+ s.Require().NoError(txn.AddDataFiles(ctx,
[]iceberg.DataFile{builder.Build()}, nil))
+ _, err = txn.Commit(ctx)
+ s.Require().NoError(err)
+
+ loaded, err := cat.LoadTable(ctx, identifier)
+ s.Require().NoError(err)
+ s.NotContains(loaded.SavedConfig(), table.IOManifestCacheEnabledKey)
+
+ plan := func(tbl *table.Table) error {
+ tasks, err :=
tbl.Scan(table.WithMaxConcurrency(1)).PlanFiles(ctx)
+ if err == nil {
+ s.Len(tasks, 1)
+ }
+
+ return err
+ }
+ s.Require().NoError(plan(loaded))
+
+ fs, err := loaded.FS(ctx)
+ s.Require().NoError(err)
+ manifests, err := loaded.CurrentSnapshot().Manifests(fs)
+ s.Require().NoError(err)
+ s.Require().Len(manifests, 1)
+
+ manifestPath := manifests[0].FilePath()
+ if parsed, parseErr := url.Parse(manifestPath); parseErr == nil &&
parsed.Scheme == "file" {
+ manifestPath = parsed.Path
+ }
+ movedPath := manifestPath + ".temporarily-moved"
+ s.Require().NoError(os.Rename(manifestPath, movedPath))
+ defer func() { s.Require().NoError(os.Rename(movedPath, manifestPath))
}()
+
+ // A fresh handle cannot read the missing manifest, while the warmed
+ // handle can still plan using its cached immutable manifest bytes.
+ cold, err := cat.LoadTable(ctx, identifier)
+ s.Require().NoError(err)
+ s.Require().Error(plan(cold))
+ s.Require().NoError(plan(loaded))
+}
+
func (s *HadoopCatalogTestSuite) TestLoadTablePropagatesMetadataReadError() {
ctx := context.Background()
walkErr := errors.New("metadata walk failed")
diff --git a/catalog/hive/hive.go b/catalog/hive/hive.go
index 58c3f1bed..209247ec9 100644
--- a/catalog/hive/hive.go
+++ b/catalog/hive/hive.go
@@ -195,6 +195,7 @@ func (c *Catalog) LoadTable(ctx context.Context, identifier
table.Identifier) (*
io.LoadFSFunc(c.opts.props, metadataLocation),
c,
table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.opts.props),
)
}
@@ -287,6 +288,7 @@ func (c *Catalog) RegisterTable(ctx context.Context,
identifier table.Identifier
io.LoadFSFunc(c.opts.props, metadataLocation),
c,
table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.opts.props),
)
if err != nil {
return nil, fmt.Errorf("failed to read table metadata from %s:
%w", metadataLocation, err)
@@ -602,7 +604,10 @@ func (c *Catalog) CommitTable(ctx context.Context,
identifier table.Identifier,
if err != nil {
return nil, "", err
}
- current, err = table.NewFromLocation(ctx, identifier,
metadataLoc, io.LoadFSFunc(c.opts.props, metadataLoc), c)
+ current, err = table.NewFromLocation(
+ ctx, identifier, metadataLoc,
io.LoadFSFunc(c.opts.props, metadataLoc), c,
+ table.WithScanPlanningIOProperties(c.opts.props),
+ )
if err != nil {
return nil, "", err
}
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index a7eed9229..c799c19c2 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -1044,6 +1044,7 @@ func (c *Catalog) LoadTable(ctx context.Context,
identifier table.Identifier) (*
io.LoadFSFunc(c.props, result.MetadataLocation.String),
c,
table.WithMetricsReporter(reporter),
+ table.WithScanPlanningIOProperties(c.props),
)
}
diff --git a/table/manifest_content_cache.go b/table/manifest_content_cache.go
new file mode 100644
index 000000000..7079016d1
--- /dev/null
+++ b/table/manifest_content_cache.go
@@ -0,0 +1,447 @@
+// 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 {
+ // A failed population must not make concurrent
scans fail
+ // when the ordinary FileIO path would have
succeeded.
+ file, err := base.Open(location)
+ if err != nil {
+ return nil, fmt.Errorf("open manifest
%s: cache population failed: %w; fallback open failed: %w",
+ location, load.err, err)
+ }
+
+ return file, nil
+ }
+ if 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()
+
+ var content []byte
+ var err error
+ var canceled bool
+ func() {
+ defer func() {
+ if recovered := recover(); recovered != nil {
+ // Always release the single-flight
waiters, even if a
+ // FileIO implementation panics during
population.
+ c.finishLoad(location, expectedLength,
load, nil,
+ fmt.Errorf("manifest content
load panicked: %v", recovered), false)
+ panic(recovered)
+ }
+ c.finishLoad(location, expectedLength, load,
content, err, canceled)
+ }()
+ content, err = readManifestContent(base, location,
expectedLength)
+ canceled = err != nil && ctx.Err() != nil
+ }()
+ if err != nil {
+ if canceled {
+ return nil, ctx.Err()
+ }
+
+ // Preserve the ordinary-open fallback for the
producer, and
+ // explain repeated uncached reads for bad manifest
descriptors.
+ log.Printf("Warning: manifest cache population failed
for %s: %v; falling back to FileIO", location, err)
+ file, fallbackErr := base.Open(location)
+ if fallbackErr != nil {
+ return nil, fmt.Errorf("open manifest %s: cache
population failed: %w; fallback open failed: %w",
+ location, err, fallbackErr)
+ }
+
+ return file, nil
+ }
+
+ 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 := c.now()
+ if c.expired(entry, now) {
+ c.removeEntryLocked(entry)
+
+ return nil, false
+ }
+ entry.lastAccess = now
+ }
+ c.lru.MoveToFront(entry.element)
+
+ return entry.content, true
+}
+
+func (c *manifestContentCache) expired(entry *manifestContentCacheEntry, now
time.Time) bool {
+ // Dividing a monotonic duration avoids overflowing on large configured
+ // millisecond values and is immune to wall-clock jumps.
+ return c.expirationIntervalMs > 0 &&
+ now.Sub(entry.lastAccess)/time.Millisecond >=
time.Duration(c.expirationIntervalMs)
+}
+
+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)
+ }
+ if c.expirationIntervalMs > 0 {
+ now := c.now()
+ // The tail is the least recently accessed entry. Reclaim
expired
+ // entries when inserting, without a background expiration
goroutine.
+ for tail := c.lru.Back(); tail != nil; tail = c.lru.Back() {
+ oldest, ok := tail.Value.(*manifestContentCacheEntry)
+ if !ok {
+ c.lru.Remove(tail)
+
+ continue
+ }
+ if !c.expired(oldest, now) {
+ break
+ }
+ c.removeEntryLocked(oldest)
+ }
+ }
+ for c.totalBytes > c.maxTotalBytes-size && c.lru.Len() > 0 {
+ tail := c.lru.Back()
+ oldest, ok := tail.Value.(*manifestContentCacheEntry)
+ if !ok {
+ c.lru.Remove(tail)
+
+ continue
+ }
+ c.removeEntryLocked(oldest)
+ }
+
+ entry := &manifestContentCacheEntry{
+ location: location,
+ content: content,
+ }
+ if c.expirationIntervalMs > 0 {
+ entry.lastAccess = c.now()
+ }
+ 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[:]); !errors.Is(err, io.EOF) {
+ if err == nil {
+ return nil, fmt.Errorf("manifest content exceeds
expected length %d", expectedLength)
+ }
+
+ return nil, err
+ }
+
+ return content, nil
+}
+
+// Used only while the owning scan goroutine is active. Manifest reads call
+// Open; other optional FileIO interfaces are deliberately not forwarded.
+type manifestContentCacheIO struct {
+ ctx context.Context
+ base iceio.IO
+ cache *manifestContentCache
+ manifest iceberg.ManifestFile
+}
+
+func (m *manifestContentCacheIO) Open(name string) (iceio.File, error) {
+ if name != m.manifest.FilePath() {
+ return m.base.Open(name)
+ }
+
+ return m.cache.open(m.ctx, m.base, m.manifest)
+}
+
+func (m *manifestContentCacheIO) Remove(name string) error {
+ return m.base.Remove(name)
+}
+
+type cachedManifestFile struct {
+ *bytes.Reader
+ location string
+}
+
+func newCachedManifestFile(location string, content []byte)
*cachedManifestFile {
+ return &cachedManifestFile{
+ Reader: bytes.NewReader(content),
+ location: location,
+ }
+}
+
+func (*cachedManifestFile) Close() error { return nil }
+
+func (f *cachedManifestFile) Stat() (fs.FileInfo, error) {
+ return cachedManifestFileInfo{
+ name: path.Base(f.location),
+ size: f.Size(),
+ }, nil
+}
+
+type cachedManifestFileInfo struct {
+ name string
+ size int64
+}
+
+func (f cachedManifestFileInfo) Name() string { return f.name }
+func (f cachedManifestFileInfo) Size() int64 { return f.size }
+func (cachedManifestFileInfo) Mode() fs.FileMode { return 0o444 }
+func (cachedManifestFileInfo) ModTime() time.Time { return time.Time{} }
+func (cachedManifestFileInfo) IsDir() bool { return false }
+func (cachedManifestFileInfo) Sys() any { return nil }
diff --git a/table/manifest_content_cache_bench_test.go
b/table/manifest_content_cache_bench_test.go
new file mode 100644
index 000000000..42e2a9e8a
--- /dev/null
+++ b/table/manifest_content_cache_bench_test.go
@@ -0,0 +1,114 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "context"
+ "fmt"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ iceio "github.com/apache/iceberg-go/io"
+)
+
+var manifestContentCacheBenchmarkSink int
+
+func BenchmarkPlanFilesManifestContentCache(b *testing.B) {
+ for _, manifestCount := range []int{8, 32} {
+ for _, latency := range []time.Duration{0, time.Millisecond} {
+ for _, cached := range []bool{false, true} {
+ cacheName := "without-cache"
+ if cached {
+ cacheName = "with-cache"
+ }
+ b.Run(fmt.Sprintf("manifests=%d/latency=%s/%s",
manifestCount, latency, cacheName), func(b *testing.B) {
+ tbl :=
newRowLimitPlanningBenchmarkTable(b, manifestCount)
+ baseFS, err :=
tbl.fsF(context.Background())
+ if err != nil {
+ b.Fatal(err)
+ }
+ fs :=
&manifestContentCacheBenchmarkIO{IO: baseFS, latency: latency}
+ tbl.fsF = testFSF(fs)
+ if cached {
+ tbl.manifestContentCache =
newManifestContentCache(0, 100<<20, 8<<20)
+ } else {
+ tbl.manifestContentCache = nil
+ }
+
+ scan := tbl.Scan(WithMaxConcurrency(1))
+ tasks, err :=
scan.PlanFiles(context.Background())
+ if err != nil {
+ b.Fatal(err)
+ }
+ if len(tasks) != manifestCount {
+ b.Fatalf("planned %d tasks,
want %d", len(tasks), manifestCount)
+ }
+ fs.reset()
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for b.Loop() {
+ tasks, err :=
scan.PlanFiles(context.Background())
+ if err != nil {
+ b.Fatal(err)
+ }
+
manifestContentCacheBenchmarkSink = len(tasks)
+ }
+ b.StopTimer()
+
b.ReportMetric(float64(fs.count())/float64(b.N), "manifest-opens/op")
+ })
+ }
+ }
+ }
+}
+
+type manifestContentCacheBenchmarkIO struct {
+ iceio.IO
+ latency time.Duration
+
+ mu sync.Mutex
+ opens int64
+}
+
+func (fs *manifestContentCacheBenchmarkIO) Open(name string) (iceio.File,
error) {
+ if strings.Contains(name, "/manifest-") {
+ if fs.latency > 0 {
+ time.Sleep(fs.latency)
+ }
+ fs.mu.Lock()
+ fs.opens++
+ fs.mu.Unlock()
+ }
+
+ return fs.IO.Open(name)
+}
+
+func (fs *manifestContentCacheBenchmarkIO) reset() {
+ fs.mu.Lock()
+ fs.opens = 0
+ fs.mu.Unlock()
+}
+
+func (fs *manifestContentCacheBenchmarkIO) count() int64 {
+ fs.mu.Lock()
+ defer fs.mu.Unlock()
+
+ return fs.opens
+}
diff --git a/table/manifest_content_cache_internal_test.go
b/table/manifest_content_cache_internal_test.go
new file mode 100644
index 000000000..f1abe524f
--- /dev/null
+++ b/table/manifest_content_cache_internal_test.go
@@ -0,0 +1,952 @@
+// 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 TestTransactionScansAndCommitReuseManifestContent(t *testing.T) {
+ fs := newTrackingCallsIO()
+ tbl, _, manifestPath := newManifestContentCacheTestTable(t, fs,
+ iceberg.Properties{IOManifestCacheEnabledKey: "true"})
+ tbl.cat = &committingLabelCatalog{metadata: tbl.metadata}
+ tasks, err := tbl.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ txn := tbl.NewTransaction()
+ require.NoError(t, txn.SetProperties(iceberg.Properties{"updated":
"true"}))
+ scan, err := txn.Scan(WithMaxConcurrency(1))
+ require.NoError(t, err)
+ tasks, err = scan.PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ committed, err := txn.Commit(t.Context())
+ require.NoError(t, err)
+ tasks, err =
committed.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+
+ assert.Equal(t, 1, fs.openCount[manifestPath],
+ "transaction and committed scans must reuse the warm immutable
manifests")
+ assert.Equal(t, "true", committed.Properties()["updated"])
+ assert.NotContains(t, tbl.Properties(), "updated", "metadata stays
independent of cached bytes")
+}
+
+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)
+
+ now := time.Now()
+ cache.now = func() time.Time { return now }
+ readCachedManifest(t, cache, fs, manifest)
+ now = now.Add(time.Hour)
+ 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)
+}
+
+// Java's manifest-content cache uses zero for no time-based expiration,
+// unlike the separate catalog table cache, where zero disables caching.
+func TestManifestContentCacheZeroExpirationKeepsWarmEntries(t *testing.T) {
+ const location = "mem://manifest-content-cache/no-expiration.avro"
+ content := []byte("manifest")
+ fs := newTrackingCallsIO()
+ require.NoError(t, fs.WriteFile(location, content))
+ config := iceberg.Properties{
+ IOManifestCacheEnabledKey: "true",
+ IOManifestCacheExpirationIntervalMsKey: "0",
+ }
+ cache := newManifestContentCacheForConfig(config)
+ require.NotNil(t, cache)
+
+ now := time.Now()
+ cache.now = func() time.Time { return now }
+ manifest := iceberg.NewManifestFile(2, location, int64(len(content)),
0, 1).Build()
+ readCachedManifest(t, cache, fs, manifest)
+ now = now.Add(72 * time.Hour)
+ readCachedManifest(t, cache, fs, manifest)
+
+ assert.Equal(t, 1, fs.openCount[location])
+ assert.Equal(t, int64(len(content)), cache.totalBytes)
+}
+
+func TestManifestContentCacheRejectsNegativeExpiration(t *testing.T) {
+ config := iceberg.Properties{
+ IOManifestCacheEnabledKey: "true",
+ IOManifestCacheExpirationIntervalMsKey: "-1",
+ }
+ require.Nil(t, newManifestContentCacheForConfig(config))
+}
+
+func TestManifestContentCacheSharesInFlightRead(t *testing.T) {
+ const manifestPath = "mem://manifest-content-cache/shared.avro"
+ content := []byte("manifest")
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(manifestPath, content))
+
+ fs := &blockingSnapshotManifestIO{
+ IO: base,
+ blockedPath: manifestPath,
+ started: make(chan struct{}),
+ release: make(chan struct{}),
+ opens: make(map[string]int),
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() { close(fs.release) }) })
+
+ manifest := iceberg.NewManifestFile(2, manifestPath,
int64(len(content)), 0, 1).Build()
+ cache := newManifestContentCache(0, 1<<20, 1<<20)
+ const callers = 16
+ errs := make(chan error, callers)
+ joined := make(chan struct{}, callers)
+ for range callers {
+ go func() {
+ // Only followers call Done() while awaiting the
blocked load.
+ ctx := &countingContext{Context: t.Context(), entered:
joined}
+ file, err := cache.open(ctx, fs, manifest)
+ if err == nil {
+ _, err = io.ReadAll(file)
+ if closeErr := file.Close(); err == nil {
+ err = closeErr
+ }
+ }
+ errs <- err
+ }()
+ }
+
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("manifest read did not start")
+ }
+ // All followers must enter the shared-load wait path. Otherwise late
+ // callers could hit the populated cache and mask broken single-flight.
+ for range callers - 1 {
+ select {
+ case <-joined:
+ case <-time.After(5 * time.Second):
+ t.Fatal("not all callers joined the in-flight manifest
read")
+ }
+ }
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ waiterCtx := ¬ifyingContext{Context: ctx, entered: make(chan
struct{})}
+ waiterErr := make(chan error, 1)
+ go func() {
+ file, err := cache.open(waiterCtx, fs, manifest)
+ if file != nil {
+ file.Close()
+ }
+ waiterErr <- err
+ }()
+ select {
+ case <-waiterCtx.entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join the manifest read")
+ }
+ cancel()
+ select {
+ case err := <-waiterErr:
+ require.ErrorIs(t, err, context.Canceled)
+ case <-time.After(5 * time.Second):
+ t.Fatal("canceled waiter did not return")
+ }
+ release.Do(func() { close(fs.release) })
+
+ for range callers {
+ require.NoError(t, <-errs)
+ }
+
+ fs.mu.Lock()
+ openCount := fs.opens[manifestPath]
+ fs.mu.Unlock()
+ assert.Equal(t, 1, openCount, "concurrent misses should share one
backend read")
+}
+
+func TestManifestContentCacheInFlightLengthMismatch(t *testing.T) {
+ const location = "mem://manifest-content-cache/inflight-length.avro"
+ content := []byte("manifest")
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(location, content))
+ fs := &blockingSnapshotManifestIO{
+ IO: base, blockedPath: location, started: make(chan struct{}),
+ release: make(chan struct{}), opens: make(map[string]int),
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() { close(fs.release) }) })
+
+ cache := newManifestContentCache(0, 1024, 1024)
+ actual := iceberg.NewManifestFile(2, location, int64(len(content)), 0,
1).Build()
+ wrongLength := iceberg.NewManifestFile(2, location,
int64(len(content)-1), 0, 1).Build()
+ type result struct {
+ content []byte
+ err error
+ }
+ read := func(ctx context.Context, manifest iceberg.ManifestFile) <-chan
result {
+ out := make(chan result, 1)
+ go func() {
+ file, err := cache.open(ctx, fs, manifest)
+ if err != nil {
+ out <- result{err: err}
+
+ return
+ }
+ got, err := io.ReadAll(file)
+ if closeErr := file.Close(); err == nil {
+ err = closeErr
+ }
+ out <- result{content: got, err: err}
+ }()
+
+ return out
+ }
+
+ producer := read(t.Context(), actual)
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("manifest read did not start")
+ }
+
+ waiterCtx := ¬ifyingContext{Context: t.Context(), entered: make(chan
struct{})}
+ waiter := read(waiterCtx, wrongLength)
+ select {
+ case <-waiterCtx.entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join the manifest read")
+ }
+ release.Do(func() { close(fs.release) })
+
+ for _, got := range []result{<-producer, <-waiter} {
+ require.NoError(t, got.err)
+ assert.Equal(t, content, got.content)
+ }
+ fs.mu.Lock()
+ opens := fs.opens[location]
+ fs.mu.Unlock()
+ assert.Equal(t, 2, opens, "a descriptor with a different length must
bypass in-flight bytes")
+}
+
+func newManifestContentCacheTestTable(
+ t testing.TB,
+ fs *trackingCallsIO,
+ cacheProperties iceberg.Properties,
+) (*Table, string, string) {
+ t.Helper()
+
+ const tableLocation = "mem://manifest-content-cache"
+ meta, err := NewMetadata(simpleSchema(), iceberg.UnpartitionedSpec,
UnsortedSortOrder,
+ tableLocation, iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+
+ const snapshotID = int64(1)
+ manifestPath := tableLocation + "/metadata/manifest.avro"
+ manifest := writeManifest(t, fs.trackingIO, snapshotID, 1, manifestPath,
+ tableLocation+"/data/file.parquet")
+ manifestListPath := tableLocation + "/metadata/snap.avro"
+ writeManifestList(t, fs.trackingIO, snapshotID, manifestListPath,
[]iceberg.ManifestFile{manifest})
+
+ schemaID := meta.CurrentSchema().ID
+ require.NoError(t, builder.AddSnapshot(&Snapshot{
+ SnapshotID: snapshotID,
+ SequenceNumber: 1,
+ TimestampMs: meta.LastUpdatedMillis() + 1,
+ ManifestList: manifestListPath,
+ Summary: &Summary{Operation: OpAppend},
+ SchemaID: &schemaID,
+ }))
+ require.NoError(t, builder.SetSnapshotRef(MainBranch, snapshotID,
BranchRef))
+ built, err := builder.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "manifest-content-cache"}, built,
tableLocation+"/metadata/metadata.json",
+ testFSF(fs), nil, WithSavedConfig(cacheProperties))
+
+ return tbl, manifestListPath, manifestPath
+}
+
+func readCachedManifest(
+ t testing.TB,
+ cache *manifestContentCache,
+ fs iceio.IO,
+ manifest iceberg.ManifestFile,
+) {
+ t.Helper()
+
+ file, err := cache.open(context.Background(), fs, manifest)
+ require.NoError(t, err)
+ _, err = io.ReadAll(file)
+ require.NoError(t, err)
+ require.NoError(t, file.Close())
+}
+
+func TestManifestContentCacheReuseAcrossCompatibleConfig(t *testing.T) {
+ config := iceberg.Properties{IOManifestCacheEnabledKey: "true"}
+ previous := newManifestContentCacheForConfig(config)
+ require.Same(t, previous, reuseManifestContentCache(previous,
newManifestContentCacheForConfig(config)))
+
+ changed := iceberg.Properties{IOManifestCacheEnabledKey: "true",
IOManifestCacheMaxTotalBytesKey: "12345"}
+ require.NotSame(t, previous, reuseManifestContentCache(previous,
newManifestContentCacheForConfig(changed)))
+ require.Nil(t, reuseManifestContentCache(previous,
newManifestContentCacheForConfig(nil)))
+}
+
+func TestManifestContentCacheUsesCatalogIOProperties(t *testing.T) {
+ fs := newTrackingCallsIO()
+ base, _, manifestPath := newManifestContentCacheTestTable(t, fs, nil)
+ require.Nil(t, base.manifestContentCache)
+
+ catalogProps := iceberg.Properties{
+ IOManifestCacheEnabledKey: "true",
+ IOManifestCacheMaxTotalBytesKey: "1048576",
+ }
+ loaded := New(base.identifier, base.metadata, base.metadataLocation,
+ base.fsF, nil, WithScanPlanningIOProperties(catalogProps))
+ require.NotNil(t, loaded.manifestContentCache)
+ require.Equal(t, int64(1048576),
loaded.manifestContentCache.maxTotalBytes)
+ require.Empty(t, loaded.SavedConfig(), "catalog FileIO settings must
not become saved table config")
+
+ for range 2 {
+ tasks, err :=
loaded.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ }
+ assert.Equal(t, 1, fs.openCount[manifestPath], "catalog-configured
cache must reuse manifest bytes")
+
+ disabled := New(base.identifier, base.metadata, base.metadataLocation,
+ base.fsF, nil,
+ WithScanPlanningIOProperties(catalogProps),
+ WithSavedConfig(iceberg.Properties{IOManifestCacheEnabledKey:
"false"}))
+ require.Nil(t, disabled.manifestContentCache, "saved table settings
must override catalog settings")
+}
+
+func TestManifestContentCacheInvalidNumericConfig(t *testing.T) {
+ for _, key := range []string{
+ IOManifestCacheExpirationIntervalMsKey,
+ IOManifestCacheMaxTotalBytesKey,
+ IOManifestCacheMaxContentLengthKey,
+ } {
+ badValues := []string{"bad", "-1"}
+ if key != IOManifestCacheExpirationIntervalMsKey {
+ badValues = append(badValues, "0")
+ }
+ for _, bad := range badValues {
+ t.Run(key+"/"+bad, func(t *testing.T) {
+ config :=
iceberg.Properties{IOManifestCacheEnabledKey: "true", key: bad}
+ assert.Nil(t,
newManifestContentCacheForConfig(config))
+ })
+ }
+ }
+}
+
+type manifestContentCacheFailOnceIO struct {
+ iceio.IO
+ failed bool
+}
+
+func (f *manifestContentCacheFailOnceIO) Open(name string) (iceio.File, error)
{
+ if !f.failed {
+ f.failed = true
+
+ return nil, io.ErrUnexpectedEOF
+ }
+
+ return f.IO.Open(name)
+}
+
+func TestManifestContentCacheFailedPopulationFallsBackWithoutCaching(t
*testing.T) {
+ const location = "mem://manifest-content-cache/fallback.avro"
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(location, []byte("test")))
+ fs := &manifestContentCacheFailOnceIO{IO: base}
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2, location, 4, 0, 1).Build()
+
+ readCachedManifest(t, cache, fs, manifest)
+ assert.Empty(t, cache.entries, "ordinary-open fallback must not
populate the cache")
+ readCachedManifest(t, cache, fs, manifest)
+ assert.Contains(t, cache.entries, location)
+}
+
+type manifestContentCacheOpenErrorsIO struct {
+ iceio.IO
+ openErrors []error
+}
+
+func (f *manifestContentCacheOpenErrorsIO) Open(string) (iceio.File, error) {
+ err := f.openErrors[0]
+ f.openErrors = f.openErrors[1:]
+
+ return nil, err
+}
+
+func TestManifestContentCachePreservesPopulationAndFallbackErrors(t
*testing.T) {
+ fs := &manifestContentCacheOpenErrorsIO{
+ openErrors: []error{io.ErrUnexpectedEOF, io.ErrClosedPipe},
+ }
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2,
"mem://manifest-content-cache/errors.avro", 4, 0, 1).Build()
+
+ file, err := cache.open(t.Context(), fs, manifest)
+ assert.Nil(t, file)
+ assert.ErrorIs(t, err, io.ErrUnexpectedEOF, "retain the original
population failure")
+ assert.ErrorIs(t, err, io.ErrClosedPipe, "retain the fallback failure")
+ assert.Empty(t, fs.openErrors, "only the producer's single fallback
should run")
+ assert.Empty(t, cache.entries)
+ assert.Empty(t, cache.loads)
+}
+
+type manifestContentCachePanicIO struct {
+ iceio.IO
+}
+
+func (manifestContentCachePanicIO) Open(string) (iceio.File, error) {
+ panic("unexpected backend panic")
+}
+
+func TestManifestContentCachePanicReleasesInFlightLoad(t *testing.T) {
+ cache := newManifestContentCache(0, 16, 16)
+ const location = "mem://manifest-content-cache/panic.avro"
+ manifest := iceberg.NewManifestFile(2, location, 4, 0, 1).Build()
+ assert.Panics(t, func() {
+ _, _ = cache.open(t.Context(), manifestContentCachePanicIO{},
manifest)
+ })
+ assert.Empty(t, cache.loads)
+}
+
+func TestTableRefreshKeepsWarmManifestCacheForMatchingConfig(t *testing.T) {
+ config := iceberg.Properties{IOManifestCacheEnabledKey: "true"}
+ fs := newTrackingCallsIO()
+ tbl, _, manifestPath := newManifestContentCacheTestTable(t, fs, config)
+ tasks, err := tbl.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ require.Equal(t, 1, fs.openCount[manifestPath])
+ oldCache := tbl.manifestContentCache
+
+ newTable := func(config iceberg.Properties) *Table {
+ return New(tbl.identifier, tbl.metadata, tbl.metadataLocation,
+ tbl.fsF, nil, WithSavedConfig(config))
+ }
+ cat := &snapshotManifestRefreshCatalog{fresh: newTable(config)}
+ tbl.cat = cat
+
+ require.NoError(t, tbl.Refresh(t.Context()))
+ require.Same(t, oldCache, tbl.manifestContentCache)
+ tasks, err = tbl.Scan(WithMaxConcurrency(1)).PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ assert.Equal(t, 1, fs.openCount[manifestPath], "refresh must reuse
immutable manifest content")
+
+ changed := iceberg.Properties{
+ IOManifestCacheEnabledKey: "true",
+ IOManifestCacheMaxContentLengthKey: "1",
+ }
+ cat.fresh = newTable(changed)
+ require.NoError(t, tbl.Refresh(t.Context()))
+ require.NotSame(t, oldCache, tbl.manifestContentCache)
+ 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], "changed limits must
bypass the old content cache")
+
+ cat.fresh = newTable(nil)
+ require.NoError(t, tbl.Refresh(t.Context()))
+ require.Nil(t, tbl.manifestContentCache)
+}
+
+func TestManifestContentCacheReclaimsExpiredEntriesOnInsert(t *testing.T) {
+ fs := newTrackingCallsIO()
+ const (
+ firstPath = "mem://manifest-content-cache/expired-1.avro"
+ secondPath = "mem://manifest-content-cache/expired-2.avro"
+ thirdPath = "mem://manifest-content-cache/new.avro"
+ )
+ for _, name := range []string{firstPath, secondPath, thirdPath} {
+ require.NoError(t, fs.WriteFile(name, []byte("test")))
+ }
+ cache := newManifestContentCache(1000, 16, 4)
+ now := time.Now()
+ cache.now = func() time.Time { return now }
+ for _, name := range []string{firstPath, secondPath} {
+ readCachedManifest(t, cache, fs, iceberg.NewManifestFile(2,
name, 4, 0, 1).Build())
+ }
+ require.Equal(t, int64(8), cache.totalBytes)
+
+ now = now.Add(2 * time.Second)
+ readCachedManifest(t, cache, fs, iceberg.NewManifestFile(2, thirdPath,
4, 0, 1).Build())
+ assert.NotContains(t, cache.entries, firstPath)
+ assert.NotContains(t, cache.entries, secondPath)
+ assert.Contains(t, cache.entries, thirdPath)
+ assert.Equal(t, int64(4), cache.totalBytes)
+}
+
+type manifestContentFailingBlockedIO struct {
+ iceio.IO
+ started chan struct{}
+ release chan struct{}
+ firstReadIsShort bool
+ firstOpenPanics bool
+ fallbackError error
+
+ mu sync.Mutex
+ opens int
+}
+
+func (f *manifestContentFailingBlockedIO) Open(name string) (iceio.File,
error) {
+ f.mu.Lock()
+ f.opens++
+ attempt := f.opens
+ f.mu.Unlock()
+ if attempt == 1 {
+ close(f.started)
+ <-f.release
+ if f.firstOpenPanics {
+ panic("unexpected backend panic")
+ }
+ if !f.firstReadIsShort {
+ return nil, io.ErrUnexpectedEOF
+ }
+ }
+ if f.fallbackError != nil {
+ return nil, f.fallbackError
+ }
+
+ return f.IO.Open(name)
+}
+
+// Failed population is a cache optimization failure, not a scan failure.
+// Every in-flight caller must still be able to fall back to ordinary FileIO.
+func TestManifestContentCacheFailedSharedLoadFallsBackForAllCallers(t
*testing.T) {
+ for _, tc := range []struct {
+ name string
+ expectedLength int64
+ firstReadIsShort bool
+ }{
+ {name: "backend error", expectedLength: 4},
+ {name: "manifest length mismatch", expectedLength: 5,
firstReadIsShort: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ const path =
"mem://manifest-content-cache/failed-shared.avro"
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(path, []byte("test")))
+ fs := &manifestContentFailingBlockedIO{
+ IO: base, started: make(chan struct{}),
release: make(chan struct{}),
+ firstReadIsShort: tc.firstReadIsShort,
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() {
close(fs.release) }) })
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2, path,
tc.expectedLength, 0, 1).Build()
+
+ producerErr := make(chan error, 1)
+ go func() {
+ file, err := cache.open(t.Context(), fs,
manifest)
+ if err == nil {
+ _, err = io.ReadAll(file)
+ if closeErr := file.Close(); err == nil
{
+ err = closeErr
+ }
+ }
+ producerErr <- err
+ }()
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("cache producer did not start")
+ }
+
+ const followers = 8
+ entered := make(chan struct{}, followers)
+ waiterErrors := make(chan error, followers)
+ for range followers {
+ go func() {
+ ctx := &countingContext{Context:
t.Context(), entered: entered}
+ file, err := cache.open(ctx, fs,
manifest)
+ if err == nil {
+ _, err = io.ReadAll(file)
+ if closeErr := file.Close();
err == nil {
+ err = closeErr
+ }
+ }
+ waiterErrors <- err
+ }()
+ }
+ for range followers {
+ select {
+ case <-entered:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join failed
shared load")
+ }
+ }
+ release.Do(func() { close(fs.release) })
+ require.NoError(t, <-producerErr, "the producer should
use one ordinary-open fallback")
+ for range followers {
+ require.NoError(t, <-waiterErrors, "every
waiter should use its own ordinary-open fallback")
+ }
+ fs.mu.Lock()
+ opens := fs.opens
+ fs.mu.Unlock()
+ assert.Equal(t, 2+followers, opens, "one failed
population plus one ordinary open per caller")
+ assert.Empty(t, cache.entries, "failed population must
not retain fallback bytes")
+
+ readCachedManifest(t, cache, fs, manifest)
+ if tc.firstReadIsShort {
+ assert.Empty(t, cache.entries, "a wrong
manifest length must never be cached")
+ } else {
+ assert.Contains(t, cache.entries, path, "a
later scan may populate the cache")
+ }
+ })
+ }
+}
+
+func TestManifestContentCachePanickingProducerUnblocksWaiter(t *testing.T) {
+ const location = "mem://manifest-content-cache/panicking-producer.avro"
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(location, []byte("test")))
+ fs := &manifestContentFailingBlockedIO{
+ IO: base, started: make(chan struct{}), release: make(chan
struct{}),
+ firstOpenPanics: true,
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() { close(fs.release) }) })
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2, location, 4, 0, 1).Build()
+
+ producerPanic := make(chan any, 1)
+ go func() {
+ defer func() { producerPanic <- recover() }()
+ _, _ = cache.open(t.Context(), fs, manifest)
+ }()
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("panicking producer did not start")
+ }
+
+ joined := make(chan struct{}, 1)
+ waiterErr := make(chan error, 1)
+ go func() {
+ ctx := &countingContext{Context: t.Context(), entered: joined}
+ file, err := cache.open(ctx, fs, manifest)
+ if err == nil {
+ _, err = io.ReadAll(file)
+ if closeErr := file.Close(); err == nil {
+ err = closeErr
+ }
+ }
+ waiterErr <- err
+ }()
+ select {
+ case <-joined:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join the panicking producer")
+ }
+ release.Do(func() { close(fs.release) })
+ select {
+ case got := <-producerPanic:
+ assert.Equal(t, "unexpected backend panic", got)
+ case <-time.After(5 * time.Second):
+ t.Fatal("panicking producer did not finish")
+ }
+ select {
+ case err := <-waiterErr:
+ require.NoError(t, err, "waiter must fall back after a producer
panics")
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter remained blocked after producer panic")
+ }
+ fs.mu.Lock()
+ opens := fs.opens
+ fs.mu.Unlock()
+ assert.Equal(t, 2, opens, "one panicking population and one waiter
fallback")
+ assert.Empty(t, cache.loads)
+ assert.Empty(t, cache.entries)
+ readCachedManifest(t, cache, fs, manifest)
+ assert.Contains(t, cache.entries, location, "a later scan can fill the
cache")
+}
+
+func TestManifestContentCacheFailedSharedLoadPreservesBothErrors(t *testing.T)
{
+ const location = "mem://manifest-content-cache/fallback-errors.avro"
+ fs := &manifestContentFailingBlockedIO{
+ IO: iceio.NewMemFS(), started: make(chan struct{}), release:
make(chan struct{}),
+ fallbackError: io.ErrClosedPipe,
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() { close(fs.release) }) })
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2, location, 4, 0, 1).Build()
+
+ producerErr := make(chan error, 1)
+ go func() {
+ file, err := cache.open(t.Context(), fs, manifest)
+ if file != nil {
+ _ = file.Close()
+ }
+ producerErr <- err
+ }()
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("producer did not start")
+ }
+ joined := make(chan struct{}, 1)
+ waiterErr := make(chan error, 1)
+ go func() {
+ ctx := &countingContext{Context: t.Context(), entered: joined}
+ file, err := cache.open(ctx, fs, manifest)
+ if file != nil {
+ _ = file.Close()
+ }
+ waiterErr <- err
+ }()
+ select {
+ case <-joined:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join failed population")
+ }
+ release.Do(func() { close(fs.release) })
+
+ for _, result := range []struct {
+ name string
+ errs <-chan error
+ }{
+ {"producer", producerErr},
+ {"waiter", waiterErr},
+ } {
+ select {
+ case err := <-result.errs:
+ require.ErrorIs(t, err, io.ErrUnexpectedEOF,
result.name)
+ require.ErrorIs(t, err, io.ErrClosedPipe, result.name)
+ assert.Contains(t, err.Error(), location, result.name)
+ case <-time.After(5 * time.Second):
+ t.Fatalf("%s did not finish", result.name)
+ }
+ }
+ fs.mu.Lock()
+ opens := fs.opens
+ fs.mu.Unlock()
+ assert.Equal(t, 3, opens, "failed population plus one fallback per
caller")
+ assert.Empty(t, cache.loads)
+ assert.Empty(t, cache.entries)
+}
+
+func TestManifestContentCacheRetriesAfterProducerCancellation(t *testing.T) {
+ const path = "mem://manifest-content-cache/canceled-producer.avro"
+ base := iceio.NewMemFS()
+ require.NoError(t, base.WriteFile(path, []byte("test")))
+ fs := &manifestContentFailingBlockedIO{
+ IO: base, started: make(chan struct{}), release: make(chan
struct{}),
+ }
+ var release sync.Once
+ t.Cleanup(func() { release.Do(func() { close(fs.release) }) })
+ cache := newManifestContentCache(0, 16, 16)
+ manifest := iceberg.NewManifestFile(2, path, 4, 0, 1).Build()
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ producerErr := make(chan error, 1)
+ go func() {
+ file, err := cache.open(ctx, fs, manifest)
+ if file != nil {
+ _ = file.Close()
+ }
+ producerErr <- err
+ }()
+ select {
+ case <-fs.started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("cache producer did not start")
+ }
+
+ joined := make(chan struct{}, 1)
+ waiterErr := make(chan error, 1)
+ go func() {
+ waiterCtx := &countingContext{Context: t.Context(), entered:
joined}
+ file, err := cache.open(waiterCtx, fs, manifest)
+ if err == nil {
+ _, err = io.ReadAll(file)
+ if closeErr := file.Close(); err == nil {
+ err = closeErr
+ }
+ }
+ waiterErr <- err
+ }()
+ select {
+ case <-joined:
+ case <-time.After(5 * time.Second):
+ t.Fatal("waiter did not join canceled producer")
+ }
+ cancel()
+ release.Do(func() { close(fs.release) })
+ require.ErrorIs(t, <-producerErr, context.Canceled)
+ require.NoError(t, <-waiterErr)
+ fs.mu.Lock()
+ opens := fs.opens
+ fs.mu.Unlock()
+ assert.Equal(t, 2, opens, "a live waiter should retry the canceled
producer just once")
+ assert.Contains(t, cache.entries, path)
+}
diff --git a/table/properties.go b/table/properties.go
index 0743e660f..f2ba0cb59 100644
--- a/table/properties.go
+++ b/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 come from catalog/FileIO properties;
+ // saved per-table configuration overrides matching catalog keys.
+ IOManifestCacheEnabledKey = "io.manifest.cache-enabled"
+ IOManifestCacheEnabledDefault = false
+
+ // IOManifestCacheExpirationIntervalMsKey controls how long an unused
cached
+ // manifest remains eligible for reuse. Expired bytes are removed
lazily on
+ // cache access or insertion. Zero disables time-based expiration,
matching
+ // Java Iceberg's ContentCache. Invalid negative values warn and disable
+ // the Go cache rather than throwing a Java ValidationException.
+ IOManifestCacheExpirationIntervalMsKey =
"io.manifest.cache.expiration-interval-ms"
+ IOManifestCacheExpirationIntervalMsDefault = 60 * 1000
+
+ // IOManifestCacheMaxTotalBytesKey bounds total retained manifest
content per
+ // independently loaded table. Commit-derived handles share the same
budget.
+ // Non-positive or unparsable values warn and disable the Go cache,
whereas
+ // Java Iceberg rejects them.
+ IOManifestCacheMaxTotalBytesKey =
"io.manifest.cache.max-total-bytes"
+ IOManifestCacheMaxTotalBytesDefault = 100 * 1024 * 1024
+
+ // IOManifestCacheMaxContentLengthKey skips caching individual manifests
+ // larger than this size. Non-positive or unparsable values warn and
disable
+ // the Go cache, whereas Java Iceberg rejects them.
+ IOManifestCacheMaxContentLengthKey =
"io.manifest.cache.max-content-length"
+ IOManifestCacheMaxContentLengthDefault = 8 * 1024 * 1024
+
// These Java-compatible properties are reserved for future task-group
// planning. They are exported so applications can share configuration
// names across Iceberg implementations.
diff --git a/table/scanner.go b/table/scanner.go
index 2bef72fed..5109c914e 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -544,14 +544,15 @@ func classifyDataFile(f iceberg.DataFile) (dataFileKind,
error) {
// close it when they are done, including early exits after remote planning
// succeeds but before all records are consumed.
type Scan struct {
- identifier Identifier
- metadata Metadata
- metadataLocation string
- ioF FSysF
- manifestCache *snapshotManifestCache
- planner ScanPlanner
- scanPlanningIOProps iceberg.Properties
- planningMode ScanPlanningMode
+ identifier Identifier
+ metadata Metadata
+ metadataLocation string
+ ioF FSysF
+ manifestCache *snapshotManifestCache
+ manifestContentCache *manifestContentCache
+ planner ScanPlanner
+ scanPlanningIOProps iceberg.Properties
+ planningMode ScanPlanningMode
// planningModeSet records an explicit WithScanPlanningMode, so a mode
that
// conflicts with the catalog directive fails instead of being
overridden.
planningModeSet bool
@@ -1298,6 +1299,7 @@ func (scan *Scan)
collectManifestEntriesWithSchemaOptionsAndMinSequenceNum(
if err != nil {
return err
}
+ fs = scan.manifestContentCache.wrap(gctx, fs, mf)
partEval, err :=
partitionEvaluators.Get(int(mf.PartitionSpecID()))
if err != nil {
return fmt.Errorf("failed to build partition
evaluator for spec %d: %w", mf.PartitionSpecID(), err)
@@ -1453,6 +1455,7 @@ func (scan *Scan) planDataManifestTasksWithOptions(
if err != nil {
return err
}
+ fs = scan.manifestContentCache.wrap(gctx, fs, manifest)
partEval, err :=
partitionEvaluators.Get(int(manifest.PartitionSpecID()))
if err != nil {
return fmt.Errorf("failed to build partition
evaluator for spec %d: %w", manifest.PartitionSpecID(), err)
diff --git a/table/table.go b/table/table.go
index 085a245f2..b2c12d29c 100644
--- a/table/table.go
+++ b/table/table.go
@@ -95,13 +95,14 @@ type CatalogIO interface {
}
type Table struct {
- identifier Identifier
- metadata Metadata
- metadataLocation string
- cat CatalogIO
- fsF FSysF
- manifestCache *snapshotManifestCache
- planner ScanPlanner
+ identifier Identifier
+ metadata Metadata
+ metadataLocation string
+ cat CatalogIO
+ fsF FSysF
+ manifestCache *snapshotManifestCache
+ manifestContentCache *manifestContentCache
+ planner ScanPlanner
// scanPlanningIOProps are the table-scoped FileIO properties supplied
by a
// catalog load response. They are separate from metadata properties
because
// REST catalogs can return FileIO configuration in the response's
config
@@ -249,6 +250,7 @@ func (t *Table) Refresh(ctx context.Context) error {
t.fsF = fresh.fsF
t.metadataLocation = fresh.metadataLocation
t.manifestCache = newSnapshotManifestCacheForMetadata(fresh.metadata)
+ t.manifestContentCache =
reuseManifestContentCache(t.manifestContentCache, fresh.manifestContentCache)
t.planner = fresh.planner
t.scanPlanningIOProps = maps.Clone(fresh.scanPlanningIOProps)
t.scanPlanningDirective = fresh.scanPlanningDirective
@@ -852,7 +854,7 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
}
}
- return New(
+ next := New(
t.identifier,
newMeta,
newLoc,
@@ -863,7 +865,10 @@ func (t Table) doCommit(ctx context.Context, updates
[]Update, reqs []Requiremen
WithScanPlanningDirective(t.scanPlanningDirective,
t.hasScanPlanningDirective),
WithLabels(t.labels),
WithSavedConfig(t.savedConfig),
- ), nil
+ )
+ next.manifestContentCache =
reuseManifestContentCache(t.manifestContentCache, next.manifestContentCache)
+
+ return next, nil
}
// rewriteRefSnapshotRequirements returns a copy of reqs with every
@@ -1370,22 +1375,23 @@ func WithArrowBatchSize(n int) ScanOption {
func (t Table) Scan(opts ...ScanOption) *Scan {
directive, directiveErr := t.ScanPlanningDirective()
s := &Scan{
- identifier: slices.Clone(t.identifier),
- metadata: t.metadata,
- metadataLocation: t.metadataLocation,
- ioF: t.fsF,
- manifestCache: t.manifestCache,
- planner: t.planner,
- scanPlanningIOProps: maps.Clone(t.scanPlanningIOProps),
- directive: directive,
- directiveErr: directiveErr,
- planningMode: ScanPlanningLocal,
- rowFilter: iceberg.AlwaysTrue{},
- selectedFields: []string{"*"},
- caseSensitive: true,
- limit: ScanNoLimit,
- concurrency: runtime.GOMAXPROCS(0),
- reporter: t.MetricsReporter(),
+ identifier: slices.Clone(t.identifier),
+ metadata: t.metadata,
+ metadataLocation: t.metadataLocation,
+ ioF: t.fsF,
+ manifestCache: t.manifestCache,
+ manifestContentCache: t.manifestContentCache,
+ planner: t.planner,
+ scanPlanningIOProps: maps.Clone(t.scanPlanningIOProps),
+ directive: directive,
+ directiveErr: directiveErr,
+ planningMode: ScanPlanningLocal,
+ rowFilter: iceberg.AlwaysTrue{},
+ selectedFields: []string{"*"},
+ caseSensitive: true,
+ limit: ScanNoLimit,
+ concurrency: runtime.GOMAXPROCS(0),
+ reporter: t.MetricsReporter(),
}
for _, opt := range opts {
@@ -1418,10 +1424,9 @@ func WithMetricsReporter(r metrics.Reporter) Option {
}
}
-// WithScanPlanningIOProperties supplies the table-scoped FileIO properties
-// that a remote scan planner needs when it builds a plan-scoped FileIO. This
-// is intended for catalog implementations whose table-load response carries
-// FileIO configuration separately from table metadata properties.
+// WithScanPlanningIOProperties supplies FileIO/catalog properties for
+// manifest content caching and remote scan planning. These can include
+// settings returned separately from table metadata by a REST catalog.
func WithScanPlanningIOProperties(props iceberg.Properties) Option {
if props == nil {
return noopTableOption
@@ -1499,6 +1504,14 @@ func New(ident Identifier, meta Metadata,
metadataLocation string, fsF FSysF, ca
for _, opt := range opts {
opt(t)
}
+ // FileIO/catalog settings are the default. Per-table saved settings
+ // (including REST LoadTable overrides) take precedence when supplied.
+ cacheConfig := t.savedConfig
+ if len(t.scanPlanningIOProps) > 0 {
+ cacheConfig = maps.Clone(t.scanPlanningIOProps)
+ maps.Copy(cacheConfig, t.savedConfig)
+ }
+ t.manifestContentCache = newManifestContentCacheForConfig(cacheConfig)
return t
}
diff --git a/table/transaction.go b/table/transaction.go
index 9b51b322d..59a6c65df 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -3516,11 +3516,12 @@ func (t *Transaction) Scan(opts ...ScanOption) (*Scan,
error) {
}
s := &Scan{
- identifier: slices.Clone(t.tbl.identifier),
- metadata: updatedMeta,
- metadataLocation: t.tbl.metadataLocation,
- ioF: t.tbl.fsF,
- manifestCache:
newSnapshotManifestCacheForMetadata(updatedMeta),
+ identifier: slices.Clone(t.tbl.identifier),
+ metadata: updatedMeta,
+ metadataLocation: t.tbl.metadataLocation,
+ ioF: t.tbl.fsF,
+ manifestCache:
newSnapshotManifestCacheForMetadata(updatedMeta),
+ manifestContentCache: t.tbl.manifestContentCache,
// Catalog planners can only see committed table state, not
metadata
// staged inside this transaction. Keep transaction scans local
so auto
// mode cannot silently return stale tasks. The catalog's
diff --git a/website/src/configuration.md b/website/src/configuration.md
index 4db0d1d7b..0ec57818c 100644
--- a/website/src/configuration.md
+++ b/website/src/configuration.md
@@ -363,12 +363,36 @@ 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. Read from catalog FileIO properties in REST, Glue, SQL,
Hive and Hadoop; REST table metadata and load-response config may override the
catalog value. |
+| `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. |
+| `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. Non-positive or unparsable values warn
and disable caching. |
+
+The catalog's FileIO properties configure manifest caching for all built-in
+catalogs. With direct `table.New` calls, use `WithScanPlanningIOProperties`
+or `WithSavedConfig`; saved config wins when the same key is present in both.
+Matching expiration interval, total byte limit and per-manifest byte limit
+preserve warm cache entries across refresh and commit. Changing a limit
replaces
+the cache.
+
+Unlike Java, which rejects invalid manifest-cache limits, Go warns and disables
+the content cache for negative or unparsable expiration or
+non-positive/unparsable size limits.
The manifest-list cache retains at most 64 snapshots and 32,768 manifest
descriptors in total per table handle. These are count limits, not byte limits:
memory use also depends on manifest paths, partition summaries, and key
metadata.
-Refreshing the table replaces its cache. This caches decoded manifest lists;
-it does not implement Java's `io.manifest.cache.*` file-content cache settings.
+Refreshing the table replaces its manifest-list cache.
+
+The optional manifest-content cache is separate. It stores immutable raw
manifest
+bytes and still decodes, projects, and filters entries independently for each
+scan. Cache misses for the same manifest share one in-flight read. If
population
+fails, its producer and each existing waiter independently fall back to one
+ordinary FileIO open; if the producer was canceled, live waiters can start
+another shared load. Failed content is not
+retained, so a later scan can retry. Manifests larger than the configured
+per-file limit bypass the cache. The current integration covers local scan
+planning; manifest lists continue to use the decoded manifest-list cache.
The Java-compatible `read.split.planning-lookback` and
`read.split.open-file-cost`
names and defaults are also exported by the Go API for shared configuration,
