perf(table): parallelize non-bulk PurgeFiles deletion

[fallintoplace]

Summary

• Parallel deletes: use up to 32 workers for non-bulk PurgeFiles deletion.
• Keep bulk deletion as the first choice and tolerate missing files.
• Collect deletion errors in input order and stop queued work on cancellation.
• Document concurrent IO.Remove calls and add an unreleased compatibility note for custom IO implementations.
• Use atomic counters in the Glue and Hive purge test callbacks.

Tests

• make test
• go test -race ./table ./io/... ./catalog/glue ./catalog/hive -count=1
• golangci-lint run --allow-parallel-runners --timeout=10m
• Regression checks: both cancellation and worker-limit tests pass 150 repetitions each across 1, 2, and 8 CPU settings.
• Removing either the cancellation guard or worker cap makes its test fail in all 60 mutation runs.

Benchmark

10,000 files with a simulated 100µs deletion delay. Median of three local runs on darwin/arm64, using -benchtime=500ms:

Workers	Time per purge
1	1,812ms
4	434ms
16	253ms
32	253ms
64	78ms

[zeroshade]

The parallelisation itself is sound — I checked it carefully because this path deletes files — but the branch has conflicts and needs a rebase before it can land.

Please rebase

mergeStateStatus is DIRTY. Thirteen PRs from a related stack merged earlier today, and the duplicated schema.go lazy-cache hunk this branch carries is already on main, which is the likely conflict — it should simply vanish on rebase.

What I verified

For a change to a destructive operation, these are the properties that matter:

• The delete set is unchanged. Same sorted fileSet, and the bulk path is untouched. Parallelisation changes concurrency and ordering only, never which files get deleted. That was the blocking-class question and it's clean.
• No metadata mutation follows the deletes, so a partial failure can't leave metadata asserting work that didn't happen.
• Deterministic error aggregation. errors.Join in per-index order, so multiple failures produce a stable, complete error rather than whichever one landed first. Worth calling out explicitly — a sibling PR (#1969) picked up a finding for exactly that pattern, and this one gets it right.
• NotFound is tolerated in the PurgeFiles closure via os.IsNotExist, which is correct for a retryable destructive operation — a re-run shouldn't fail because the first attempt already succeeded.
• Bounded fan-out. min(max(maxConcurrency,1), len(files)), so no unbounded concurrent deletes against object storage and no self-inflicted rate limiting.
• Race safety. Per-index deleted/errors slices, no shared accumulator. Context is checked before dispatching jobs, and cancellation surfaces as an error.

Minor — no mid-flight cancellation test (table/orphan_cleanup.go:763-774)

Coverage includes partial failures, pre-cancelled queued work, and the concurrency bound — the gap is specifically cancellation while deletes are in flight. On a file-deletion path that's where a partial-state bug would hide, so a test asserting the error makes clear the operation was incomplete would be a worthwhile permanent guard.

Follow-up observation — not this PR

The generic non-bulk DeleteOrphanFiles path still treats NotFound as an error, while the PurgeFiles path now tolerates it. Pre-existing and outside this diff, but the two paths now diverge in retry behaviour, which is worth knowing about and probably worth reconciling separately.

CI green, all 3 commits signed off.

---

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]

The shape of this is right, and the recent rebase cleaned up the one piece of noise: the MarshalJSON change had already landed on main as a separate fix, so dropping it here is exactly right.

The worker-pool redesign is the strong part. The deleted and deleteErrors slots are index-disjoint so the workers are lock-free by construction, error aggregation walks the input in order so the returned error is deterministic, and wg.Wait() gives a clean happens-before before those slots are read back.

I'd hold this before merging though, mostly around the concurrency default and one test.

The bigger one: PurgeFiles sizes the pool with runtime.GOMAXPROCS(0), but what we're parallelizing is object-store round trips, not CPU work. On a typical cloud box that's 2-8 concurrent deletes, where object-store clients usually run 32-64 in flight. It's a real win over sequential, but for a perf PR it leaves close to an order of magnitude on the table, and unlike the orphan-cleanup path there's no option to tune it. I'd either raise the default to something I/O-appropriate or add a WithPurgeMaxConcurrency mirroring WithCleanupMaxConcurrency.

