iceberg: cut shredder allocations, and add write/commit performance harnesses

[Jeffail]

Some iceberg sink performance work. One production change plus the measurement tooling that motivated it — happy to split them if you'd rather review separately.

The production change: shredder allocations (2e80f3e)

Shredding a record built two maps per struct — an index of input keys by their match-key, and a set of keys that matched a schema field. Both exist for case-insensitive matching, where several input keys can fold onto one schema field and that ambiguity needs reporting rather than silently resolving.

When matching is case-sensitive though, the match-key is the identity function, so neither map earns its keep: a field's value is just value[field.Name], and case-collisions can't happen because Go map keys are themselves case-sensitive. That's the default (case_sensitive_columns: true), so from what I can tell it's the path almost all traffic takes.

So this splits the two apart — shredStructExact for case-sensitive matching with direct lookups, and the existing body moves to shredStructFolded unchanged. Unknown-field detection now counts how many input keys a schema field claimed and skips the scan entirely when they're all accounted for (the steady state once a schema settles), building a lookup set only on the rare path where something genuinely is unknown.

BenchmarkShredWide at GOMAXPROCS=1, benchstat, n=8:

                sec/op                      B/op                        allocs/op
before          4.369µ                      4.312Ki                     71
after           1.472µ  -66.3% (p=0.000)    1.609Ki  -62.7% (p=0.000)    41  -42.3% (p=0.000)

Worth being clear that's the shredder in isolation. Earlier profiling put shredding at roughly 27% of the sink's CPU, so my read is the end-to-end saving should be appreciable but a good deal smaller than 66% — I haven't measured that yet, so please treat the sink-level number as unquantified rather than implied.

Since it's purely an optimisation it shouldn't change observable behaviour at all, so TestShredStructPathsAgree runs the same schema and record through both implementations and requires identical emitted values (in order), identical new-field notifications (compared as sets, since map iteration order is random) and identical errors. Twelve cases: full matches, the exact-match steady state, one and several unknown keys, an unknown key nested inside a known struct, missing optionals, explicit nulls, both required-field error paths, and the empty-record / empty-schema edges.

The tooling (17f8f54, 42e4bc6)

No production code, all flag-gated or skipped by default:

1. A throughput bench for the Databricks e2e harness plus local profiling configs and a wide-schema shredder micro-benchmark (the one the numbers above come from).
2. A commit-regime harness — wraps the existing in-memory test catalog with a configurable per-commit delay and a commit counter, then drives N concurrent submitters at it and reports records/sec, records/commit and submissions/commit across commit latency, max_in_flight and records-per-submission. Nothing writes parquet or touches object storage, so encode and upload cost don't confound it.

A fixed injected delay seems like a fair stand-in for a real catalog here: measurement against a live engine-backed catalog found commit latency near-flat across a 667x range of batch sizes, so latency looks roughly constant with respect to batch size — and unlike a hosted service you can sweep it across regimes instead of being pinned to one.

[claude]

Test coverage: the nested comparison in this test is vacuous.

The shredder is constructed with caseSensitive: true, and only the top-level call is dispatched explicitly. Nested structs are reached via shredValue → shredStruct, and shredStruct dispatches on rs.caseSensitive — so it routes to shredStructExact for both the "exact" and the "folded" run.

Result: below the top level both sinks are produced by the same implementation, so the "unknown key nested inside a known struct" and "all fields present" cases compare shredStructExact against itself. Any divergence in nested struct handling (nested unknown-field notification, nested required-field errors, nested path cloning) cannot be detected, even though the test doc-comment and the commit message both claim nested coverage — and this test is the stated safety net for a behaviour-preserving optimisation.

Suggested fix: drive the comparison through the public entry point with two shredders instead of calling the internal helpers directly — NewRecordShredder(schema, true).Shred(record, sink) vs. NewRecordShredder(schema, false).Shred(record, sink) — so the whole recursion (top level and nested) goes exact-all-the-way on one side and folded-all-the-way on the other. Keep the direct shredStructExact/shredStructFolded calls only if you additionally want the top-level-only assertion.

[Jeffail]

Good catch, and you are right that it was vacuous below the root — shredStruct dispatches on rs.caseSensitive on every recursion, so with one case-sensitive shredder both runs went through shredStructExact for nested structs and the test compared the fast path against itself. Fixed in 1bf1770 by driving both runs through the public Shred entry point with two shredders, as you suggested, so one side is exact all the way down and the other folded all the way down.

I dropped the direct helper calls entirely rather than keeping them alongside — the public-entry comparison subsumes the top-level assertion, so keeping both seemed like it would only add a second thing to keep in sync.

Two things I added on top:

1. Three doubly-nested cases, so the nested error and notification paths are actually reachable: an unknown key two levels down, and a required leaf both missing and explicitly null two levels down.
2. A check that the test now has teeth — I injected a mutation that drops nested unknown-field notifications from the fast path only, and both nested cases fail, where by your argument they could not have before.

One caveat now called out in the doc-comment: the comparison is only legitimate for a case-unambiguous corpus, so every field name and record key in it is lower-case. Input that differs in case is exactly where the two modes are meant to diverge, so that belongs in the case-sensitivity tests rather than here.

[claude]

Benchmark results not recorded (CONTRIBUTING.md §1.3.4).

This is a hot-path serialization change with measured throughput numbers quoted in the commit message (sec/op 4.369µ → 1.472µ, -62.7% B/op), and the PR also adds files under internal/impl/iceberg/bench/. CONTRIBUTING.md §1.3.4 requires following the reporting requirements in docs/benchmarking.md and recording results under docs/benchmark-results/, and docs/benchmarking.md "Keeping Results Up to Date" is explicit on both counts:

2. When modifying a connector's performance path — Re-run the benchmark and append a new dated section to the results file. This includes changes to batching, buffering, connection handling, serialization, or any code that sits in the hot path.
3. During code review — […] It will flag PRs that add or modify bench/ directories without updating results files, and PRs that include throughput numbers in the description without recording them in docs/benchmark-results/.

docs/benchmark-results/iceberg.md already exists but is untouched by this PR. Please append a dated section to it with the BenchmarkShredWide before/after numbers (plus environment and PR link). The commit message notes the end-to-end effect "has not been measured yet" — that's fine to state in the section, but the micro-benchmark delta and the new bench harnesses still need recording there.

Separately, profile_config.yaml / profile_config_schema.yaml are not mentioned in internal/impl/iceberg/bench/README.md nor wired into the directory's Taskfile.yaml, which docs/benchmarking.md §4 and §6 ask for.

[Jeffail]

Fair, and I had missed that docs/benchmarking.md is explicit about both halves of this. Addressed in e9c8bd2.

Appended two dated sections to docs/benchmark-results/iceberg.md: the BenchmarkShredWide before/after (benchstat over n=8, with environment, PR link and what changed) and the commit-regime sweep, each with its reproduction command.

I have tried to make both hard to over-read, since that felt like the real risk in recording a -66%:

• the shredder section states plainly that the number is the micro-benchmark in isolation, that earlier profiling put shredding at ~27% of sink CPU so the end-to-end effect should be much smaller, and that no sink-level throughput figure in the file has been re-measured for this change;
• the commit-regime section notes it writes no parquet and touches no object storage, so its rec/sec are ratios for comparing coalescing behaviour and are not comparable with the localhost or live-catalog sections.

On the second half: profile_config.yaml and profile_config_schema.yaml are now documented in the bench README and wired into the Taskfile as bench:profile and bench:profile:schema, plus a bench:shredder task for the micro-benchmark (needs no infrastructure). I ran bench:shredder to check the wiring rather than just that the YAML parses.

Still outstanding and not something I wanted to guess at: the CHANGELOG entry. 4.105.0 is released and there is no Unreleased heading, so I would be inventing a version number — happy to add it wherever you would like it.

[claude]

require called from a non-test goroutine.

This require.NoError (and the require.NoError inside recordCountDataFile, also reached from these goroutines) runs on a goroutine that is not the one running the test. require failures call t.FailNow(), which testify documents as only valid from the test goroutine — it invokes runtime.Goexit(), so on a Commit failure the goroutine unwinds mid-way and the test then proceeds to evaluate the commits assertions against a half-completed run, producing a confusing failure rather than the real cause.

The project's test patterns call this out directly: "require calls FailNow() which panics when called from a non-test goroutine."

Suggested fix: have each goroutine capture its error (e.g. into a per-index slice or an errors.Join-able collector, or build the data files on the test goroutine before spawning) and assert with require after wg.Wait() on the test goroutine. assert.NoError inside the goroutine would also be safe if you prefer to keep the check in place.

[Jeffail]

You are right, and thanks — this one would have been genuinely misleading in failure. require calls FailNow, which is only valid on the test goroutine, so on a commit failure the submitter unwound mid-loop via runtime.Goexit and the test then evaluated its commit-count assertions against a half-finished run, reporting a count mismatch rather than the actual error.

Fixed in 1bf1770. recordCountDataFile now returns an error instead of asserting, submitters record their first error, and the test goroutine asserts on errors.Join(errs...) after wg.Wait() — before deriving any rate, since a rate from a partial run would be meaningless anyway. The coalescing test now builds its data files up front on the test goroutine so its submitters only have to commit. Both pass under -race.