perf(rest): fetch remote scan plan tasks concurrently

[fallintoplace]

Summary

• Concurrency: Fetch each remote plan-task frontier concurrently. Honor table.WithMaxConcurrency(n) through ScanPlanningRequest.MaxConcurrency. Default to runtime.GOMAXPROCS(0) and cap workers at the frontier size.
• Ordering: Preserve breadth-first response order, once-only handle expansion, and cycle protection. Each frontier finishes before the next one starts.
• Errors: Return the first failure in handle order with its handle name. Cancel unfinished later requests while allowing earlier handles to finish.
• Validation: Reject malformed task payloads before returning a partial scan plan.
• Schema: Guard the keys and values of exported JSON fields. Document why marshaling must avoid copying the lazy atomic caches.

Benchmark

go test ./catalog/rest -run '^$' -bench '^BenchmarkCollectScanTasks64Handles$' -benchtime=1x -count=5 -benchmem

Apple M1 Pro, darwin/arm64, Go 1.26.3. Each run fetches 64 handles with 10 ms of simulated server latency per request. The table shows medians from five runs. Each row uses the explicit concurrency limit named in the benchmark subtest.

Explicit fetch limit	Time/op	Bytes/op	Allocs/op
1	1056.3 ms	745 KiB	8,821
2	492.2 ms	752 KiB	8,820
4	232.1 ms	796 KiB	9,010
8	128.4 ms	975 KiB	9,441
16	74.0 ms	1025 KiB	9,753
32	39.2 ms	1414 KiB	11,337
64	22.5 ms	1705 KiB	12,520

These numbers measure the simulated fetch workload. Higher limits reduce elapsed time and use more memory.

Tests

• GOFLAGS=-p=1 make test
• go test -race -p=1 . ./catalog/rest -count=1
• GOMAXPROCS=1 go test ./catalog/rest -run '^TestCollectScanTasks' -count=1
• golangci-lint run --timeout=10m --allow-serial-runners
• Confirmed the schema regression test fails if the ID copy is removed, or a new JSON field is added with or without omitempty.

[zeroshade]

The concurrency structure here is well built — bounded, cancellable, order-preserving — but the error path loses determinism, and that's worth fixing before it lands.

Major — concurrent failures produce a timing-dependent error (catalog/rest/scan_planning.go:362-363)

The helper returns group.Wait() directly. errgroup retains only the error from the first worker to fail, so with more than one failure in flight the surfaced error depends on goroutine scheduling.

Concretely: a table dropped mid-plan plus an expired task handle can surface catalog.ErrNoSuchTable on one run and ErrNoSuchPlanTask on the next, from identical server state. Before this change fetches happened in handle order, so the first failure in that order won deterministically. Any caller branching on those sentinels with errors.Is now behaves differently run to run, and the remaining errors are discarded with no stated policy.

Suggested fix: collect per-index errors, then either return the handle-order-first error to preserve the previous contract, or errors.Join them in stable index order. Exclude sibling context.Canceled errors caused by the group's own cancellation, otherwise the joined error gets noisy. A test with two concurrent failures asserting a deterministic sentinel would lock it down.

Minor — fetch concurrency isn't tunable

remoteScanTaskFetchConcurrency = 8 (catalog/rest/scan_planning.go:54-58) is an unexported constant with no corresponding Option, and options.go exposes nothing for it. 8 is a sensible default and there's no correctness issue — but callers planning against a rate-limited or unusually beefy catalog can't adjust it. Worth exposing if that's easy; fine to leave if you'd rather not grow the option surface yet.

What's right

Credit where it's due, since this is the part that's easy to get wrong:

• Fan-out is bounded via errgroup.SetLimit(8) (:347) rather than one goroutine per task — the failure mode I was most concerned about, and it's handled.
• errgroup.WithContext(ctx) (:346) with groupCtx passed to every FetchScanTasks call (:351), so cancellation and timeouts propagate and the first error cancels siblings promptly.
• Result order is preserved: responses land in indexed slots, so concurrent completion still reassembles deterministically and BFS order is maintained.
• Writes go to disjoint slots, so no shared-accumulator race.
• Existing tests cover ordering, the concurrency bound, cancellation, and dedup.