The other: TestDeleteFilesParallelStopsQueuedWorkOnCancellation only covers the pre-cancelled case. With one file the pool collapses to a single worker and the sender's ctx.Done() fires before anything reaches jobs, so the mid-dispatch guard the test is named for never runs, and the test would still pass if that guard were deleted. I'd add a case with more files than workers that cancels mid-flight.

A few smaller things I'd like to settle before merge:

• make the cancellation error placement in deleteFilesSequential match deleteFilesParallel (join it last), and note in the PurgeFiles doc that non-bulk errors now come back as a single joined value
• the benchmark's deleted[len(deleted)-1] is only safe because deleteFunc never fails; a wrapping wrapError guards it and surfaces failures
• a one-line comment on the inner ctx.Err() priority-drain check

Once those are settled, happy to take another pass and approve.

[laskoviymishka]

runtime.GOMAXPROCS(0) is a CPU-count heuristic, but what we're parallelizing here is object-store round trips, not CPU work. On a typical cloud box that caps us at 2-8 concurrent deletes, and since throughput is roughly concurrency over latency, 4 workers at ~100ms/op is ~40 files/sec. That's a real win over sequential, but object-store clients usually run 32-64 in flight, so we're leaving close to an order of magnitude on the table.

The orphan-cleanup path already exposes WithCleanupMaxConcurrency; PurgeFiles has no knob at all. I'd either bump the default to something I/O-appropriate (32 is defensible) or add a WithPurgeMaxConcurrency option mirroring the cleanup one, and note in the doc that GOMAXPROCS is a CPU default that wants raising for remote stores. wdyt?

[laskoviymishka]

This only exercises the pre-cancelled case: ctx is already cancelled before the call, and with a single file workers is min(max(4,1), 1) == 1, so the sender's case <-ctx.Done() fires immediately and nothing ever lands on jobs. The post-receive guard (if err := ctx.Err(); err != nil inside the worker) never runs, so this would still pass if that guard were deleted.

I'd add a second case that actually exercises mid-dispatch: say 50 files with 4 workers, cancel from inside deleteFunc after a handful of calls complete, then assert the deleted count is well short of 50 and err carries context.Canceled. That's the path that proves queued work actually stops. wdyt?

[laskoviymishka]

Minor asymmetry with the parallel path: here the context error gets errors.Join'd into result inline, mid-chain with any delete failures, whereas deleteFilesParallel appends cancellationErr last. Both share the same caller in deleteFiles, so depending only on how maxConcurrency slices the same input, callers get structurally different error nesting. errors.Is(context.Canceled) works either way, but string and unwrap inspection differ.

I'd accumulate the context error separately here and join it last too, so the two paths produce the same shape. While we're at it, a line in the PurgeFiles doc noting non-bulk errors now come back as a single joined value would help, since the old path returned per-file errors directly.

[laskoviymishka]

deleted[len(deleted)-1] is safe only because this bench's deleteFunc always succeeds, so every index ends up in deleted. Worth noting the underlying shape though: if a delete fails and wrapError returns nil (as it does here), that index lands in neither deleted nor deleteErrors and silently drops from both return slices. Extend this bench to model partial failures and deleted could be empty, so this indexes out of bounds.

Simplest guard is to make wrapError actually wrap (func(_ string, err error) error { return err }); that keeps the sink safe and surfaces any failure through b.Fatal. wdyt?

[laskoviymishka]

The outer case <-ctx.Done() and this inner ctx.Err() check look redundant but both are load-bearing: when a job and ctx.Done() are both ready, select picks at random, so this inner check is what stops a worker from doing real I/O after cancellation when the random pick happened to land on jobs. Correct and known pattern, just subtle. A one-line comment here would save the next maintainer from re-deriving it.

[laskoviymishka]

Small one: var allErrors []error reads cleaner here. errors.Join treats nil and empty-non-nil the same, so the explicit make([]error, 0) signals an intentional non-nil-empty that doesn't actually matter, and it trips the perfsprint-style lints.

[zeroshade]

Rebase done, and both my items and @laskoviymishka's are addressed. I re-verified the data-safety properties from scratch rather than re-reading the diff, including mutation-testing the new cancellation test. Approving, with three follow-ups below that I don't think need to block.

What I verified this round

