parquet: json_parquet_encode allocates a third as often, reuses writers, caches by metadata

[almostintuitive]

Summary

Stacked on #3. Follow-ups to json_parquet_encode, one commit each, each checked against the replaced pipeline by the parity tests and the fuzz target. They're followed by fixes for everything an adversarial review of the first version found.

1. Drop the write buffer and per-row key strings.
2. Parse and coerce without allocating. Keys and fields are read as raw JSON. A string or number is coerced by the same strconv call the Bloblang coercion makes; bools and nested values still go through the original helpers.
3. Keep each partition's rows in its own arena. Each partition's values and bytes come from chunks that start at one row and 64 B and double up to 256 rows and 16 KiB. They're freed when the partition's file is written.
4. columns[].cache_by (opt-in). An interpolated column is evaluated once per batch for each distinct combination of the listed metadata fields.
5. Reuse writers through a pool, reset for each file.
6. Fewer, larger files. The benchmark now covers one file, few, many, and one per message. The docs explain what each file costs, and a json_parquet_encode_files counter makes the file count visible in production.

Review fixes

• Long strings were corrupted. A text value over 4 KiB was "copied" with []byte(b) of a []byte, which doesn't copy. The next message parsed would overwrite it, including with bytes from other partitions. It now always copies. A new parity test and new fuzz seeds send long strings among other messages; both fail without the fix.
• cache: true could key the cache on too few fields. Bloblang doesn't report everything a value reads: a call with a non-literal argument reports only its arguments, and a lambda only its receiver. The option is now cache_by: [fields], and the cache is keyed on exactly those fields. What Bloblang does report is still checked, so reading a message field, the whole metadata, or an unlisted field is rejected. What it can't find is listed in the docs. The public XUnwrapper is gone; the expression is parsed through the manager's Bloblang environment, as group_by_value does.
• An optional interpolated column was always NULL (predates this branch). It now uses the column's definition level.
• Reused writers. The actual cause is that parquet-go 0.29.0's byte-array column buffer keeps its scratch buffer across Reset. On a reset writer, a page with a single empty value writes empty min/max, which a new writer leaves out. A file now gets a new writer only when a text column has exactly one non-null value and it's empty. As a precaution it also does when the file could span pages and contains an empty string; the tests couldn't show that second case is needed. A new differential test covers every encoding and codec, partitioned and not. It encodes the same batches with pooled writers and with a new writer for every file, and requires identical bytes and full reuse on clean batches. Without the check, 15 of its 36 configurations fail. Writers are reset to nil before going back to the pool, so they don't keep their last file reachable.
• Tests:
  • The count() test failed under -count=2, because Bloblang's counters are process-global; it now uses a new counter each run.
  • The escaped-field-name case replaced a name with itself; it now really escapes names.
  • The parity harness and the benchmark enable the cache through one helper that finds the interpolated column, instead of assuming it's the first.
• Memory:
  • Arenas start at one row, so tiny partitions no longer hold several times their rows.
  • A rejected row no longer creates and discards a group and arena chunks.
  • Files are written into one buffer per batch and copied out at their size, instead of each keeping what doubling left unused.
  • Strings the processor built itself are no longer copied.
• Speed: column names that need no unescaping are looked up as they appear, without a scan. Parsing a message takes about 14% less time.

Results

10 MiB batch, median over repeated runs on a 16-vCPU AMD EPYC. The PR's pipeline caches the interpolated column, as the docs example does.

	base branch	this PR
26 files: time / allocated / allocations / peak heap	426 ms / 104 MB / 2.30M / 75 MiB	312 ms / 69 MB / 0.76M / 59 MiB
710 files	661 ms / 202 MB / 2.61M / 101 MiB	436 ms / 93 MB / 0.91M / 80 MiB
one file	—	319 ms / 95–102 MB / 0.72M / 81–89 MiB
one file per message (35,966 files)	—	5.3 s / 1,177 MB / 8.49M / 223 MiB (1,484 MB / 277 MiB before the review fixes)

The legacy pipeline on the same batches takes about 2 s, allocates 485–605 MB and peaks at 184–231 MiB. A batch of 10,000 rejected rows allocates 6.9 MB (10.3 MB before the review fixes).

Things to know

• cache_by is a contract. The value must depend on nothing but the listed fields. Functions that read the message without naming a field (content()) or that change between calls (now(), count()) can't be detected.
• Evaluation order. A row's partition path is built before its values are coerced. When both would reject a row, only the logged error changes.
• Upstream fix. Fixing parquet-go's Reset upstream would remove the need for new writers entirely. That isn't filed yet.

Test plan

• go test -count=2 ./internal/impl/parquet/, and go test ./internal/bloblang/... ./public/service/ ./internal/impl/kafka/
• Mutation checks, each caught by the tests:
  • parsing integer strings in base 10;
  • skipping string unescaping;
  • an absent field taking the previous message's value;
  • reusing a writer regardless of empty strings;
  • dropping the one-value rule;
  • looking up names without unescaping them;
  • the long-string copy.
• Fuzz after every commit (30–180 s), plus 5 minutes at the end: no divergence
• golangci-lint v2.13.2 on all touched packages: 0 issues
• Docs regenerated with bento_docs_gen

🤖 Generated with Claude Code