From d2acbfeea35bd948909ca855875847de10ee8a2d Mon Sep 17 00:00:00 2001 From: John Kerl Date: Fri, 19 Jun 2026 12:27:22 -0400 Subject: [PATCH] Batch-allocate per-record objects; reuse CSV writer field buffer After batch-arena field allocation, profiling cat over 1M-record CSV showed the remaining ~5M allocations were almost entirely per-record (one each): the Mlrmap struct, the RecordAndContext wrapper, the CSV writer's []string, and the go-csv parser's own buffers. Address the first three: - mlrval.RecordArena gains NewRecord(), vending the Mlrmap struct itself from a per-batch slab (respecting --no-hash-records). Rolled out to every line-based reader (CSV, CSV-lite, TSV, DKVP, NIDX, PPRINT, XTAB, DKVPX) in place of NewMlrmapAsRecord. - The CSV reader batch-allocates RecordAndContext wrappers from a per-batch slab instead of one heap object per record (comment/output-string entries still allocate individually, but they are rare). - RecordWriterCSV reuses a single fieldsBuffer []string across records instead of allocating one per Write; WriteCSVRecordMaybeColorized consumes it synchronously and the writer is single-goroutine, so this is safe. Effect (big.*, 1M records, cat, best of 5): csv 0.26 -> 0.22 dkvp 0.51 -> 0.45 (Mlrmap slab) For CSV, cat's allocation-object count drops ~5.0M -> ~2.1M. The remaining ~2M are the go-csv parser's per-record backing string and field slice, which are intrinsic to parsing and would require a zero-copy/batch-slab parser rework. A CPU profile of cat now shows it is I/O-bound (syscall ~56%, bufio read+flush), with allocation/GC down to ~10% -- i.e. further allocation trimming no longer moves cat's wall-clock. GOGC=off confirms (no change). Verified: go test ./pkg/... and full regression suite pass; output is byte-identical across all formats including record-retaining verbs (tac), hashed and --no-hash-records. Co-Authored-By: Claude Opus 4.8 --- pkg/input/record_reader_csv.go | 14 ++++++++++++-- pkg/input/record_reader_csvlite.go | 4 ++-- pkg/input/record_reader_dkvp_nidx.go | 4 ++-- pkg/input/record_reader_dkvpx.go | 2 +- pkg/input/record_reader_pprint.go | 6 +++--- pkg/input/record_reader_tsv.go | 4 ++-- pkg/input/record_reader_xtab.go | 2 +- pkg/mlrval/record_arena.go | 27 +++++++++++++++++++++++++++ pkg/output/record_writer_csv.go | 11 ++++++++++- 9 files changed, 60 insertions(+), 14 deletions(-) diff --git a/pkg/input/record_reader_csv.go b/pkg/input/record_reader_csv.go index 94a6f6ae3..b6d9476fc 100644 --- a/pkg/input/record_reader_csv.go +++ b/pkg/input/record_reader_csv.go @@ -214,6 +214,12 @@ func (reader *RecordReaderCSV) getRecordBatch( } arena := mlrval.NewRecordArena(nfields) + // Batch-allocate the RecordAndContext wrappers too: one slab for the whole + // batch instead of one heap object per record. Comment/output-string entries + // (rare) still allocate individually via maybeConsumeComment. + racSlab := make([]types.RecordAndContext, len(csvRecords)) + racIndex := 0 + for _, csvRecord := range csvRecords { if reader.needHeader { @@ -242,7 +248,7 @@ func (reader *RecordReaderCSV) getRecordBatch( } } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() nh := int64(len(reader.header)) nd := int64(len(csvRecord)) @@ -281,7 +287,11 @@ func (reader *RecordReaderCSV) getRecordBatch( context.UpdateForInputRecord() - recordsAndContexts = append(recordsAndContexts, types.NewRecordAndContext(record, context)) + rac := &racSlab[racIndex] + racIndex++ + rac.Record = record + rac.Context = *context + recordsAndContexts = append(recordsAndContexts, rac) } return recordsAndContexts, false diff --git a/pkg/input/record_reader_csvlite.go b/pkg/input/record_reader_csvlite.go index ffcb8ce90..040f87d39 100644 --- a/pkg/input/record_reader_csvlite.go +++ b/pkg/input/record_reader_csvlite.go @@ -224,7 +224,7 @@ func getRecordBatchExplicitCSVHeader( return } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { if reader.useVoidRep && field == reader.voidRep { @@ -335,7 +335,7 @@ func getRecordBatchImplicitCSVHeader( } } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { if reader.useVoidRep && field == reader.voidRep { diff --git a/pkg/input/record_reader_dkvp_nidx.go b/pkg/input/record_reader_dkvp_nidx.go index 670caf51d..daf32d36c 100644 --- a/pkg/input/record_reader_dkvp_nidx.go +++ b/pkg/input/record_reader_dkvp_nidx.go @@ -170,7 +170,7 @@ func (reader *RecordReaderDKVPNIDX) getRecordBatch( } func recordFromDKVPLine(reader *RecordReaderDKVPNIDX, line string) (*mlrval.Mlrmap, error) { - record := mlrval.NewMlrmapAsRecord() + record := reader.recordArena.NewRecord() dedupeFieldNames := reader.readerOptions.DedupeFieldNames pairs := reader.fieldSplitter.Split(line) @@ -212,7 +212,7 @@ func recordFromDKVPLine(reader *RecordReaderDKVPNIDX, line string) (*mlrval.Mlrm } func recordFromNIDXLine(reader *RecordReaderDKVPNIDX, line string) (*mlrval.Mlrmap, error) { - record := mlrval.NewMlrmapAsRecord() + record := reader.recordArena.NewRecord() values := reader.fieldSplitter.Split(line) diff --git a/pkg/input/record_reader_dkvpx.go b/pkg/input/record_reader_dkvpx.go index a781814d3..776d2d967 100644 --- a/pkg/input/record_reader_dkvpx.go +++ b/pkg/input/record_reader_dkvpx.go @@ -171,7 +171,7 @@ func (reader *RecordReaderDKVPX) getRecordBatch( arena := mlrval.NewRecordArena(nfields) for _, omap := range dkvpxRecords { - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() for pe := omap.Head; pe != nil; pe = pe.Next { arena.PutDeferred(record, pe.Key, pe.Value, dedupeFieldNames) diff --git a/pkg/input/record_reader_pprint.go b/pkg/input/record_reader_pprint.go index 01de3cb2e..982432738 100644 --- a/pkg/input/record_reader_pprint.go +++ b/pkg/input/record_reader_pprint.go @@ -239,7 +239,7 @@ func (reader *RecordReaderPprintFixedSplit) getRecords( } } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { arena.PutDeferred(record, reader.headerStrings[i], field, dedupeFieldNames) @@ -472,7 +472,7 @@ func getRecordBatchExplicitPprintHeader( return } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { arena.PutDeferred(record, reader.headerStrings[i], field, dedupeFieldNames) @@ -596,7 +596,7 @@ func getRecordBatchImplicitPprintHeader( } } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { arena.PutDeferred(record, reader.headerStrings[i], field, dedupeFieldNames) diff --git a/pkg/input/record_reader_tsv.go b/pkg/input/record_reader_tsv.go index 85b455ce4..db57b8060 100644 --- a/pkg/input/record_reader_tsv.go +++ b/pkg/input/record_reader_tsv.go @@ -193,7 +193,7 @@ func getRecordBatchExplicitTSVHeader( return } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { field = lib.TSVDecodeField(field) @@ -300,7 +300,7 @@ func getRecordBatchImplicitTSVHeader( } } - record := mlrval.NewMlrmapAsRecord() + record := arena.NewRecord() if !reader.readerOptions.AllowRaggedCSVInput { for i, field := range fields { field = lib.TSVDecodeField(field) diff --git a/pkg/input/record_reader_xtab.go b/pkg/input/record_reader_xtab.go index 4b71c00f1..4a2cd950c 100644 --- a/pkg/input/record_reader_xtab.go +++ b/pkg/input/record_reader_xtab.go @@ -275,7 +275,7 @@ func (reader *RecordReaderXTAB) getRecordBatch( func (reader *RecordReaderXTAB) recordFromXTABLines( stanza []string, ) (*mlrval.Mlrmap, error) { - record := mlrval.NewMlrmapAsRecord() + record := reader.recordArena.NewRecord() dedupeFieldNames := reader.readerOptions.DedupeFieldNames for _, line := range stanza { diff --git a/pkg/mlrval/record_arena.go b/pkg/mlrval/record_arena.go index 6a327c229..f32be0092 100644 --- a/pkg/mlrval/record_arena.go +++ b/pkg/mlrval/record_arena.go @@ -28,8 +28,17 @@ type RecordArena struct { ei int vi int chunk int + + // records is a slab of Mlrmap structs vended by NewRecord, so the record + // container itself is batch-allocated alongside its fields. + records []Mlrmap + ri int } +// arenaChunkRecords is the slab size (in records) for NewRecord. It tracks the +// nominal records-per-batch so a batch typically draws from one slab. +const arenaChunkRecords = 512 + // NewRecordArena returns an arena whose first slabs are sized to nfieldsHint // (clamped to a sane range). Subsequent slabs, if needed, use arenaChunkFields. func NewRecordArena(nfieldsHint int) *RecordArena { @@ -43,6 +52,24 @@ func NewRecordArena(nfieldsHint int) *RecordArena { return &RecordArena{chunk: chunk} } +// NewRecord vends a fresh empty record (lazily-hashed, like NewMlrmapAsRecord) +// from the arena's record slab, batch-allocating the Mlrmap struct itself. It +// respects the global hashRecords setting: with --no-hash-records it returns an +// unhashed map, matching NewMlrmapAsRecord. +func (a *RecordArena) NewRecord() *Mlrmap { + if !hashRecords { + return newMlrmapUnhashed() + } + if a.ri >= len(a.records) { + a.records = make([]Mlrmap, arenaChunkRecords) + a.ri = 0 + } + m := &a.records[a.ri] + a.ri++ + m.autoHash = true + return m +} + // PutDeferred appends a field to mlrmap, drawing the entry and its deferred-type // value from the arena slabs. It mirrors PutReferenceMaybeDedupe's semantics for // duplicate keys. The value is built from the raw input string with type diff --git a/pkg/output/record_writer_csv.go b/pkg/output/record_writer_csv.go index c0f424cf5..1505f8403 100644 --- a/pkg/output/record_writer_csv.go +++ b/pkg/output/record_writer_csv.go @@ -19,6 +19,12 @@ type RecordWriterCSV struct { firstRecordKeys []string firstRecordNF int64 quoteAll bool // For double-quote around all fields + // fieldsBuffer is a reusable scratch slice for the stringified field values + // of the record currently being written. WriteCSVRecordMaybeColorized + // consumes it synchronously and does not retain it, and the writer processes + // one record at a time, so a single buffer can be shared across records to + // avoid a per-record allocation. + fieldsBuffer []string } func NewRecordWriterCSV(writerOptions *cli.TWriterOptions) (*RecordWriterCSV, error) { @@ -76,7 +82,10 @@ func (writer *RecordWriterCSV) Write( outputNF = writer.firstRecordNF } - fields := make([]string, outputNF) + if int64(cap(writer.fieldsBuffer)) < outputNF { + writer.fieldsBuffer = make([]string, outputNF) + } + fields := writer.fieldsBuffer[:outputNF] var i int64 = 0 for pe := outrec.Head; pe != nil; pe = pe.Next { if i < writer.firstRecordNF && pe.Key != writer.firstRecordKeys[i] {