Partial failure deletes everything else and aggregates every error. 40 files, 8 workers, failures injected at indices 3/17/39: 40 attempted, 37 deleted, and the joined error names all three failing paths in input order. No silent drops, no first-error-wins truncation.

Missing files are still success. With the PurgeFiles closure shape (fs.Remove + os.IsNotExist swallow), all three of gone-1 / ok / gone-2 come back as deleted with err == nil — so a re-run after a partial purge doesn't fail on already-deleted files.

Concurrency bound is real and is 32. 200 files, all workers held open simultaneously: maxActive == 32 == defaultPurgeMaxConcurrency. Never exceeded.

The new mid-flight cancellation test is load-bearing. I mutation-tested it rather than trusting it: removing the sender's case <-ctx.Done(): makes TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation deadlock and fail (goroutine [chan send] at orphan_cleanup.go:820, 90 s timeout). So it genuinely pins the queued-work-stops property, which is what both of us asked for.

Rebase confirmed: 0 commits behind main, PR is the top 3 commits on fc0e6fca, and the duplicated schema.go hunk is gone as expected.

go build ./... clean; targeted tests -count=20 pass; -race -count=5 pass; gofmt clean on all 5 files; golangci-lint run ./table/... → 0 issues.

Major — PurgeFiles now calls iceio.IO.Remove concurrently, and nothing says so

This is the one thing I'd like addressed before or shortly after merge. io/io.go's IO interface documents Open and Remove with no mention of goroutine-safety — I grepped, there is no concurrency contract anywhere in the file. Before this PR, PurgeFiles' non-bulk path called Remove from a single goroutine; now it calls it from up to 32.

The evidence that this bites is inside this PR: catalog/glue/glue_test.go and catalog/hive/hive_test.go are only in the diff because their failRemoveIO callbacks had to be converted from bool/int to atomic.Bool/atomic.Int64 (commit 362c27f3). Two in-repo IO stubs needed fixing; any user with a custom iceio.IO whose Remove touches unsynchronized state now gets a data race on drop-table-with-purge, with no release-note signal.

Suggested fix: a sentence on Remove in io/io.go stating implementations must be safe for concurrent use, and a line in the PurgeFiles doc noting non-bulk deletion invokes Remove concurrently. Cheap, and it converts a silent breaking change into a documented one.

Minor

1. The inner worker guard is still uncovered. orphan_cleanup.go:803-807 — the ctx.Err() check inside the case index, ok := <-jobs: arm. @laskoviymishka's original point was that the old test would still pass if that guard were deleted; that is still true of the new test. I removed the guard and ran TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation at -count=50: all pass. The reason is that once the sender breaks out and closes jobs, nothing more arrives, so calls stays at 4 regardless of the inner check.

The guard is genuinely load-bearing in production — sender blocked in jobs <- index plus a cancelled context means the worker's select sees both arms ready and picks at random — it's just not pinned by a test. The comment you added at :801-802 explaining why it exists is good and I'd keep it either way.

2. The benchmark grid doesn't include the shipped default. orphan_cleanup_bench_test.go:140 sweeps concurrency ∈ {1, 4, 16}, but defaultPurgeMaxConcurrency is 32. The value that actually ships is the one value not measured, and the PR description's table (1/4/16 workers) has the same gap. Adding 32 — and 64, since the point of raising it was that object stores run deeper queues — would justify the constant with data instead of reasoning.

3. Latent trap in the deleteFilesParallel contract. If wrapError returns nil for a file whose deleteFunc failed, that index lands in neither deleted nor deleteErrors and vanishes from both return values. Probed directly: files [a b c] with b failing and a nil-returning wrapper gives deleted=[a c], err=<nil> — b is silently dropped.

Not reachable today: both production call sites always wrap non-nil, and the benchmark's wrapper now passes the error through with a b.Fatal guard, which resolves @laskoviymishka's instance of this. But the helper is unexported and general, so a future third caller could reintroduce it. Either document that wrapError must return non-nil, or treat a nil wrap as "deleted" so the file can't disappear from both lists.

Prior items

Mine (COMMENTED, 2026-09-01):

