laskoviymishka commented on code in PR #1904:
URL: https://github.com/apache/iceberg-go/pull/1904#discussion_r3874752198
##########
table/conflict_validation.go:
##########
@@ -181,6 +184,184 @@ type conflictContext struct {
// Resolved once; subsequent validator calls reuse the walk.
concurrent []Snapshot
+
+ // Lazily caches raw manifest bytes for this validation attempt. The
+ // cache is intentionally scoped to the context because all validators
+ // inspect the same immutable metadata snapshot, while
ManifestFile.Entries
+ // still performs descriptor-specific inheritance for each logical read.
+ manifestIO *conflictManifestIO
+
+ // Parsed manifest-list descriptors are also shared across validators.
+ manifestLists map[int64]conflictManifestList
+}
+
+type conflictManifestList struct {
+ manifests []iceberg.ManifestFile
+ err error
+}
+
+// conflictManifestIO is a read-through cache used only while conflict
+// validators inspect one conflictContext. Manifest files are immutable after
+// they are committed, so the same path can be safely served from the cached
+// bytes to later validators without reopening the backing object store.
+//
+// The cache stores raw bytes instead of decoded entries. The first read still
+// streams from the backing IO and is recorded only when it reaches EOF, so an
+// early validator exit does not force a full manifest read. Later reads stream
+// decoded entries from the completed bytes and let each ManifestFile
+// descriptor apply its own inheritance metadata.
+type conflictManifestIO struct {
+ base iceio.IO
+ files map[string][]byte
Review Comment:
This flips the peak memory profile and the PR doesn't call it out. The old
`snap.entries` path streamed and GC'd one manifest at a time, so peak manifest
bytes on the heap was O(one file). Here every fully-read manifest stays in
`files` for the life of the context, with no bound or eviction, so peak becomes
O(all manifests read during the attempt).
`validateDataFilesExist` walks every data manifest of the head snapshot, and
under the concurrent-commit load this optimization is meant to help, that's
exactly the case where the footprint grows most: 1000 manifests at 100KB is
100MB held per attempt, times N in-flight attempts.
I'd either document that bound and argue it's acceptable (the manifests were
going to be read anyway; only the simultaneity is new), or add a max-bytes
guard that falls back to uncached reads past a threshold. wdyt?
##########
table/conflict_validation.go:
##########
@@ -181,6 +184,184 @@ type conflictContext struct {
// Resolved once; subsequent validator calls reuse the walk.
concurrent []Snapshot
+
+ // Lazily caches raw manifest bytes for this validation attempt. The
+ // cache is intentionally scoped to the context because all validators
+ // inspect the same immutable metadata snapshot, while
ManifestFile.Entries
+ // still performs descriptor-specific inheritance for each logical read.
+ manifestIO *conflictManifestIO
+
+ // Parsed manifest-list descriptors are also shared across validators.
+ manifestLists map[int64]conflictManifestList
+}
+
+type conflictManifestList struct {
+ manifests []iceberg.ManifestFile
+ err error
+}
+
+// conflictManifestIO is a read-through cache used only while conflict
+// validators inspect one conflictContext. Manifest files are immutable after
+// they are committed, so the same path can be safely served from the cached
+// bytes to later validators without reopening the backing object store.
+//
+// The cache stores raw bytes instead of decoded entries. The first read still
+// streams from the backing IO and is recorded only when it reaches EOF, so an
+// early validator exit does not force a full manifest read. Later reads stream
+// decoded entries from the completed bytes and let each ManifestFile
+// descriptor apply its own inheritance metadata.
+type conflictManifestIO struct {
+ base iceio.IO
+ files map[string][]byte
+}
Review Comment:
There are no compile-time guards that these satisfy the IO/File interfaces,
so if `iceio.File` ever grows a method the break surfaces at the distant `Open`
call site rather than here.
```go
var (
_ iceio.IO = (*conflictManifestIO)(nil)
_ iceio.File = (*conflictManifestFile)(nil)
_ iceio.File = (*conflictManifestRecordingFile)(nil)
)
```
##########
table/conflict_validation.go:
##########
@@ -181,6 +184,184 @@ type conflictContext struct {
// Resolved once; subsequent validator calls reuse the walk.
concurrent []Snapshot
+
+ // Lazily caches raw manifest bytes for this validation attempt. The
+ // cache is intentionally scoped to the context because all validators
+ // inspect the same immutable metadata snapshot, while
ManifestFile.Entries
+ // still performs descriptor-specific inheritance for each logical read.
+ manifestIO *conflictManifestIO
+
+ // Parsed manifest-list descriptors are also shared across validators.
+ manifestLists map[int64]conflictManifestList
+}
+
+type conflictManifestList struct {
+ manifests []iceberg.ManifestFile
+ err error
+}
+
+// conflictManifestIO is a read-through cache used only while conflict
+// validators inspect one conflictContext. Manifest files are immutable after
+// they are committed, so the same path can be safely served from the cached
+// bytes to later validators without reopening the backing object store.
+//
+// The cache stores raw bytes instead of decoded entries. The first read still
+// streams from the backing IO and is recorded only when it reaches EOF, so an
+// early validator exit does not force a full manifest read. Later reads stream
+// decoded entries from the completed bytes and let each ManifestFile
+// descriptor apply its own inheritance metadata.
+type conflictManifestIO struct {
+ base iceio.IO
+ files map[string][]byte
+}
+
+func newConflictManifestIO(base iceio.IO) *conflictManifestIO {
+ return &conflictManifestIO{
+ base: base,
+ files: make(map[string][]byte),
+ }
+}
+
+func (c *conflictManifestIO) Open(name string) (iceio.File, error) {
+ if data, ok := c.files[name]; ok {
+ return newConflictManifestFile(name, data), nil
+ }
+
+ f, err := c.base.Open(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return &conflictManifestRecordingFile{
+ base: f,
+ cache: c,
+ name: name,
+ cacheable: true,
+ }, nil
+}
+
+func (c *conflictManifestIO) Remove(name string) error {
+ delete(c.files, name)
+
+ return c.base.Remove(name)
+}
+
+func (c *conflictManifestIO) Stat(name string) (fs.FileInfo, error) {
+ if data, ok := c.files[name]; ok {
+ return conflictManifestFileInfo{name: name, size:
int64(len(data))}, nil
+ }
+
+ if statIO, ok := c.base.(iceio.StatIO); ok {
+ return statIO.Stat(name)
+ }
+
+ f, err := c.Open(name)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+
+ return f.Stat()
+}
+
+type conflictManifestFile struct {
+ *bytes.Reader
+ name string
+ size int64
+}
+
+func newConflictManifestFile(name string, data []byte) *conflictManifestFile {
+ return &conflictManifestFile{
+ Reader: bytes.NewReader(data),
+ name: name,
+ size: int64(len(data)),
+ }
+}
+
+func (f *conflictManifestFile) Close() error { return nil }
+
+func (f *conflictManifestFile) Stat() (fs.FileInfo, error) {
+ return conflictManifestFileInfo{name: f.name, size: f.size}, nil
+}
+
+type conflictManifestRecordingFile struct {
+ base iceio.File
+ cache *conflictManifestIO
+ name string
+ data []byte
+ cacheable bool
+ complete bool
+}
+
+func (f *conflictManifestRecordingFile) Read(p []byte) (int, error) {
+ n, err := f.base.Read(p)
+ if f.cacheable && n > 0 {
+ f.data = append(f.data, p[:n]...)
+ }
+ if f.cacheable && errors.Is(err, io.EOF) {
Review Comment:
The "record only on EOF so early exits don't cache" contract is derived from
EOF-signal timing, not enforced. The io.Reader contract explicitly permits
returning `(n, io.EOF)` in one call, and if the backing reader ever does that
(or drains the file before the decoder yields entries), `complete` flips true
and an abandoned iteration still caches the bytes.
`TestConflictValidationDoesNotCacheEarlyManifestExit` only passes because
local-FS `Read` happens to return `(n, nil)` then `(0, io.EOF)` separately, so
the test proves less than it claims. A different FS, buffer size, or OS could
silently break the stated invariant. I'd tie completeness to a known size from
`Stat` (or make the decision explicit) rather than to when EOF arrives.
Separately, since the reader contract forbids wrapping `io.EOF`, `errors.Is`
here is just a slower `err == io.EOF` on a per-Read hot path. wdyt?
##########
table/conflict_validation.go:
##########
@@ -181,6 +184,184 @@ type conflictContext struct {
// Resolved once; subsequent validator calls reuse the walk.
concurrent []Snapshot
+
+ // Lazily caches raw manifest bytes for this validation attempt. The
+ // cache is intentionally scoped to the context because all validators
+ // inspect the same immutable metadata snapshot, while
ManifestFile.Entries
+ // still performs descriptor-specific inheritance for each logical read.
+ manifestIO *conflictManifestIO
+
+ // Parsed manifest-list descriptors are also shared across validators.
+ manifestLists map[int64]conflictManifestList
+}
+
+type conflictManifestList struct {
+ manifests []iceberg.ManifestFile
+ err error
+}
+
+// conflictManifestIO is a read-through cache used only while conflict
+// validators inspect one conflictContext. Manifest files are immutable after
+// they are committed, so the same path can be safely served from the cached
+// bytes to later validators without reopening the backing object store.
+//
+// The cache stores raw bytes instead of decoded entries. The first read still
+// streams from the backing IO and is recorded only when it reaches EOF, so an
+// early validator exit does not force a full manifest read. Later reads stream
+// decoded entries from the completed bytes and let each ManifestFile
+// descriptor apply its own inheritance metadata.
+type conflictManifestIO struct {
+ base iceio.IO
+ files map[string][]byte
+}
+
+func newConflictManifestIO(base iceio.IO) *conflictManifestIO {
+ return &conflictManifestIO{
+ base: base,
+ files: make(map[string][]byte),
+ }
+}
+
+func (c *conflictManifestIO) Open(name string) (iceio.File, error) {
+ if data, ok := c.files[name]; ok {
+ return newConflictManifestFile(name, data), nil
+ }
+
+ f, err := c.base.Open(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return &conflictManifestRecordingFile{
+ base: f,
+ cache: c,
+ name: name,
+ cacheable: true,
+ }, nil
+}
+
+func (c *conflictManifestIO) Remove(name string) error {
+ delete(c.files, name)
+
+ return c.base.Remove(name)
Review Comment:
A read-through cache that forwards `Remove` to the backing store can delete
a committed manifest, which is supposed to be immutable. No caller hits it
today (only `Open` is used), but it's a sharp edge the moment this IO gets
wired into anything with a cleanup path.
I'd drop the `c.base.Remove` forwarding and return an unsupported error
instead. Thoughts?
##########
table/conflict_validation.go:
##########
@@ -317,21 +508,30 @@ func validateDataFilesExist(ctx *conflictContext,
referencedPaths []string) erro
return fmt.Errorf("%w: branch %q missing on current metadata",
ErrCommitDiverged, ctx.branch)
}
- for entry, err := range head.entries(ctx.fs,
iceberg.ManifestContentData) {
- if err != nil {
- return fmt.Errorf("iterating data files for current
head %d: %w", head.SnapshotID, err)
- }
- // A DELETED entry means the file was removed (e.g. rewritten
by a
- // concurrent compaction); it is no longer live data a
pos-delete can
- // apply to, so it does not satisfy existence.
- if entry.Status() == iceberg.EntryStatusDELETED {
+ manifests, err := ctx.manifestsFor(*head)
+ if err != nil {
+ return fmt.Errorf("iterating data files for current head %d:
%w", head.SnapshotID, err)
+ }
+ for _, mf := range manifests {
+ if mf.ManifestContent() != iceberg.ManifestContentData {
continue
}
- path := entry.DataFile().FilePath()
- if _, ok := needed[path]; ok {
- delete(needed, path)
- if len(needed) == 0 {
- return nil
+ for entry, err := range mf.Entries(ctx.manifestReadIO(), false)
{
+ if err != nil {
+ return fmt.Errorf("iterating data files for
current head %d: %w", head.SnapshotID, err)
+ }
+ // A DELETED entry means the file was removed (e.g.
rewritten by a
+ // concurrent compaction); it is no longer live data a
pos-delete can
+ // apply to, so it does not satisfy existence.
+ if entry.Status() == iceberg.EntryStatusDELETED {
+ continue
+ }
+ path := entry.DataFile().FilePath()
+ if _, ok := needed[path]; ok {
+ delete(needed, path)
+ if len(needed) == 0 {
+ return nil
Review Comment:
This early return abandons the manifest iteration, so `mf.Entries` closes
with `complete=false` and the manifest never lands in the cache.
`validateDataFilesExist` usually runs first in a RowDelta and hits this exit in
the first manifest with a small `needed` set, then
`validateAddedDataFilesMatchingFilter` re-opens the same manifest from the
backing store.
So the "each manifest opened once" benefit only holds on full-read paths,
not this common one. I'd either read the rest of the current manifest to EOF
before returning, or note the limitation in the comment. wdyt?
##########
table/conflict_validation.go:
##########
@@ -181,6 +184,184 @@ type conflictContext struct {
// Resolved once; subsequent validator calls reuse the walk.
concurrent []Snapshot
+
+ // Lazily caches raw manifest bytes for this validation attempt. The
+ // cache is intentionally scoped to the context because all validators
+ // inspect the same immutable metadata snapshot, while
ManifestFile.Entries
+ // still performs descriptor-specific inheritance for each logical read.
+ manifestIO *conflictManifestIO
+
+ // Parsed manifest-list descriptors are also shared across validators.
+ manifestLists map[int64]conflictManifestList
+}
+
+type conflictManifestList struct {
+ manifests []iceberg.ManifestFile
+ err error
+}
+
+// conflictManifestIO is a read-through cache used only while conflict
+// validators inspect one conflictContext. Manifest files are immutable after
+// they are committed, so the same path can be safely served from the cached
+// bytes to later validators without reopening the backing object store.
+//
+// The cache stores raw bytes instead of decoded entries. The first read still
+// streams from the backing IO and is recorded only when it reaches EOF, so an
+// early validator exit does not force a full manifest read. Later reads stream
+// decoded entries from the completed bytes and let each ManifestFile
+// descriptor apply its own inheritance metadata.
+type conflictManifestIO struct {
+ base iceio.IO
+ files map[string][]byte
+}
+
+func newConflictManifestIO(base iceio.IO) *conflictManifestIO {
+ return &conflictManifestIO{
+ base: base,
+ files: make(map[string][]byte),
+ }
+}
+
+func (c *conflictManifestIO) Open(name string) (iceio.File, error) {
+ if data, ok := c.files[name]; ok {
+ return newConflictManifestFile(name, data), nil
+ }
+
+ f, err := c.base.Open(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return &conflictManifestRecordingFile{
+ base: f,
+ cache: c,
+ name: name,
+ cacheable: true,
+ }, nil
+}
+
+func (c *conflictManifestIO) Remove(name string) error {
+ delete(c.files, name)
+
+ return c.base.Remove(name)
+}
+
+func (c *conflictManifestIO) Stat(name string) (fs.FileInfo, error) {
+ if data, ok := c.files[name]; ok {
+ return conflictManifestFileInfo{name: name, size:
int64(len(data))}, nil
+ }
+
+ if statIO, ok := c.base.(iceio.StatIO); ok {
+ return statIO.Stat(name)
+ }
+
+ f, err := c.Open(name)
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+
+ return f.Stat()
+}
+
+type conflictManifestFile struct {
+ *bytes.Reader
+ name string
+ size int64
+}
+
+func newConflictManifestFile(name string, data []byte) *conflictManifestFile {
+ return &conflictManifestFile{
+ Reader: bytes.NewReader(data),
+ name: name,
+ size: int64(len(data)),
+ }
+}
+
+func (f *conflictManifestFile) Close() error { return nil }
+
+func (f *conflictManifestFile) Stat() (fs.FileInfo, error) {
+ return conflictManifestFileInfo{name: f.name, size: f.size}, nil
+}
+
+type conflictManifestRecordingFile struct {
+ base iceio.File
+ cache *conflictManifestIO
+ name string
+ data []byte
+ cacheable bool
+ complete bool
+}
+
+func (f *conflictManifestRecordingFile) Read(p []byte) (int, error) {
+ n, err := f.base.Read(p)
+ if f.cacheable && n > 0 {
+ f.data = append(f.data, p[:n]...)
+ }
+ if f.cacheable && errors.Is(err, io.EOF) {
+ f.complete = true
+ }
+
+ return n, err
+}
+
+func (f *conflictManifestRecordingFile) ReadAt(p []byte, offset int64) (int,
error) {
+ f.cacheable = false
+
+ return f.base.ReadAt(p, offset)
+}
+
+func (f *conflictManifestRecordingFile) Seek(offset int64, whence int) (int64,
error) {
+ f.cacheable = false
+
+ return f.base.Seek(offset, whence)
+}
+
+func (f *conflictManifestRecordingFile) Stat() (fs.FileInfo, error) {
+ return f.base.Stat()
+}
+
+func (f *conflictManifestRecordingFile) Close() error {
+ err := f.base.Close()
+ if err == nil && f.cacheable && f.complete {
+ f.cache.files[f.name] = f.data
+ }
+
+ return err
+}
+
+type conflictManifestFileInfo struct {
+ name string
+ size int64
+}
+
+func (f conflictManifestFileInfo) Name() string { return f.name }
+func (f conflictManifestFileInfo) Size() int64 { return f.size }
+func (f conflictManifestFileInfo) Mode() fs.FileMode { return 0 }
+func (f conflictManifestFileInfo) ModTime() time.Time { return time.Time{} }
+func (f conflictManifestFileInfo) IsDir() bool { return false }
+func (f conflictManifestFileInfo) Sys() any { return nil }
+
+func (c *conflictContext) manifestReadIO() *conflictManifestIO {
+ if c.manifestIO == nil {
+ c.manifestIO = newConflictManifestIO(c.fs)
+ }
+
+ return c.manifestIO
+}
+
+func (c *conflictContext) manifestsFor(snap Snapshot) ([]iceberg.ManifestFile,
error) {
+ if c.manifestLists == nil {
+ c.manifestLists = make(map[int64]conflictManifestList)
+ }
+ if cached, ok := c.manifestLists[snap.SnapshotID]; ok {
+ return cached.manifests, cached.err
+ }
+
+ manifests, err := snap.Manifests(c.manifestReadIO())
Review Comment:
The manifest-list bytes recorded here are never reused. `manifestsFor`
caches the parsed `[]iceberg.ManifestFile` and short-circuits on it, so every
later call returns from the parse cache without re-reading the path, so the raw
list bytes just sit in `files` for the life of the context.
I'd pass `c.fs` here instead of `c.manifestReadIO()`. Only the data/delete
manifest Avro files get read by multiple validators and actually benefit from
the byte cache; the manifest list is read exactly once per snapshot.
##########
table/conflict_validation_bench_test.go:
##########
@@ -0,0 +1,150 @@
+// 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"
+ "fmt"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+)
+
+func BenchmarkConflictValidationSharedManifestReads(b *testing.B) {
+ for _, entryCount := range []int{1_000, 10_000} {
+ for _, validatorCount := range []int{2, 8} {
+ b.Run(fmt.Sprintf("entries=%d/validators=%d",
entryCount, validatorCount), func(b *testing.B) {
+ baseContext :=
newConflictValidationBenchmarkContext(b, entryCount)
+ fs :=
baseContext.fs.(*conflictValidationBenchmarkIO)
+ visit := func(Snapshot, iceberg.ManifestEntry)
error { return nil }
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ // Each iteration models one commit
attempt. Validators
+ // within that attempt share the
context and its cache.
+ ctx := &conflictContext{fs:
baseContext.fs, concurrent: baseContext.concurrent}
+ for range validatorCount {
+ if err :=
ctx.forEachAddedEntry(iceberg.ManifestContentData, visit); err != nil {
Review Comment:
The benchmark loops the same validator, so it only measures the full-read
path where the cache pays off cleanly; that's where the 16->2 opens number
comes from. The realistic RowDelta shape (`validateDataFilesExist` with an
early exit, then `validateAddedDataFilesMatchingFilter`) is exactly the case
where a manifest gets re-opened, and it's neither benchmarked nor tested.
A sub-benchmark for that composition, plus a test that runs
`validateDataFilesExist` then `validateAddedDataFilesMatchingFilter` on one
context and asserts data manifests open at most once, would keep the
opened-once claim honest. wdyt?
--
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]