CI green (15/15); BLOCKED is just the approval gate.

---

This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Go maintainer. The findings cite the project's review criteria; if you think one of them is mis-applied, please reply on the PR and a maintainer will weigh in.

More on how to contribute to Apache Iceberg Go: CONTRIBUTING.md

[laskoviymishka]

This is in good shape and I'd approve once a few small things are settled, none of them blocking.

The concurrent frontier fetch is a clean design: indexing responses by handle position keeps the result order deterministic no matter how completion timing shakes out, the errgroup bounding is right, and I especially like that the error path now forces the first handle-order error to win rather than whichever request loses the race, with a test that pins exactly that. Each of the new tests pins a distinct invariant.

The schema.go change is a good catch (copying the whole Schema by value was dragging its atomic.Pointer cache fields along, a real data race under concurrent marshals) and it's already isolated in its own fix(schema): commit, so I'm not fussed about it riding along here.

The remaining items are all take-or-leave and I left them inline: a short comment on the Alias{...} literal so a future field doesn't get silently dropped, wrapping the frontier fetch error with the handle for diagnosability, and a one-line note that the new schema race test only bites under -race. The fixed concurrency of 8 and the tier-barrier-versus-work-stealing gap against Java are worth a thought for later but aren't blockers.

Nice work overall.

[laskoviymishka]

not a blocker, but I wonder if this wants to be tunable rather than a fixed 8. Java derives its fetch parallelism from max(2, availableProcessors()) via the worker pool size, so it scales with the box and can be tuned for server rate limits. A catalog option or table property (something like rest.fetch-scan-tasks.concurrency) would match that. Fine as a follow-up, just flagging.

[laskoviymishka]

one perf thought, definitely not blocking: because we fetch a whole frontier and only advance once fetchScanTaskFrontier returns, each BFS level waits on its slowest handle before the next level's handles start. Java's ScanTaskIterable pushes child handles onto a shared queue as each response lands, so idle workers pick up the next level without waiting for their peers. For a deep, uneven fanout (1 to 8 to 64) that tier barrier can leave workers idle. The current shape is simpler and keeps ordering clean, so fine to keep, but maybe worth a follow-up if fanout depth ever gets expensive.

[laskoviymishka]

tiny thing: this is make([]string, 0) while the handles allocation a few lines up uses a capacity hint. var nextFrontier []string reads a touch cleaner. Purely cosmetic, I checked and staticcheck doesn't flag either form so it won't affect CI.

[laskoviymishka]

nice fix on the ordering here, forcing the first handle-order error to win instead of whichever request loses the race is exactly right, and the new test pins it well.

Small follow-on while you're in here: the error still goes back bare, so a failure inside an 8-way frontier doesn't say which handle expired, and the rest of the file wraps (WaitForPlan, remoteScanTasks). Recording it wrapped, e.g. errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, err), keeps the errors.Is chain (and the handle-order selection below) intact.

[laskoviymishka]

agreed with this fix, and I like that it landed as its own fix(schema): commit so it's cherry-pickable.

One small thing on the literal itself: the old *(*Alias)(s) picked up every exported field automatically, and this now hard-codes the two we have. Correct today, but a new json-tagged field on Schema would be silently dropped here with no compile error and no test failure (the JSONEq golden only catches unexpected additions, not omissions). A short comment above the literal saying it must list every marshalled field, and why it can't just copy the struct, would keep the next person from missing it. Non-blocking, wdyt?

[laskoviymishka]

this is a nice race reproducer, but it only proves anything under go test -race. There's no assertion that fails if the race comes back, so without the detector it passes against the old code too. A one-line comment noting it's meant to run under -race would stop the next person assuming it guards the fix on its own.

[zeroshade]

The concurrency core is genuinely correct — I verified bounding, ordering, cancellation and shared-state safety hard, and they all hold. But the error-determinism fix that was the sole blocking item last round doesn't work: errgroup's cancel-cause propagation defeats the context.Canceled filter, so the sentinel still flips with timing, and the new wrapper now attaches the wrong handle name.

