zeroshade commented on code in PR #1668:
URL: https://github.com/apache/iceberg-go/pull/1668#discussion_r3732158762
##########
table/table.go:
##########
@@ -327,24 +327,51 @@ func (t Table) AllManifests(ctx context.Context)
iter.Seq2[iceberg.ManifestFile,
}
type list = tblutils.Enumerated[[]iceberg.ManifestFile]
- g := errgroup.Group{}
-
- n := len(t.metadata.Snapshots())
- ch := make(chan list, n)
-
- for i, sn := range t.metadata.Snapshots() {
+ snapshots := t.metadata.Snapshots()
+ n := len(snapshots)
+ workCtx, cancel := context.WithCancel(ctx)
+ jobs := make(chan int)
+ ch := make(chan list, max(1, min(n, 16)))
Review Comment:
Non-blocking: because `out` inside `MakeSequencedChan` is allocated with
capacity `n`, this `cap 16` provides essentially no backpressure — peak memory
is still all `n` manifest lists held at once. Worth a short comment so a future
reader doesn't mistake this expression for a memory bound.
##########
table/all_manifests_internal_test.go:
##########
@@ -64,10 +66,85 @@ func TestAllManifestsCompletesAfterErrorChannelCloses(t
*testing.T) {
}
}
+func TestAllManifestsLimitsConcurrentReads(t *testing.T) {
Review Comment:
Suggestions for hardening, none blocking — the bound itself is well covered
by this test. Cases that would round it out: a worker error cancels the
remaining workers and the *first* error is what surfaces; an early `break` out
of the range leaks no goroutines (goleak would catch this cheaply); context
cancellation propagates to in-flight workers; and the `n == 0` / `n == 1` edges
where `max(1, min(n, ...))` collapses.
##########
table/table.go:
##########
@@ -327,24 +327,51 @@ func (t Table) AllManifests(ctx context.Context)
iter.Seq2[iceberg.ManifestFile,
}
type list = tblutils.Enumerated[[]iceberg.ManifestFile]
- g := errgroup.Group{}
-
- n := len(t.metadata.Snapshots())
- ch := make(chan list, n)
-
- for i, sn := range t.metadata.Snapshots() {
+ snapshots := t.metadata.Snapshots()
+ n := len(snapshots)
+ workCtx, cancel := context.WithCancel(ctx)
+ jobs := make(chan int)
+ ch := make(chan list, max(1, min(n, 16)))
+ workers := max(1, min(n, min(runtime.GOMAXPROCS(0), 16)))
Review Comment:
Two small things here, both optional.
1. `16` appears on this line and the one above as an unnamed literal. A
named constant would make the intent legible and keep the two in sync.
2. `GOMAXPROCS(0)` derives a CPU bound for what is purely remote I/O. On a
two-core CI runner, a 1000-snapshot history now fetches two-wide, which is
likely slower than intended. Consider a bound independent of CPU count — and
possibly an option, since `Delete`/`Overwrite` already expose a concurrency
knob (`WithOverwriteConcurrency`), which leaves `AllManifests` as the odd one
out.
##########
table/table.go:
##########
@@ -327,24 +327,51 @@ func (t Table) AllManifests(ctx context.Context)
iter.Seq2[iceberg.ManifestFile,
}
type list = tblutils.Enumerated[[]iceberg.ManifestFile]
- g := errgroup.Group{}
-
- n := len(t.metadata.Snapshots())
- ch := make(chan list, n)
-
- for i, sn := range t.metadata.Snapshots() {
+ snapshots := t.metadata.Snapshots()
+ n := len(snapshots)
+ workCtx, cancel := context.WithCancel(ctx)
+ jobs := make(chan int)
+ ch := make(chan list, max(1, min(n, 16)))
+ workers := max(1, min(n, min(runtime.GOMAXPROCS(0), 16)))
+ g, groupCtx := errgroup.WithContext(workCtx)
+
+ for range workers {
Review Comment:
Worth a follow-up: `errgroup.Group.SetLimit(workers)` plus a single
scheduling goroutine yields the same bound in roughly five lines and removes
the hand-rolled `jobs` channel entirely. The current implementation is correct
— this is purely a simplification, so it's fine to defer or skip.
##########
table/all_manifests_internal_test.go:
##########
@@ -64,10 +66,85 @@ func TestAllManifestsCompletesAfterErrorChannelCloses(t
*testing.T) {
}
}
+func TestAllManifestsLimitsConcurrentReads(t *testing.T) {
+ const snapshotCount = 64
+ const maxWorkers = 2
+
+ previousMaxProcs := runtime.GOMAXPROCS(maxWorkers)
+ defer runtime.GOMAXPROCS(previousMaxProcs)
+
+ var trackingFS *manifestTrackingIO
+ tbl, _ := tableWithManifestListsUsingIO(t, snapshotCount, func(memFS
*iceio.MemFS) iceio.IO {
+ trackingFS = &manifestTrackingIO{IO: memFS, delay: 5 *
time.Millisecond}
+
+ return trackingFS
+ })
+
+ for mf, err := range tbl.AllManifests(context.Background()) {
+ require.NoError(t, err)
+ require.NotNil(t, mf)
+ }
+
+ trackingFS.mu.Lock()
+ maxOpen := trackingFS.maxOpen
+ trackingFS.mu.Unlock()
+ require.Greater(t, maxOpen, 1)
Review Comment:
Non-blocking, but this is the one assertion I'd soften. `maxOpen > 1`
requires two goroutines to overlap within the 5 ms sleep, so it can flake on a
loaded or effectively single-CPU runner. The upper-bound assertion on the next
line is the property this PR actually adds; consider keeping only that, or
having the tracking IO rendezvous through a semaphore or `WaitGroup` so the
overlap is deterministic rather than timing-derived. Mutating global
`GOMAXPROCS` is also worth a second look if tests ever run in parallel here.
--
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]