• Rebase — Fixed, verified 0 behind main.
• Mid-flight cancellation test — Fixed, and mutation-verified load-bearing (above).
• Follow-up observation that DeleteOrphanFiles still treats NotFound as an error while PurgeFiles tolerates it — still open, still out of scope. Worth a separate issue.

@laskoviymishka (CHANGES_REQUESTED, 2026-09-01):

• runtime.GOMAXPROCS(0) too low for I/O — Fixed. defaultPurgeMaxConcurrency = 32, which is one of the two options offered. Still no WithPurgeMaxConcurrency knob, but the ask was either/or.
• Mid-dispatch cancellation test — Partially fixed. The new test genuinely exercises mid-flight cancellation, but the specific inner guard named in the comment remains uncovered (Minor 1).
• Sequential path should join the cancellation error last — Fixed, :763 errors.Join(result, cancellationErr), matching :842-844.
• PurgeFiles doc note that non-bulk errors come back joined — Fixed, doc updated.
• Benchmark deleted[len(deleted)-1] guard — Fixed differently but adequately. Rather than making wrapError wrap, you made it pass the error through and added if err != nil { b.Fatal(err) }, which closes the out-of-bounds path.
• One-line comment on the inner ctx.Err() priority-drain check — Fixed, :801-802.
• var allErrors []error instead of make([]error, 0) — Fixed, :833.

Description

Accurate on every claim I checked. Two gaps: it doesn't mention the catalog/glue and catalog/hive test files (2 of 5 changed files) or why they needed atomics — which is the clearest signal of the concurrency contract change in Major above — and the benchmark table stops at 16 workers while the code ships 32.

---

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]

Production code is correct and race-free (probe-verified: bound enforced exactly, cancellation guard fires 119/300 runs), but the two tests meant to pin the cancellation guard and the concurrency cap both pass unchanged when those mechanisms are deleted, so the prior review's central finding remains unfixed.

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

• partially fixed — Cancellation test only exercises the pre-cancelled case; would still pass if the post-receive guard were deleted. Add a mid-dispatch case with 50 files/4 workers.

Verification performed

go build ./... => OK; go vet ./table ./io ./catalog/glue ./catalog/hive => OK; go test -race -count=1 ./table/ => ok 22.460s; ./catalog/glue/ => ok 3.391s; ./catalog/hive/ => ok 2.887s; ./io/ => ok 3.338s; go test -race ./io/... ./catalog/hadoop/ ./catalog/sql/ => all ok; benchmark BenchmarkPurgeFilesNonBulkDeletion runs clean and reproduces the claimed speedup (10k files `@100us`: 1.55s at concurrency=1 vs 40.2ms at 32). Mutation runs: guard-removal 200/200 pass (should fail); worker-bound removal 50/50 pass (should fail); defaultPurgeMaxConcurrency=1 correctly FAILS; nil-wrapper removal correctly FAILS. Worktree left clean at 9372950 (git status empty, probes deleted).

---

This review was drafted by an AI-assisted tool and confirmed by an Apache Iceberg Go maintainer. After you've addressed the points above and pushed an update, an Apache Iceberg Go maintainer — a real person — will take the next look at the PR. 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 Apache Iceberg Go handles maintainer review: CONTRIBUTING.md.

[zeroshade]

major — Mid-flight cancellation test still does not reach the post-receive ctx guard

TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation calls cancel() and only then close(release), so ctx.Done() is always ready strictly before any worker can return to the case index, ok := <-jobs receive. The sender therefore always wins the select race and no worker ever reaches the inner guard at orphan_cleanup.go:801-806. This is the exact defect the prior review named ('would still pass if that guard were deleted') and it is still true. Fix: close(release) BEFORE cancel() (or release a subset first) so workers are contending on the jobs receive at the moment of cancellation, and assert deleteFunc is not invoked again after cancellation is observed.

Evidence

Deleted the guard block at orphan_cleanup.go:801-806 -> `go test -count=200 -run TestDeleteFilesParallelStopsQueuedWorkOnMidFlightCancellation ./table/` => `ok ... 0.582s`; full `go test -race -count=1 ./table/` => `ok ... 20.047s`. Instrumented guard with an atomic counter: 'guard fired total: 0' across TestDeleteFiles*/TestPurgeFiles*/TestOrphan*. An unbarriered probe on the same code recorded 119 firings across 300 cancelled runs, proving the guard is reachable and load-bearing -- just never by the suite.