Blocking — catalog/rest/scan_planning.go:365-375: the handle-order error selection is inert

errgroup.WithContext builds its context with context.WithCancelCause and cancels with the winning error as the cause (x/sync/errgroup/errgroup.go:49,96,126). net/http reports context.Cause(ctx) — not context.Canceled — when a request's context is cancelled (net/http/transport.go:677,1573,1584,2462,2958). So a sibling aborted by the group's own cancellation comes back carrying the other handle's REST error, and !errors.Is(err, context.Canceled) at :370 never fires for the case it was written for.

Probe with identical server state and handle order ["table","plan-task"], varying only which one 404s first:

which 404s first	pre-change serial	head
index 0 (table)	catalog.ErrNoSuchTable	catalog.ErrNoSuchTable ✓
index 1 (plan-task)	catalog.ErrNoSuchTable	ErrNoSuchPlanTask ✗

In the second row errors.Is(err, context.Canceled) is false, and the message is literally:

fetching scan tasks for handle "table": Post "http://…/tasks": NoSuchPlanTaskException

So the comment at :366-368 ("Preserve the serial fetch contract: … the first error in handle order wins") is false, the pre-change contract was ErrNoSuchTable in both timings, and the wrapper added by 1af5d3f5 glues the wrong handle name onto another handle's error — making diagnostics worse than the bare group.Wait() it replaced.

catalog/rest/scan_planning_test.go:1325-1352,1354-1368 cannot detect this. orderedScanTaskErrorTransport.RoundTrip never reads req.Context(); it blocks on a channel then unconditionally synthesizes a 404. Handle "table" therefore always produces its own error, so assert.NotErrorIs(t, err, ErrNoSuchPlanTask) is a tautology. With any context-honouring transport — including the real one — that assertion fails.

Suggested fix: stop depending on cause-shaped cancellation detection. Use a plain errgroup.Group with a context you cancel yourself, so the cause stays context.Canceled and the filter means what it says:

fetchCtx, cancel := context.WithCancel(ctx)
defer cancel()

var group errgroup.Group
group.SetLimit(remoteScanTaskFetchConcurrency)
for i, handle := range handles {
    group.Go(func() error {
        response, err := r.FetchScanTasks(fetchCtx, ident, FetchScanTasksRequest{PlanTask: handle})
        if err != nil {
            errs[i] = fmt.Errorf("fetching scan tasks for handle %q: %w", handle, err)
            cancel() // plain cancel -> siblings see context.Canceled
            return err
        }
        responses[i] = response
        return nil
    })
}

Alternatively drop the ambition: errors.Join the non-cancel errors in index order, or state honestly that the first error to occur wins and remove both the comment and the handle wrapper. Either way the test needs a transport that honours req.Context(), plus the reversed-latency case above.

Major — scan_planning.go:55-58: the concurrency constant ignores an existing public knob, and 8 isn't a break-even

table/table.go:1253-1254 documents WithMaxConcurrency as setting "the maximum concurrency for table scan and plan operations". The local planning path honours scan.concurrency (table/scanner.go:1121,1126,1247,1250); the remote path now hardcodes 8, and table.ScanPlanningRequest has no field to carry the option, so it cannot reach the planner at all. Every other SetLimit in the repo derives from GOMAXPROCS or an option — table/orphan_cleanup.go:511,561, table/transaction.go:2654, table/snapshot_producers.go:463, table/arrow_scanner.go:100,233, table/equality_delete_reader.go:553. This introduces a second convention.

And 8 isn't a knee. Extending your own benchmark shape with a concurrency axis (10 ms/req fake server, medians of 3×3):

handles	1	2	4	8	16	32	64	128
64	680 ms	333	169	84.2	42.3	22.1	11.9	—
256	2990 ms	1426	690	351	201	95.9	47.0	26.9

