parquet: json_parquet_encode; kafka_franz: copy_record_values

[almostintuitive]

Summary

Two changes that let a JSON→Parquet archiver hold its batches cheaply:

• json_parquet_encode (new batch processor, internal/impl/parquet/processor_json_encode.go). It encodes flat JSON objects straight into Parquet, row by row, and writes the same bytes as the pipeline it replaces:

  • a per-column Bloblang mapping (.not_null().string(), .int32(), .int64(), .number() / .or(NaN).number());
  • a catch that drops rejected rows;
  • group_by_value on a partition path;
  • parquet_encode.

  How it guarantees that:

  • the schema and writer are parquet_encode's, built by the same code (GenerateStructType, parquet.SchemaOf, GenericWriter);
  • each column is coerced by the same internal/value helpers the Bloblang methods call (IToString, IToInt32, IToInt, IToNumber);
  • coercion runs on the Go values Bento's own JSON decoding produces (json.Number, last duplicate key wins, U+FFFD for invalid UTF-8, exactly one document).

  Rows it rejects are dropped, logged, and counted in json_parquet_encode_dropped. An optional partition.path ({column} and {column|go-time-layout} placeholders) splits the batch into one file per path.

• kafka_franz.copy_record_values (advanced, default false). Copies each record's value into its message, and drops the record's key, value and headers once metadata is extracted. Without it, a message held for a long time (for example in a large output batch) keeps its whole fetch response in memory, along with every other record fetched from that broker in the same request.

Parity tests

processor_json_encode_parity_test.go and processor_json_encode_parity_cases_test.go run the replaced pipeline and json_parquet_encode as real in-process streams over the same batches. They require byte-identical Parquet files, the same partitions, the same metadata, and the same rows dropped.

• Every column type × optionality against 68 values: every number literal form and range edge, numeric strings (hex, octal, underscores, NaN, inf), unicode, invalid UTF-8, escapes, bools, nulls, nested values, and missing fields. Each column is asserted to have both kept and dropped values, so a case can't pass by both sides writing nothing.
• Malformed messages: empty, whitespace, truncated, arrays and scalars, two documents, trailing garbage, BOM, duplicate keys, deep nesting, 1 MiB strings.
• Metadata: missing topic, short topic, empty segment, unicode.
• Partitions: around midnight, before the epoch, far future, fractional, string and overflowing times; path fields of every JSON kind.
• Random batches: 300 seeded batches (60,313 messages, 3,137 dropped), plus a 10 MiB batch (40,949 messages across 747 files).
• Fuzz: FuzzJSONParquetParity ran 15 minutes locally, 8.1M executions and 1,151 interesting inputs, with no divergence.
• Mutation check: 13 hand-made mutants of the processor. 12 were caught; the 13th showed a redundant check, which was removed.

The one intentional divergence: an optional UTF8 column reads null or absent as NULL. The mapping it's modelled on would have rejected that row. This is covered by TestJSONParquetEncodeOptionalStringIsNull.

Benchmarks

AMD EPYC (16 vCPU), 10 MiB batch of synthetic rows across 710 partitions (processor_json_encode_bench_test.go):

	legacy pipeline	json_parquet_encode
encode time	2.30 s	0.66 s (3.5x faster)
allocations	12.4M	2.6M
bytes allocated	603 MB	201 MB
held while batching (per byte of JSON)	5.15 B	2.93 B
peak heap during encode	230 MiB	101 MiB

Rows are collected per partition, and each partition's file is encoded and released before the next one starts, the same way parquet_encode handles each group from group_by_value. An earlier revision kept one writer per partition open until the whole batch was read. On this batch that meant 710 writers with their column buffers, and a 300 MiB peak, more than the legacy path. Each file gets a new writer, because a writer reused through Reset doesn't write the same bytes; the parity tests catch that.

Test plan

• go test ./internal/impl/parquet/ ./internal/impl/kafka/
• go vet on both packages
• fuzz, 15 minutes
• make docs: component docs regenerated. The lint step fails locally on the DuckDB/sql_raw docs, which need a CGO build; unrelated to this change.
• golangci-lint v2.13.2 (built with Go 1.27.1) on both packages: 0 issues
• fuzz, 90 seconds after the one-file-at-a-time change: 2.1M executions, no divergence
• CI: the new Aperiodic Test workflow (.github/workflows/aperiodic_test.yml) runs vet, tests and golangci-lint on ubicloud-standard-2. It covers only the Go packages the PR touches (the nearest directory with Go files to each changed file) and checks that go.mod/go.sum are tidy when they change. Upstream's Test workflow is still disabled on this fork.

🤖 Generated with Claude Code