[zeroshade]

major — assert.LessOrEqual(fsys.MaxActive(), maxWorkers) is vacuous - the concurrency cap is unverified

purgeDeleteTrackingIO is constructed with targetActive: 2 (line 1838), so the test releases the barrier as soon as two removals are concurrently active. The worker pool never saturates, so MaxActive() never approaches the cap and the upper-bound assertion can never fail. Bounded concurrency is the headline safety property of this PR and nothing pins it. Fix: set targetActive to defaultPurgeMaxConcurrency so all workers are held in flight, then assert MaxActive() == defaultPurgeMaxConcurrency exactly.

Evidence

Replaced `workers := min(max(maxConcurrency, 1), len(files))` (orphan_cleanup.go:773) with `workers := len(files)` -- 64 workers against a documented cap of 32 -- and `go test -count=50 -run TestPurgeFilesDeletesNonBulkFilesConcurrently ./table/` => `ok ... 0.556s` (50/50 pass). A probe using a barrier that actually holds every worker observed the correct behaviour: 'PROBE-A bound=8 files=64 observedPeak=8', confirming the code is right and only the assertion is toothless.

[zeroshade]

minor — Defensive branch for a condition no caller can produce, plus a test that exists only to cover it

if wrappedErr == nil { wrappedErr = err } guards against a wrapError that returns nil. Both production call sites (orphan_cleanup.go:731 and :1409) unconditionally return a non-nil fmt.Errorf, so this is unreachable in production; the only caller that can trigger it is TestDeleteFilesParallelPreservesErrorWithNilWrapper, written for that purpose. The prior review asked for the bench's wrapError to be fixed (it was); this production fallback is extra. AGENTS/house style: don't add error handling for scenarios that can't happen. Suggest dropping both the branch and the test.

[zeroshade]

minor — Retroactive concurrency contract on the public IO interface deserves a release note

'Implementations must be safe for concurrent use by multiple goroutines.' tightens the contract of an exported extension point. Downstream users who registered a custom IO via io.Register and whose Remove was only ever driven serially by PurgeFiles will now be called from up to 32 goroutines. Every in-repo implementation already complies, so nothing breaks here, but this is a behavioural change for third-party implementors and should be called out in the changelog rather than only in a doc comment.

[zeroshade]

Both prior majors are verifiably fixed — deleting the post-receive ctx guard and inflating the worker cap each turn a named test RED — leaving only an untested (but trivially correct) sequential cancellation guard.

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

Verification performed

go build ./... (OK); go vet ./table/... ./io/... (clean); go test ./table/ -race -timeout=300s (ok, 23.657s); targeted -race runs of TestDeleteFilesParallel*/TestPurgeFiles*; 4 in-place mutations (post-receive ctx guard removed -> RED; workers cap +2 -> RED; recordCancellation no-op -> RED on both cancellation tests; sequential ctx guard removed -> GREEN, gap); throwaway pr1973_probe_test.go with 300-iter randomized-cancellation invariant probe, goroutine-leak check, and 200-case parallel-vs-sequential differential probe (all pass, no races), since deleted.

---

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]

minor — New ctx cancellation guard in deleteFilesSequential is pinned by no test

This PR adds context cancellation to deleteFilesSequential (the function previously took no ctx at all). The guard is production-reachable: deleteFiles routes here when cfg.maxConcurrency <= 1, and maxConcurrency defaults to runtime.GOMAXPROCS(0) (orphan_cleanup.go:284), so single-CPU containers take this path by default; WithCleanupMaxConcurrency(1) also selects it. The parallel twin is now well pinned, but this one is not. Suggest a small test that cancels ctx after N sequential deletes and asserts the returned error is context.Canceled and that len(deletedFiles) < len(orphanFiles).

[zeroshade]

nit — assert.Greater(t, maxWorkers, 1) is a compile-time-constant tautology

maxWorkers is 'const maxWorkers = defaultPurgeMaxConcurrency' (32), so this asserts 32 > 1 and exercises no code under test. The adjacent assert.Equal(t, maxWorkers, fsys.MaxActive()) already carries the real signal. Harmless, but it is exactly the kind of assertion that reads as coverage without providing any.