Scaling is linear in the limit out to 128 — the work is latency-bound, so there's no knee anywhere near 8. The chosen value sits mid-slope and leaves 7.1× on the table at 64 handles and 13× at 256. Either wire WithMaxConcurrency through ScanPlanningRequest and default to min(GOMAXPROCS, len(handles)) like the rest of the repo, or justify 8 as a deliberate politeness cap toward rate-limited catalogs — but say so in the comment, since it currently only explains the frontier split, not the number.

What checks out

Worth recording, because I tried hard to break these and couldn't:

• Ordering is genuinely deterministic. 60 random fanout graphs (24 handles, random children including self/back edges producing cycles, 1–4 duplicate roots), 3 reps each, randomized 0–4 ms per-request latency, under -race, compared against a verbatim copy of the pre-change serial BFS: envelope count and full file-path order identical in all 180 runs.
• Bound is exactly 8 (64 slow handles → max observed in-flight 8).
• No goroutine leak, prompt cancellation — parent cancel() mid-frontier returns in <1 ms with context.Canceled; group.Wait() joins every worker.
• Post-failure request amplification is benign — 200 handles with h0 404ing issues only 8 requests and returns in 3 ms, because http.Client refuses to dial an already-cancelled context.
• Benchmark claim reproduces: reverting only the scan_planning.go hunk gives 671.3 ms → 93.7 ms = 7.16× (you claimed ~8× on an M1 Pro). Not a stale table.
• -race -count=5 clean; golangci-lint 0 issues; CI 15/15. git diff against the merge base shows zero removed test lines, so nothing was weakened.

Minor

• :375 — the fallback returns waitErr unwrapped, so the handle context this commit exists to add is absent exactly when every error was a cancellation.
• :370 — filters context.Canceled only, not context.DeadlineExceeded.
• :314-324 — the terminal iteration calls fetchScanTaskFrontier with an empty handles slice, allocating an errgroup and two slices to do nothing. if len(handles) == 0 { break }.
• :340 — fetchScanTaskFrontier is the only function in this 1286-line file with no doc comment.
• On failure the concurrent path issues up to limit requests the serial path never would (8 vs 1). Worth a sentence in the collectScanTasks doc.
• scan_planning_test.go:1231 — fetched = append(...) from the HTTP handler with no mutex; safe only because every frontier in that test has exactly one handle. Latent trap.
• Allocations roughly double (834 KB/8467 → 1.74 MB/10815 at 64 handles). Fine for a 7× win, just undisclosed.

Prior items

1. Concurrent failures produce a timing-dependent error → Still open. The structure I asked for was implemented, but the exclusion is inert (above). Slightly regressed: the error now carries a false handle label.
2. Fetch concurrency isn't tunable → Still open. catalog/rest/options.go isn't in the diff. I'd upgrade this to Major given WithMaxConcurrency already documents itself as covering plan operations.
3. @laskoviymishka's :329 var nextFrontier []string → Fixed. :354 handle wrapping → fixed in form, broken in substance (above). schema.go:357 and schema_test.go:2307 comments → Fixed.

Description

• The schema.go / schema_test.go changes (2 of 5 files) aren't mentioned. The docs(schema) head commit is not a stray unrelated change — it answers review items — but it's cross-package and invisible from the description. One line in the Summary would stop a reader being surprised by a root-package diff on a perf(rest) PR.
• The benchmark table isn't reproducible from the committed tree: "Serial baseline (limit 1)" requires editing a const, and BenchmarkCollectScanTasks64Handles has no concurrency axis. Parameterizing over the limit would also produce the scaling data the Major item wants.

---

This review was drafted by an AI-assisted tool and confirmed by an Iceberg Go maintainer. The findings cite the project's review criteria; if you think one is mis-applied, please reply and a maintainer will weigh in.

[zeroshade]

The concurrent frontier fetch preserves handle order and deterministically reports the first error in serial handle order, CI is green at the new head, and both prior review threads were answered — but the description still advertises a fixed 8 workers when the code now derives concurrency from the scan's own limit.

Re-review verification: 4 of 6 prior findings confirmed fixed at ca8224f (each verified by mutating the fix and observing the suite go red, not by taking the claim on trust). Still open:

