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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
