laskoviymishka commented on code in PR #1970:
URL: https://github.com/apache/iceberg-go/pull/1970#discussion_r3903424820
##########
table/transaction.go:
##########
@@ -3210,6 +3210,7 @@ func (t *Transaction) Scan(opts ...ScanOption) (*Scan,
error) {
metadata: updatedMeta,
metadataLocation: t.tbl.metadataLocation,
ioF: t.tbl.fsF,
+ manifestCache: t.tbl.manifestCache,
Review Comment:
Transaction scans plan over `updatedMeta` (which includes the staged
snapshot) but share the committed table's cache, so a staged snapshot's
manifest set lands in the committed cache under its staged ID. The bounding
commit takes most of the sting out of this: those orphaned entries now age out
of the LRU instead of lingering until `Refresh`, so it's smaller than it was
last round. It's still a little surprising that a committed table's cache picks
up snapshots that may never commit; a fresh `newSnapshotManifestCache()` for
transaction scans would keep the two views cleanly separate. Not blocking.
Thoughts?
##########
table/snapshot_manifest_cache.go:
##########
@@ -0,0 +1,182 @@
+// 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"
+ "slices"
+ "strings"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ lru "github.com/hashicorp/golang-lru/v2"
+)
+
+// snapshotManifestCacheSize bounds the number of decoded manifest lists held
+// by a table. A table normally reuses its current snapshot, while a bounded
+// history is enough to retain useful locality for repeated historical scans.
+const snapshotManifestCacheSize = 64
+
+// snapshotManifestSet keeps the complete manifest list and its content
+// partitions together. Manifest descriptors are immutable for an Iceberg
+// snapshot, so a table can safely share this decoded result across scans.
+type snapshotManifestSet struct {
+ all []iceberg.ManifestFile
+ data []iceberg.ManifestFile
+ deletes []iceberg.ManifestFile
+}
+
+func newSnapshotManifestSet(manifests []iceberg.ManifestFile)
snapshotManifestSet {
+ set := snapshotManifestSet{all: slices.Clone(manifests)}
+ if len(manifests) == 0 {
+ return set
+ }
+
+ set.data = make([]iceberg.ManifestFile, 0, len(manifests))
+ set.deletes = make([]iceberg.ManifestFile, 0, len(manifests))
+ for _, manifest := range set.all {
Review Comment:
This partitions negatively, so anything that isn't `ManifestContentDeletes`
falls into `data`, where the old append-scan loop was a positive `==
ManifestContentData` check. Identical with today's two content values, but the
forward-compat posture flips: if a future spec rev adds a third content type,
the old code kept it out of the append scan and this quietly treats it as data.
Java's `ManifestContent.fromId` hard-errors on anything but 0/1. Not blocking,
but a positive check for the data partition would preserve the old semantics.
wdyt?
##########
table/snapshot_manifest_cache_internal_test.go:
##########
@@ -0,0 +1,286 @@
+// 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"
+ "fmt"
+ "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 TestTableScansReuseSnapshotManifestList(t *testing.T) {
+ fs := newTrackingCallsIO()
+ meta, err := NewMetadata(simpleSchema(), iceberg.UnpartitionedSpec,
UnsortedSortOrder,
+ "mem://snapshot-manifest-cache",
iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+ builder, err := MetadataBuilderFromBase(meta, "")
+ require.NoError(t, err)
+
+ const snapshotID = int64(1)
+ manifestPath := "mem://snapshot-manifest-cache/metadata/manifest.avro"
+ manifest := writeManifest(t, fs.trackingIO, snapshotID, 1, manifestPath,
+ "mem://snapshot-manifest-cache/data/file.parquet")
+ manifestListPath := "mem://snapshot-manifest-cache/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", "snapshot-manifest-cache"}, built,
"metadata.json", testFSF(fs), nil)
+ for range 2 {
+ tasks, err :=
tbl.Scan(WithMaxConcurrency(1)).PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.Len(t, tasks, 1)
+ }
+
+ assert.Equal(t, 1, fs.openCount[manifestListPath], "manifest list
should be decoded once across scans")
+ assert.Equal(t, 2, fs.openCount[manifestPath], "manifest entries still
need to be read for each plan")
+}
+
+func TestSnapshotManifestCacheSeparatesContentAndProtectsSlices(t *testing.T) {
+ data := iceberg.NewManifestFile(2, "data.avro", 10, 0, 1).Build()
+ deleteManifest := iceberg.NewManifestFile(2, "delete.avro", 20, 0, 1).
+ Content(iceberg.ManifestContentDeletes).
+ Build()
+ cache := newSnapshotManifestCache()
+
+ fs := iceio.NewMemFS()
+ var list bytes.Buffer
+ sequenceNumber := int64(1)
+ require.NoError(t, iceberg.WriteManifestList(2, &list, 1, nil,
&sequenceNumber, 0,
+ []iceberg.ManifestFile{data, deleteManifest}))
+ const listPath = "mem://snapshot-manifest-cache/separate.avro"
+ require.NoError(t, fs.WriteFile(listPath, list.Bytes()))
+
+ set, err := cache.get(context.Background(), Snapshot{SnapshotID: 1,
ManifestList: listPath}, fs)
+ require.NoError(t, err)
+ require.Len(t, set.allManifests(), 2)
+ require.Len(t, set.dataManifests(), 1)
+ require.Len(t, set.deleteManifests(), 1)
+
+ all := set.allManifests()
+ all[0] = nil
+ assert.Equal(t, "data.avro", set.allManifests()[0].FilePath())
+ dataManifests := set.dataManifests()
+ deleteManifests := set.deleteManifests()
+ assert.Equal(t, "data.avro", dataManifests[0].FilePath())
+ assert.Equal(t, "delete.avro", deleteManifests[0].FilePath())
+}
+
+func TestSnapshotManifestCacheSharesInFlightRead(t *testing.T) {
+ const listPath = "mem://snapshot-manifest-cache/concurrent.avro"
+ base := iceio.NewMemFS()
+ var list bytes.Buffer
+ sequenceNumber := int64(1)
+ require.NoError(t, iceberg.WriteManifestList(2, &list, 1, nil,
&sequenceNumber, 0,
+ []iceberg.ManifestFile{}))
+ require.NoError(t, base.WriteFile(listPath, list.Bytes()))
+
+ fs := &blockingSnapshotManifestIO{
+ IO: base,
+ blockedPath: listPath,
+ started: make(chan struct{}),
+ release: make(chan struct{}),
+ opens: make(map[string]int),
+ }
+ cache := newSnapshotManifestCache()
+ snapshot := Snapshot{SnapshotID: 1, ManifestList: listPath}
+ const callers = 16
+ errs := make(chan error, callers)
+ for range callers {
+ go func() {
+ _, err := cache.get(context.Background(), snapshot, fs)
+ errs <- err
+ }()
+ }
+
+ select {
+ case <-fs.started:
+ case <-time.After(time.Second):
+ t.Fatal("manifest-list read did not start")
+ }
+ close(fs.release)
Review Comment:
If this `t.Fatal` fires, the `close(fs.release)` never runs and all 16
goroutines are stuck: the owner inside `Open`, the 15 waiters on `entry.ready`.
Under `-race` a leaked goroutine tends to surface as a false-positive race in
whatever test runs next. The canceled-waiter test you added this round already
has the fix, a `sync.Once` plus `t.Cleanup` that closes `release`
unconditionally. Worth mirroring here, and using `t.Context()` in the
goroutines while you're at it.
##########
table/snapshot_manifest_cache.go:
##########
@@ -0,0 +1,182 @@
+// 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"
+ "slices"
+ "strings"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ lru "github.com/hashicorp/golang-lru/v2"
+)
+
+// snapshotManifestCacheSize bounds the number of decoded manifest lists held
+// by a table. A table normally reuses its current snapshot, while a bounded
+// history is enough to retain useful locality for repeated historical scans.
+const snapshotManifestCacheSize = 64
+
+// snapshotManifestSet keeps the complete manifest list and its content
+// partitions together. Manifest descriptors are immutable for an Iceberg
+// snapshot, so a table can safely share this decoded result across scans.
+type snapshotManifestSet struct {
+ all []iceberg.ManifestFile
+ data []iceberg.ManifestFile
+ deletes []iceberg.ManifestFile
+}
+
+func newSnapshotManifestSet(manifests []iceberg.ManifestFile)
snapshotManifestSet {
+ set := snapshotManifestSet{all: slices.Clone(manifests)}
+ if len(manifests) == 0 {
+ return set
+ }
+
+ set.data = make([]iceberg.ManifestFile, 0, len(manifests))
+ set.deletes = make([]iceberg.ManifestFile, 0, len(manifests))
+ for _, manifest := range set.all {
+ if manifest.ManifestContent() == iceberg.ManifestContentDeletes
{
+ set.deletes = append(set.deletes, manifest)
+ } else {
+ set.data = append(set.data, manifest)
+ }
+ }
+
+ return set
+}
+
+func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.all)
+}
+
+func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.data)
+}
+
+func (s snapshotManifestSet) deleteManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.deletes)
+}
+
+type snapshotManifestCacheKey struct {
+ snapshotID int64
+ manifestList string
+ hasEmbeddedSources bool
+ embeddedManifestSources string
+}
+
+func snapshotManifestCacheKeyFor(snapshot Snapshot) snapshotManifestCacheKey {
+ var (
+ hasEmbeddedSources bool
+ embeddedManifestSources string
+ )
+ if snapshot.ManifestList == "" {
+ hasEmbeddedSources = snapshot.ManifestLocations != nil
+ embeddedManifestSources =
strings.Join(snapshot.ManifestLocations, "\x00")
Review Comment:
A one-liner noting that `\x00` can't appear in a valid file path or URI (so
the join is collision-free) would make this obviously safe rather than
something the next reader has to reason through.
##########
table/snapshot_manifest_cache.go:
##########
@@ -0,0 +1,182 @@
+// 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"
+ "slices"
+ "strings"
+ "sync"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ lru "github.com/hashicorp/golang-lru/v2"
+)
+
+// snapshotManifestCacheSize bounds the number of decoded manifest lists held
+// by a table. A table normally reuses its current snapshot, while a bounded
+// history is enough to retain useful locality for repeated historical scans.
+const snapshotManifestCacheSize = 64
+
+// snapshotManifestSet keeps the complete manifest list and its content
+// partitions together. Manifest descriptors are immutable for an Iceberg
+// snapshot, so a table can safely share this decoded result across scans.
+type snapshotManifestSet struct {
+ all []iceberg.ManifestFile
+ data []iceberg.ManifestFile
+ deletes []iceberg.ManifestFile
+}
+
+func newSnapshotManifestSet(manifests []iceberg.ManifestFile)
snapshotManifestSet {
+ set := snapshotManifestSet{all: slices.Clone(manifests)}
+ if len(manifests) == 0 {
+ return set
+ }
+
+ set.data = make([]iceberg.ManifestFile, 0, len(manifests))
+ set.deletes = make([]iceberg.ManifestFile, 0, len(manifests))
+ for _, manifest := range set.all {
+ if manifest.ManifestContent() == iceberg.ManifestContentDeletes
{
+ set.deletes = append(set.deletes, manifest)
+ } else {
+ set.data = append(set.data, manifest)
+ }
+ }
+
+ return set
+}
+
+func (s snapshotManifestSet) allManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.all)
+}
+
+func (s snapshotManifestSet) dataManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.data)
+}
+
+func (s snapshotManifestSet) deleteManifests() []iceberg.ManifestFile {
+ return slices.Clone(s.deletes)
+}
+
+type snapshotManifestCacheKey struct {
+ snapshotID int64
+ manifestList string
+ hasEmbeddedSources bool
+ embeddedManifestSources string
+}
+
+func snapshotManifestCacheKeyFor(snapshot Snapshot) snapshotManifestCacheKey {
+ var (
+ hasEmbeddedSources bool
+ embeddedManifestSources string
+ )
+ if snapshot.ManifestList == "" {
+ hasEmbeddedSources = snapshot.ManifestLocations != nil
+ embeddedManifestSources =
strings.Join(snapshot.ManifestLocations, "\x00")
+ }
+
+ return snapshotManifestCacheKey{
+ snapshotID: snapshot.SnapshotID,
+ manifestList: snapshot.ManifestList,
+ hasEmbeddedSources: hasEmbeddedSources,
+ embeddedManifestSources: embeddedManifestSources,
+ }
+}
+
+type snapshotManifestCacheEntry struct {
+ ready chan struct{}
+ manifests snapshotManifestSet
+ err error
+}
+
+// snapshotManifestCache memoizes successful manifest-list reads and shares an
+// in-flight read with concurrent scans. Completed reads use a bounded LRU so a
+// historical scan cannot retain every snapshot for the lifetime of a table.
+// Failed reads are removed so a transient object-store error does not poison
+// the cache.
+type snapshotManifestCache struct {
+ mu sync.Mutex
+ entries map[snapshotManifestCacheKey]*snapshotManifestCacheEntry
+ complete *lru.Cache[snapshotManifestCacheKey, snapshotManifestSet]
+}
+
+func newSnapshotManifestCache() *snapshotManifestCache {
+ complete, err := lru.New[snapshotManifestCacheKey,
snapshotManifestSet](snapshotManifestCacheSize)
+ if err != nil {
+ panic(err)
+ }
+
+ return &snapshotManifestCache{
+ entries:
make(map[snapshotManifestCacheKey]*snapshotManifestCacheEntry),
+ complete: complete,
+ }
+}
+
+func (c *snapshotManifestCache) get(
+ ctx context.Context,
+ snapshot Snapshot,
+ fio iceio.IO,
+) (snapshotManifestSet, error) {
+ if c == nil {
+ manifests, err := snapshot.Manifests(fio)
+
+ return newSnapshotManifestSet(manifests), err
+ }
+
+ key := snapshotManifestCacheKeyFor(snapshot)
+ c.mu.Lock()
+ if manifests, ok := c.complete.Get(key); ok {
+ c.mu.Unlock()
+
+ return manifests, nil
+ }
+ if entry, ok := c.entries[key]; ok {
+ c.mu.Unlock()
+
+ select {
+ case <-entry.ready:
+ return entry.manifests, entry.err
+ default:
+ select {
+ case <-entry.ready:
+ return entry.manifests, entry.err
+ case <-ctx.Done():
+ return snapshotManifestSet{}, ctx.Err()
+ }
+ }
+ }
+
+ entry := &snapshotManifestCacheEntry{ready: make(chan struct{})}
+ c.entries[key] = entry
+ c.mu.Unlock()
+
+ manifests, err := snapshot.Manifests(fio)
Review Comment:
The bounding commit plumbed `ctx` into `get`, which lets a waiter bail out
of its wait, but the producer still reads through a context-less
`snapshot.Manifests(fio)` and nothing cleans up on an early exit. So both
failure shapes are still here: if that read stalls, the in-flight
`entries[key]` sits there until it returns, and every retrying waiter with a
fresh ctx re-blocks against it; if it panics, the `delete` and `close` at the
bottom never run and the entry is a zombie with `ready` never closed. Before
this PR each caller read independently and timed out on its own; now one stuck
read serializes everyone behind it until `Refresh`.
A deferred guard that deletes the entry and closes `ready` on any early
return covers the panic and error cases; the stall itself needs ctx threaded
into the read. I'd still want a test for the retry-after-stuck-producer path,
since the canceled-waiter test covers the cancel but not the case where a
waiter escapes, retries, and resolves only once the producer finishes. 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]