• not fixed — Each BFS level waits on its slowest handle before the next level starts, unlike Java's shared queue
• partially fixed — Schema.MarshalJSON alias literal hard-codes two fields; a new json-tagged field would be silently dropped

Verification performed

In worktree .pi-worktrees/pr1959 at ca8224f: 'go build ./...' PASS; 'go vet ./catalog/rest ./table' clean; 'go test -race -count=2 ./catalog/rest' PASS (12.972s); 'go test -race -count=1 -run TestCollectScanTasks ./catalog/rest' PASS (1.931s); 'go test -count=1 ./catalog/rest' PASS (5.918s); worktree left clean (git status --porcelain empty). PR CI at ca8224f: all checks green (the earlier failing build/test jobs were on the superseded commit 96acb8d).

---

This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Go maintainer. The findings below are observations, not blockers; an Apache Iceberg Go maintainer — a real person — will take the next look at the PR. If you think a finding is mis-applied, please reply on the PR and a maintainer will weigh in.

More on how Apache Iceberg Go handles maintainer review: CONTRIBUTING.md.

[zeroshade]

minor — PR description still advertises a fixed 8 workers; the code derives concurrency from the scan

The summary and benchmark table describe 'up to 8 concurrent REST requests' and an '8 workers' row, but concurrency now comes from req.MaxConcurrency (populated from the scan's WithMaxConcurrency) and falls back to runtime.GOMAXPROCS(0), not 8. Since the tunability request was addressed in code, please update the description and re-label the benchmark row, otherwise the recorded numbers cannot be reproduced from the stated configuration.

[zeroshade]

minor — Hand-maintained alias literal can silently drop a future Schema field

Replacing *(*Alias)(s) with Alias{ID: ..., IdentifierFieldIDs: ...} is the right fix for copying the lazy atomic caches, but correctness is now maintained by comment alone: adding a json-tagged field to Schema would drop it from the serialized output with no compile error and no failing test. A small reflection-based test asserting the marshaled key set matches Schema's json-tagged exported fields would make the invariant self-enforcing.

[zeroshade]

nit — collectScanTasks is now reachable only from tests

PlanFiles calls collectScanTasksWithConcurrency directly, so the collectScanTasks wrapper survives solely as a test entry point with a GOMAXPROCS default. That is harmless, but a reader will assume it is the production path. Either drop it and have the four tests pass an explicit limit, or note in its doc comment that it is the default-concurrency convenience wrapper.

[zeroshade]

All three prior findings are genuinely fixed, and the previously unverified ordering/error-determinism and sibling-cancel guarantees are now proven non-vacuous by five in-place mutations that each turn a specifically named test red.

Re-review verification: 4 of 4 prior findings confirmed fixed at a954db0 (each verified by mutating the fix and observing the suite go red, not by taking the claim on trust).

Verification performed

All in worktree .pi-worktrees/pr1959 at a954db0. Baseline: go build ./... OK; go vet ./catalog/rest/ . OK; go test -race -timeout=300s ./catalog/rest/ -> ok 7.459s (full package); go test -timeout=300s -run 'TestMarshalSchema|TestSerializeSchema|TestUnmarshalSchema|TestSchema' . -> ok. Mutation runs (each restored via git checkout -- <file>): 5 separate in-place mutations of scan_planning.go and schema.go, each confirmed RED against its named test with explicit -timeout=120s. Final state: git status --porcelain empty, go build ./... and go vet clean.

---

This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Go maintainer. The maintainer approving this PR has read the findings and signed off. If something feels off, please reply on the PR and a maintainer will follow up.

More on how Apache Iceberg Go handles maintainer review: CONTRIBUTING.md.

[zeroshade]

nit — Unreachable maxConcurrency normalization in fetchScanTaskFrontier

fetchScanTaskFrontier re-applies the 'if maxConcurrency <= 0 { maxConcurrency = runtime.GOMAXPROCS(0) }' fallback, but its only caller (collectScanTasksWithConcurrency, scan_planning.go:337) already normalized the value at :316. The branch is dead for every current call path. Harmless defensiveness in an unexported helper; drop it or keep normalization in exactly one place.