From be84da852788f77cf2e25b15a95d1e283d253d3b Mon Sep 17 00:00:00 2001 From: Brian Bockelman Date: Sat, 22 Aug 2026 18:31:10 -0500 Subject: [PATCH 1/2] collections: drop per-segment byOff map (binary search) + alias the arena-offset map Two more per-segment columnar-metadata copies the heap profile flagged, no on-disk format change. byOff: colNative kept a resident map[arenaOffset]index -- one entry per record per columnarized segment, ~164MB of live heap on a production archive -- for an O(1) lookup. The offset map is already in hand and strictly ascending (records append in order), so an O(log n) binary search (indexOf) replaces it; the map is removed. offs: colSegment.offs []uint32 was materialized from the section on every open. It is packed little-endian u32 on disk, so it becomes offsB []byte that ALIASES the mmap on a reopened segment (zero heap; a built one holds a compact buffer), read via offAt/offsLen -- the same aliasing #219 applied to the block offset arrays. Marshal output is byte-identical; readU32s is gone, packU32s added for the build path. ~196MB off the resident plateau, and byOff no longer scales with record count. TestColSegOffsAliasAndIndexOf covers the aliased read-back and indexOf (present -> index; below/between/above -> not-found). Full collections suite green (both stores). Co-Authored-By: Claude Opus 4.8 --- collections/colagg.go | 4 ++-- collections/colblock_layout_test.go | 29 +++++++++++++++++++++++++- collections/colblock_segment_test.go | 2 +- collections/colescclass_test.go | 2 +- collections/colgroup_bench_test.go | 2 +- collections/colgroup_test.go | 16 +++++++------- collections/colgroupblock_test.go | 4 ++-- collections/colgroupcount.go | 4 ++-- collections/colgroupread_test.go | 2 +- collections/colmulti.go | 4 ++-- collections/colnative.go | 29 ++++++++++++++++++-------- collections/colnative_build.go | 4 ++-- collections/colpersist.go | 30 +++++++++++++-------------- collections/colpersist_test.go | 4 ++-- collections/colpresence.go | 8 +++---- collections/colscan.go | 17 +++++++++++++-- collections/colscope.go | 4 ++-- collections/colstatsmulti.go | 8 +++---- collections/colvec.go | 4 ++-- collections/colvecgroup.go | 8 +++---- collections/regioncodec_bench_test.go | 2 +- collections/store.go | 4 ++-- collections/vecphase_test.go | 4 ++-- 23 files changed, 122 insertions(+), 73 deletions(-) diff --git a/collections/colagg.go b/collections/colagg.go index 5373c7a6..6d251a29 100644 --- a/collections/colagg.go +++ b/collections/colagg.go @@ -204,10 +204,10 @@ func (c *Collection) scanNumValues(fieldID uint32, bc *blockCache, fn func(colVa for _, blk := range cs.blocks { blk.scanInt(bidx, bc, func(k int, present bool, v int64) { gk := base + k - if gk >= len(cs.offs) { + if gk >= cs.offsLen() { return // truncated offs: cannot establish visibility, so do not count it } - o := cs.offs[gk] + o := cs.offAt(gk) if !(recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0) { return // not visible at this snapshot } diff --git a/collections/colblock_layout_test.go b/collections/colblock_layout_test.go index 0fd51e07..691dfbc3 100644 --- a/collections/colblock_layout_test.go +++ b/collections/colblock_layout_test.go @@ -5,6 +5,33 @@ import ( "testing" ) +// TestColSegOffsAliasAndIndexOf checks the arena-offset map reads back correctly from its packed-u32 +// aliased form, and that indexOf (the binary search replacing the per-segment byOff map) maps an +// arena offset to its record index -- returning not-found for offsets no record holds. +func TestColSegOffsAliasAndIndexOf(t *testing.T) { + offs := []uint32{5, 12, 30, 31, 100} // strictly ascending, as a real segment's are + cs := &colSegment{offsB: packU32s(offs)} + if cs.offsLen() != len(offs) { + t.Fatalf("offsLen=%d, want %d", cs.offsLen(), len(offs)) + } + for i, o := range offs { + if cs.offAt(i) != o { + t.Errorf("offAt(%d)=%d, want %d", i, cs.offAt(i), o) + } + } + cn := &colNative{seg: cs} + for i, o := range offs { + if k, ok := cn.indexOf(o); !ok || k != i { + t.Errorf("indexOf(%d)=(%d,%v), want (%d,true)", o, k, ok, i) + } + } + for _, absent := range []uint32{0, 6, 29, 50, 101} { // below, between, and above real offsets + if k, ok := cn.indexOf(absent); ok { + t.Errorf("indexOf(%d) found index %d, want not-found", absent, k) + } + } +} + // TestColBlockLayoutSharedAcrossBlocks is the regression for the per-block metadata heap: every base // block of a segment must reference ONE shared *colLayout (not a per-block copy of the hot/cold // partition + column offsets), and the per-record offsets must round-trip through the packed-u32 @@ -31,7 +58,7 @@ func TestColBlockLayoutSharedAcrossBlocks(t *testing.T) { for i := range offs { offs[i] = uint32(i * 7) } - orig := &colSegment{blocks: []*columnarBlock{b1, b2}, offs: offs} + orig := &colSegment{blocks: []*columnarBlock{b1, b2}, offsB: packU32s(offs)} got := unmarshalColSegment(marshalColSegment(orig, c.intern.Name), identityCodec{}, c.intern.Intern) if got == nil || len(got.blocks) != 2 { diff --git a/collections/colblock_segment_test.go b/collections/colblock_segment_test.go index 69fb2941..1fd1bd97 100644 --- a/collections/colblock_segment_test.go +++ b/collections/colblock_segment_test.go @@ -34,7 +34,7 @@ func segmentWires(t *testing.T, seg *segment) [][]byte { // oneBlockColSeg wraps a single block as a segment accelerator, for tests that build a block // directly rather than from a segment. func oneBlockColSeg(blk *columnarBlock, offs []uint32) *colSegment { - return &colSegment{blocks: []*columnarBlock{blk}, offs: offs} + return &colSegment{blocks: []*columnarBlock{blk}, offsB: packU32s(offs)} } // TestBuildColumnarFromSegment transcodes a real segment's records into columnar ROW-GROUP blocks diff --git a/collections/colescclass_test.go b/collections/colescclass_test.go index 958fb5e9..56b9b72a 100644 --- a/collections/colescclass_test.go +++ b/collections/colescclass_test.go @@ -139,7 +139,7 @@ func TestEscapeClassSurvivesPersistence(t *testing.T) { rows = append(rows, s.encode(wire.Ad(iw))) } blk := encodeColumnarBlock(s, rows, resolveColLayout(s, nil), identityCodec{}, nil) - cs := &colSegment{blocks: []*columnarBlock{blk}, offs: make([]uint32, n)} + cs := &colSegment{blocks: []*columnarBlock{blk}, offsB: make([]byte, n*4)} blob := marshalColSegment(cs, func(id uint32) (string, bool) { return c.intern.Name(id) }) got := unmarshalColSegment(blob, identityCodec{}, func(name string) uint32 { return c.intern.Intern(name) }) if got == nil { diff --git a/collections/colgroup_bench_test.go b/collections/colgroup_bench_test.go index 7023999c..7c79c4d6 100644 --- a/collections/colgroup_bench_test.go +++ b/collections/colgroup_bench_test.go @@ -85,7 +85,7 @@ func regroup(tb testing.TB, c *Collection, g colGrouping) (blocks, recs int) { d := seg.dict.Load() bl, offs := buildColumnarFromSegment(seg.data, seg.used, seg.codec, c.regionCodec(), st.schema, st.hot, g, func(dst, w []byte) ([]byte, bool) { return c.recordToInternedDict(d, dst, w) }) - seg.colblk.Store(&colSegment{blocks: bl, offs: offs}) + seg.colblk.Store(&colSegment{blocks: bl, offsB: packU32s(offs)}) blocks += len(bl) recs += len(offs) } diff --git a/collections/colgroup_test.go b/collections/colgroup_test.go index d669c43a..71e76cce 100644 --- a/collections/colgroup_test.go +++ b/collections/colgroup_test.go @@ -144,8 +144,8 @@ func TestRowGroupsBoundedByBytes(t *testing.T) { } total += b.n } - if total != len(cs.offs) { - t.Errorf("blocks cover %d records but offs has %d", total, len(cs.offs)) + if total != cs.offsLen() { + t.Errorf("blocks cover %d records but offs has %d", total, cs.offsLen()) } } } @@ -198,8 +198,8 @@ func TestRowGroupsBoundedByRows(t *testing.T) { multi++ } // Every record in the segment is covered exactly once by the groups. - if total != len(cs.offs) { - t.Errorf("blocks cover %d records but offs has %d", total, len(cs.offs)) + if total != cs.offsLen() { + t.Errorf("blocks cover %d records but offs has %d", total, cs.offsLen()) } } } @@ -295,8 +295,8 @@ func TestColSegmentPersistMultiGroup(t *testing.T) { if len(got.blocks) != len(cs.blocks) { t.Fatalf("reloaded %d row groups, want %d", len(got.blocks), len(cs.blocks)) } - if len(got.offs) != len(cs.offs) { - t.Fatalf("reloaded %d offs, want %d", len(got.offs), len(cs.offs)) + if got.offsLen() != cs.offsLen() { + t.Fatalf("reloaded %d offs, want %d", got.offsLen(), cs.offsLen()) } for i := range cs.blocks { a, b := cs.blocks[i], got.blocks[i] @@ -367,12 +367,12 @@ func TestColSegmentRejectsGroupOffsMismatch(t *testing.T) { t.Fatal("fixture produced a single row group; the mismatch guard would not be exercised") } // A truthful blob reloads. - if unmarshalColSegment(marshalColSegment(&colSegment{blocks: blocks, offs: offs}, store.intern.Name), + if unmarshalColSegment(marshalColSegment(&colSegment{blocks: blocks, offsB: packU32s(offs)}, store.intern.Name), identityCodec{}, store.intern.Intern) == nil { t.Fatal("a well-formed multi-group blob failed to reload") } // Drop one record from offs: the counts no longer sum, and the reload must refuse. - short := &colSegment{blocks: blocks, offs: offs[:len(offs)-1]} + short := &colSegment{blocks: blocks, offsB: packU32s(offs[:len(offs)-1])} if got := unmarshalColSegment(marshalColSegment(short, store.intern.Name), identityCodec{}, store.intern.Intern); got != nil { t.Error("a blob whose block counts do not sum to its offs length was accepted; a scan " + diff --git a/collections/colgroupblock_test.go b/collections/colgroupblock_test.go index 00d8d6fa..b29e011c 100644 --- a/collections/colgroupblock_test.go +++ b/collections/colgroupblock_test.go @@ -271,7 +271,7 @@ func TestGroupBlocksRoundTripThroughPersistence(t *testing.T) { } blk := encodeColumnarBlock(base, rows, resolveColLayout(base, nil), identityCodec{}, nil) g.blocks = buildGroupBlocks([]*colGroup{g}, []*colLayout{resolveColLayout(g.schema, nil)}, iws, identityCodec{}, nil) - cs := &colSegment{blocks: []*columnarBlock{blk}, offs: make([]uint32, n), groups: []*colGroup{g}} + cs := &colSegment{blocks: []*columnarBlock{blk}, offsB: make([]byte, n*4), groups: []*colGroup{g}} blob := marshalColSegment(cs, func(id uint32) (string, bool) { return c.intern.Name(id) }) if blob == nil { @@ -342,7 +342,7 @@ func TestGroupSectionRejectsInconsistentSelection(t *testing.T) { } blk := encodeColumnarBlock(base, rows, resolveColLayout(base, nil), identityCodec{}, nil) g.blocks = buildGroupBlocks([]*colGroup{g}, []*colLayout{resolveColLayout(g.schema, nil)}, iws, identityCodec{}, nil) - cs := &colSegment{blocks: []*columnarBlock{blk}, offs: make([]uint32, n), groups: []*colGroup{g}} + cs := &colSegment{blocks: []*columnarBlock{blk}, offsB: make([]byte, n*4), groups: []*colGroup{g}} nameOf := func(id uint32) (string, bool) { return c.intern.Name(id) } internName := func(s string) uint32 { return c.intern.Intern(s) } diff --git a/collections/colgroupcount.go b/collections/colgroupcount.go index e47c0fc4..bf0d0190 100644 --- a/collections/colgroupcount.go +++ b/collections/colgroupcount.go @@ -349,9 +349,9 @@ func (c *Collection) schemaScanGroupStats(groupID uint32, aggIDs []uint32, preds live := 0 for k := 0; k < blk.n; k++ { gk := base + k - vis := gk < len(cs.offs) + vis := gk < cs.offsLen() if vis { - o := cs.offs[gk] + o := cs.offAt(gk) vis = recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 } keep[k] = vis diff --git a/collections/colgroupread_test.go b/collections/colgroupread_test.go index 11de2dea..8fe84f06 100644 --- a/collections/colgroupread_test.go +++ b/collections/colgroupread_test.go @@ -26,7 +26,7 @@ func groupSchemaReadFixture(t *testing.T, c *Collection, n int) (*colScope, *col g.ids = append(g.ids, f.id) } g.blocks = buildGroupBlocks([]*colGroup{g}, []*colLayout{resolveColLayout(g.schema, nil)}, iws, c.regionCodec(), nil) - seg := &colSegment{blocks: []*columnarBlock{blk}, offs: make([]uint32, n), groups: []*colGroup{g}} + seg := &colSegment{blocks: []*columnarBlock{blk}, offsB: make([]byte, n*4), groups: []*colGroup{g}} bc, err := newBlockCache(1 << 20) if err != nil { diff --git a/collections/colmulti.go b/collections/colmulti.go index c2f8360c..e55b6192 100644 --- a/collections/colmulti.go +++ b/collections/colmulti.go @@ -148,9 +148,9 @@ func (c *Collection) schemaScanCountMulti(preds []fieldPred, bc *blockCache) int live := 0 for k := 0; k < blk.n; k++ { gk := base + k - vis := gk < len(cs.offs) + vis := gk < cs.offsLen() if vis { - o := cs.offs[gk] + o := cs.offAt(gk) vis = recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 } keep[k] = vis diff --git a/collections/colnative.go b/collections/colnative.go index c61a548d..bf25593c 100644 --- a/collections/colnative.go +++ b/collections/colnative.go @@ -2,12 +2,26 @@ package collections import ( "errors" + "sort" "sync" "sync/atomic" "github.com/PelicanPlatform/classad/collections/wire" ) +// indexOf returns the record index for arena offset off, and whether the segment holds a +// columnarized record there. The segment's offset map is strictly ascending (records are appended +// in order), so this is a binary search -- replacing a resident arena-offset -> index map that cost +// per-segment heap proportional to the record count. +func (cn *colNative) indexOf(off uint32) (int, bool) { + n := cn.seg.offsLen() + i := sort.Search(n, func(i int) bool { return cn.seg.offAt(i) >= off }) + if i < n && cn.seg.offAt(i) == off { + return i, true + } + return 0, false +} + // COLUMNAR-NATIVE SEALED SEGMENTS. // // A sealed segment normally stores each record's whole ad, compressed on its own, and the columnar @@ -43,10 +57,10 @@ import ( // segment dictionary is. type colNative struct { seg *colSegment // schema + blocks + per-record arena offsets - // byOff maps an arena record offset to its index among the segment's data records, so a - // reader holding an offset (from the key index, say) can find its columns. Built once when - // the payload is published; the alternative is a binary search over seg.offs on every read. - byOff map[uint32]int + // A reader holding an arena record offset finds its record index by BINARY SEARCH over the + // segment's ascending offset map (indexOf), not a resident map: an arena-offset -> index map + // was ~5 bytes per record of live heap for every columnarized segment (hundreds of MB on a + // large archive), for an O(1) lookup that O(log n) over the offsets already in hand replaces. // cache decompresses the columnar regions once per block rather than per record read. cache *blockCache // dict is the segment's attribute dictionary when its records are interned, for translating @@ -110,10 +124,7 @@ func publishColNative(c *Collection, seg *segment) { // per segment made ristretto's fixed admission metadata scale with segment count (a // multi-GB reopen leak on a large archive). A nil cache (creation failed) is valid -- // blockCache methods then decompress every time -- so it never damages the segment. - cn := &colNative{seg: cs, byOff: make(map[uint32]int, len(cs.offs)), cache: c.sharedColCache()} - for i, ro := range cs.offs { - cn.byOff[ro] = i - } + cn := &colNative{seg: cs, cache: c.sharedColCache()} cn.dict = seg.dict.Load() // The payload's cold tails are keyed by the SEGMENT's dictionary, so readers translate // through that dictionary rather than trusting this process's intern numbering. @@ -208,7 +219,7 @@ func (c *Collection) recordWireIn(seg *segment, data []byte, off uint32, buf []b } return raw, nil } - k, ok := cn.byOff[off] + k, ok := cn.indexOf(off) if !ok { return raw, nil // not a columnarized record (a marker, or written after the transform) } diff --git a/collections/colnative_build.go b/collections/colnative_build.go index f5f8b162..1880920e 100644 --- a/collections/colnative_build.go +++ b/collections/colnative_build.go @@ -104,7 +104,7 @@ func (c *Collection) columnarizeSegment(sh *shard, src *segment, s *adSchema, ho // would remove one record's attributes on the strength of another's membership. return nil, nil, nil } - cs := &colSegment{blocks: blocks, offs: offs, dictKeyed: d != nil} + cs := &colSegment{blocks: blocks, offsB: packU32s(offs), dictKeyed: d != nil} // Re-key the pinned groups onto this segment's selections: the schema and members are shared, the // per-block bitmaps are not. for gi, g := range groups { @@ -197,7 +197,7 @@ func (c *Collection) columnarizeSegment(sh *shard, src *segment, s *adSchema, ho } // The offsets moved, so the columnar payload must describe the NEW ones or a reader would map // a record to another record's columns. - cs.offs = newOffs + cs.offsB = packU32s(newOffs) blob = marshalColSegment(cs, func(id uint32) (string, bool) { return c.intern.Name(id) }) if blob == nil || len(blob) == 0 { dst.retire() diff --git a/collections/colpersist.go b/collections/colpersist.go index 5c836f92..c7c908a5 100644 --- a/collections/colpersist.go +++ b/collections/colpersist.go @@ -244,10 +244,10 @@ func marshalColSegment(cs *colSegment, nameOf func(uint32) (string, bool)) []byt for _, b := range cs.blocks { dst = appendColBlock(dst, b) } - dst = appendU32(dst, uint32(len(cs.offs))) - for _, o := range cs.offs { - dst = appendU32(dst, o) - } + // offsB is already packed little-endian u32: write the count then the bytes verbatim, + // byte-identical to the old element-by-element encoding. + dst = appendU32(dst, uint32(len(cs.offsB)/4)) + dst = append(dst, cs.offsB...) // Group schemas and their selections, after the base blocks so a reader that has already // validated the base segment can reject a bad group section on its own. dst = appendU32(dst, uint32(len(cs.groups))) @@ -502,14 +502,14 @@ func unmarshalColSegment(data []byte, codec Codec, internName func(string) uint3 blocks = append(blocks, b) total += b.n } - offs := readU32s(c) + offsB := c.u32SliceBytes() if c.err != nil { return nil } // The blocks' record counts must sum to the offs length, or a scan would map a record to the // wrong arena offset and read another record's MVCC seq -- a wrong answer rather than a slow // one. Reject instead, and let the segment rebuild. - if total != len(offs) { + if total != len(offsB)/4 { return nil } if remap != nil { @@ -518,7 +518,7 @@ func unmarshalColSegment(data []byte, codec Codec, internName func(string) uint3 b.remap = remap } } - cs := &colSegment{blocks: blocks, offs: offs, dictKeyed: remap == nil} + cs := &colSegment{blocks: blocks, offsB: offsB, dictKeyed: remap == nil} ng := int(c.u32()) if c.err != nil || ng < 0 { return nil @@ -578,14 +578,12 @@ func unmarshalColSegment(data []byte, codec Codec, internName func(string) uint3 return cs } -func readU32s(c *cursor) []uint32 { - n := int(c.u32()) - if n < 0 || !c.need(0) { - return nil +// packU32s encodes a uint32 slice as packed little-endian bytes -- the form colSegment.offsB and a +// columnar block's offset arrays hold, so a reopen can alias the mmap instead of materializing ints. +func packU32s(vs []uint32) []byte { + b := make([]byte, 0, len(vs)*4) + for _, v := range vs { + b = binary.LittleEndian.AppendUint32(b, v) } - out := make([]uint32, n) - for i := range out { - out[i] = c.u32() - } - return out + return b } diff --git a/collections/colpersist_test.go b/collections/colpersist_test.go index abe8472f..b2abd89b 100644 --- a/collections/colpersist_test.go +++ b/collections/colpersist_test.go @@ -40,8 +40,8 @@ func TestColSegmentMarshalRoundTrip(t *testing.T) { codec.Name(), gb.n, blk.n, gb.bitsStride, blk.bitsStride, len(gb.hotFields()), len(blk.hotFields()), len(gb.coldFields()), len(blk.coldFields())) } for i := range offs { - if got.offs[i] != offs[i] { - t.Fatalf("%s: offs[%d]=%d, want %d", codec.Name(), i, got.offs[i], offs[i]) + if got.offAt(i) != offs[i] { + t.Fatalf("%s: offs[%d]=%d, want %d", codec.Name(), i, got.offAt(i), offs[i]) } } // Every record reconstructs byte-identically from the decoded block. diff --git a/collections/colpresence.go b/collections/colpresence.go index b2f3e6be..cc25d116 100644 --- a/collections/colpresence.go +++ b/collections/colpresence.go @@ -148,10 +148,10 @@ func (c *Collection) schemaScanPresenceCount(pred presencePred, bc *blockCache) if blk.fieldAbsentFromBlock(idx) { for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(cs.offs) { + if gk >= cs.offsLen() { break } - o := cs.offs[gk] + o := cs.offAt(gk) if recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 { tally(true) } @@ -161,10 +161,10 @@ func (c *Collection) schemaScanPresenceCount(pred presencePred, bc *blockCache) } for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(cs.offs) { + if gk >= cs.offsLen() { break } - o := cs.offs[gk] + o := cs.offAt(gk) if !(recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0) { continue // not visible at this snapshot } diff --git a/collections/colscan.go b/collections/colscan.go index 3bc9c2bf..8ce6c3f8 100644 --- a/collections/colscan.go +++ b/collections/colscan.go @@ -1,6 +1,7 @@ package collections import ( + "encoding/binary" "sort" "strings" "sync/atomic" @@ -33,12 +34,24 @@ type colSegment struct { // intern ids, which are renumbered at every Open -- so the section names them instead. dictKeyed bool blocks []*columnarBlock - offs []uint32 + // offsB is the per-record arena offset map (record index -> its byte offset in the segment), + // packed as little-endian uint32 -- the exact on-disk form. A reopened segment ALIASES the mmap + // here (zero heap); a freshly built one holds a compact heap buffer. Read via offAt/offsLen. It + // was []uint32 materialized on read; the aliased bytes remove that per-segment copy. The offsets + // are strictly ascending (records are appended in order), which indexOf relies on. + offsB []byte // groups are the group schemas' selections, one colGroupBlock per group per base block. Empty // when the collection carries no group schemas. groups []*colGroup } +// offsLen is the number of records the arena-offset map covers. +func (cs *colSegment) offsLen() int { return len(cs.offsB) / 4 } + +// offAt returns record i's arena byte offset, decoded from the packed u32 map (which aliases the +// mmap on a reopened segment). +func (cs *colSegment) offAt(i int) uint32 { return binary.LittleEndian.Uint32(cs.offsB[i*4:]) } + // schema returns the schema all of cs's blocks were built under, or nil if it carries none. func (cs *colSegment) schema() *adSchema { if cs == nil || len(cs.blocks) == 0 { @@ -354,7 +367,7 @@ func (c *Collection) buildColSegment(seg *segment, s *adSchema, hot []int) *colS if len(blocks) == 0 { return nil } - cs := &colSegment{blocks: blocks, offs: offs, dictKeyed: d != nil} + cs := &colSegment{blocks: blocks, offsB: packU32s(offs), dictKeyed: d != nil} // Re-key the pinned groups onto this segment's selections: the schema and members are shared, // the per-block bitmaps are not. for gi, g := range groups { diff --git a/collections/colscope.go b/collections/colscope.go index c9d2526e..7c636e68 100644 --- a/collections/colscope.go +++ b/collections/colscope.go @@ -541,10 +541,10 @@ func (c *Collection) countBlockScoped(cs *colScope, resolver func(name string, s count := 0 for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(seg.offs) { + if gk >= seg.offsLen() { break } - o := seg.offs[gk] + o := seg.offAt(gk) if !(recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0) { continue } diff --git a/collections/colstatsmulti.go b/collections/colstatsmulti.go index df32f553..8a1b7a03 100644 --- a/collections/colstatsmulti.go +++ b/collections/colstatsmulti.go @@ -204,10 +204,10 @@ func (c *Collection) schemaScanStatsMulti(aggID uint32, preds []fieldPred, bc *b // One fused pass: visibility, value, and the aggregated column's own predicate. for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(cs.offs) { + if gk >= cs.offsLen() { break } - o := cs.offs[gk] + o := cs.offAt(gk) if !(recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0) { continue } @@ -226,9 +226,9 @@ func (c *Collection) schemaScanStatsMulti(aggID uint32, preds []fieldPred, bc *b live := 0 for k := 0; k < blk.n; k++ { gk := base + k - vis := gk < len(cs.offs) + vis := gk < cs.offsLen() if vis { - o := cs.offs[gk] + o := cs.offAt(gk) vis = recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 } keep[k] = vis diff --git a/collections/colvec.go b/collections/colvec.go index 673a1e44..c56fe8f6 100644 --- a/collections/colvec.go +++ b/collections/colvec.go @@ -427,10 +427,10 @@ func (c *Collection) VectorEvalCount(q *vm.Query) (int, bool) { nLive := 0 for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(seg.offs) { + if gk >= seg.offsLen() { break } - o := seg.offs[gk] + o := seg.offAt(gk) if recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 { live[k/64] |= 1 << uint(k%64) nLive++ diff --git a/collections/colvecgroup.go b/collections/colvecgroup.go index c33bd310..03f0524f 100644 --- a/collections/colvecgroup.go +++ b/collections/colvecgroup.go @@ -114,10 +114,10 @@ func (c *Collection) vecGroupStats(q *vm.Query, groupID uint32, aggIDs []uint32, nLive := 0 for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(seg.offs) { + if gk >= seg.offsLen() { break } - o := seg.offs[gk] + o := seg.offAt(gk) if recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 { live[k/64] |= 1 << uint(k%64) nLive++ @@ -203,10 +203,10 @@ func (c *Collection) groupBlockScoped(cs *colScope, resolver func(name string, s col numCol, aggCols []numCol, bc *blockCache, acc map[groupKey]*groupAcc) bool { for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(seg.offs) { + if gk >= seg.offsLen() { break } - o := seg.offs[gk] + o := seg.offAt(gk) if !(recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0) { continue } diff --git a/collections/regioncodec_bench_test.go b/collections/regioncodec_bench_test.go index ee93f287..c8e9fd42 100644 --- a/collections/regioncodec_bench_test.go +++ b/collections/regioncodec_bench_test.go @@ -53,7 +53,7 @@ func BenchmarkRegionCodecAggregate(b *testing.B) { bl, offs := buildColumnarFromSegment(seg.data, seg.used, seg.codec, regionCodec, st.schema, st.hot, defaultColGrouping(), func(dst, w []byte) ([]byte, bool) { return c.recordToInternedDict(d, dst, w) }) - seg.colblk.Store(&colSegment{blocks: bl, offs: offs}) + seg.colblk.Store(&colSegment{blocks: bl, offsB: packU32s(offs)}) for _, blk := range bl { bytes += len(blk.coldNumComp) + len(blk.strComp) + len(blk.coldComp) } diff --git a/collections/store.go b/collections/store.go index 8071450c..b5244723 100644 --- a/collections/store.go +++ b/collections/store.go @@ -1128,7 +1128,7 @@ func (c *Collection) scanWindows(s0 uint64, wins []segWindow, qp queryPlan, emit cn := r.w.seg.colNative.Load() colDecided := false if pre != nil && cn != nil { - if k, ok := cn.byOff[r.off]; ok { + if k, ok := cn.indexOf(r.off); ok { matches, decided := pre.test(cn, k) if decided { qp.stats.columnDecided() @@ -1144,7 +1144,7 @@ func (c *Collection) scanWindows(s0 uint64, wins []segWindow, qp queryPlan, emit // just decided FROM the columns. Anything less certain reassembles, because a narrowed ad // cannot be handed to the matcher -- its seed set is not closed. if projCS != nil && cn != nil && (constQuery || colDecided) { - if k, ok := cn.byOff[r.off]; ok { + if k, ok := cn.indexOf(r.off); ok { if out, ok := c.projectFromColumns(cn, r, k, qp.proj, projCS, &projScratch); ok { qp.stats.matched() // projected straight from columns; no reassembly if qp.ws != nil { diff --git a/collections/vecphase_test.go b/collections/vecphase_test.go index 0d80d19b..73369ac3 100644 --- a/collections/vecphase_test.go +++ b/collections/vecphase_test.go @@ -54,10 +54,10 @@ func (c *Collection) vecScanPhase(q *vm.Query, phase int) int { nLive := 0 for k := 0; k < blk.n; k++ { gk := base + k - if gk >= len(seg.offs) { + if gk >= seg.offsLen() { break } - o := seg.offs[gk] + o := seg.offAt(gk) if recSeq(w.data, o) <= s0 && recSuperseded(w.data, o) > s0 { nLive++ } From e88e46847e3865e93c41a47dd888dae2ec550c5a Mon Sep 17 00:00:00 2001 From: Brian Bockelman Date: Sat, 22 Aug 2026 18:44:58 -0500 Subject: [PATCH 2/2] collections: block zones + strDict maps -> sorted slices (no format bump) The last per-block columnar-metadata maps. A block kept zones (map[int]blockZone, ~174MB in the profile) and strDict (map[int]strDictField, ~156MB) -- one map each per block over ~100K+ blocks, whose bucket overhead was ~2x the entries. A block-format bump to alias them would be data loss on columnar-native segments (their schema'd attributes live only in the block payload, so rejecting an old section drops the only copy) and would need dual-format read to be safe. But the map overhead is a representation choice, not on-disk: parse the SAME on-disk entries into a sorted-by-idx slice and binary-search them (zone(), strDictOf()). Byte- identical marshal, so every existing segment gets the smaller form immediately with no migration. seg.zones (the per-segment zone map keyed by attr id) is unrelated and untouched. TestColBlockLayoutSharedAcrossBlocks extended: both slices sorted, every entry findable via its accessor, fixture non-vacuous. Full collections suite green. Co-Authored-By: Claude Opus 4.8 --- collections/colblock.go | 58 +++++++++++++++++++++++++---- collections/colblock_layout_test.go | 28 ++++++++++++++ collections/colpersist.go | 43 +++++++++++---------- collections/colvec.go | 2 +- collections/strdict.go | 23 ++++++++++-- collections/strdict_test.go | 4 +- 6 files changed, 121 insertions(+), 37 deletions(-) diff --git a/collections/colblock.go b/collections/colblock.go index ae0bedcc..8e3c1cf9 100644 --- a/collections/colblock.go +++ b/collections/colblock.go @@ -4,6 +4,7 @@ import ( "encoding/binary" "errors" "math" + "sort" "sync/atomic" "github.com/PelicanPlatform/classad/collections/wire" @@ -243,7 +244,10 @@ type columnarBlock struct { // Per-field string dictionaries and their code columns (see strdict.go): a string predicate compares // codes instead of walking the positional string region. Empty when no string field in this block had // values repetitive enough to be worth one. - strDict map[int]strDictField + // strDict is a sorted slice keyed by field index (binary search via strDictOf), not a map, for + // the same reason as zones: it removed hundreds of MB of per-block Go-map overhead. On-disk form + // unchanged (already written idx-sorted). + strDict []strDictEntry strDictComp []byte // the distinct values, fold-sorted, compressed strCodeComp []byte // the code columns, columnar, compressed @@ -254,7 +258,11 @@ type columnarBlock struct { // // Computed while encoding, where the values are already in hand -- a separate pass would cost the // scan it is meant to save. - zones map[int]blockZone + // + // A sorted slice keyed by field index (binary search via zone()), not a map: the map's bucket + // overhead was ~2x the entries and, at one map per block over ~100K blocks, hundreds of MB of + // live heap. The on-disk form is unchanged -- the entries are just parsed into a slice. + zones []blockZoneEntry // escClass says, per schema field, whether this block's escapes of it are all MISSING (the // attribute is absent, so the escape bit proves it undefined), all EXCEPTIONAL (present but @@ -285,6 +293,39 @@ type blockZone struct { escaped bool } +// blockZoneEntry is one numeric field's zone in a block's zone slice, which is sorted by idx so a +// lookup binary-searches (see zone) instead of hashing -- the slice replaced a per-block map whose +// bucket overhead dominated columnar metadata heap. +type blockZoneEntry struct { + idx int + blockZone +} + +// zone returns field idx's numeric zone and whether the block carries one. +func (b *columnarBlock) zone(idx int) (blockZone, bool) { + i := sort.Search(len(b.zones), func(i int) bool { return b.zones[i].idx >= idx }) + if i < len(b.zones) && b.zones[i].idx == idx { + return b.zones[i].blockZone, true + } + return blockZone{}, false +} + +// strDictEntry is one string field's dictionary metadata in a block's strDict slice, sorted by idx +// (see strDictOf). Replaced a per-block map for the same heap reason as zones. +type strDictEntry struct { + idx int + strDictField +} + +// strDictOf returns field idx's string-dictionary metadata and whether the block has one. +func (b *columnarBlock) strDictOf(idx int) (strDictField, bool) { + i := sort.Search(len(b.strDict), func(i int) bool { return b.strDict[i].idx >= idx }) + if i < len(b.strDict) && b.strDict[i].idx == idx { + return b.strDict[i].strDictField, true + } + return strDictField{}, false +} + // encodeColumnarBlock builds a columnar block from row-form records (each the output of // adSchema.encode). hotNumFields lists the schema field indices (int/real) to keep uncompressed // -- the popular ones, by query demand. Bools and the escape bitmap are always in the hot @@ -417,7 +458,7 @@ func encodeColumnarBlock(s *adSchema, recs [][]byte, layout *colLayout, regionCo b.zones = numericZones(s, b, recs) b.escClass, b.escExcRecs, b.escAbsent = classifyEscapes(s, recs, coldToField) if dicts != nil { - b.strDict = dicts + b.strDict = sortStrDict(dicts) b.strDictComp = regionCodec.Compress(nil, dictRaw) b.strCodeComp = regionCodec.Compress(nil, codeRaw) } @@ -567,11 +608,11 @@ func buildColumnarFromSegmentGrouped(data []byte, upto int, arenaCodec, regionCo // numericZones computes each numeric field's [min,max] over the records being encoded, and whether // any record escaped that field. Reads the row-form records directly, since encodeColumnarBlock has // them in hand. -func numericZones(s *adSchema, b *columnarBlock, recs [][]byte) map[int]blockZone { +func numericZones(s *adSchema, b *columnarBlock, recs [][]byte) []blockZoneEntry { if len(recs) == 0 { return nil } - out := make(map[int]blockZone, len(b.hotFields())+len(b.coldFields())) + out := make([]blockZoneEntry, 0, len(b.hotFields())+len(b.coldFields())) for _, idx := range append(append([]int(nil), b.hotFields()...), b.coldFields()...) { f := s.fields[idx] z := blockZone{zoneRange: zoneRange{Min: math.Inf(1), Max: math.Inf(-1)}} @@ -598,15 +639,16 @@ func numericZones(s *adSchema, b *columnarBlock, recs [][]byte) map[int]blockZon // Every record escaped: there is no range, and the block can never be pruned on it. z.Min, z.Max, z.escaped = math.Inf(-1), math.Inf(1), true } - out[idx] = z + out = append(out, blockZoneEntry{idx: idx, blockZone: z}) } + sort.Slice(out, func(i, j int) bool { return out[i].idx < out[j].idx }) return out } // mayMatch reports whether this block could hold a record satisfying every test on field idx. A block // with no zone for the field, or an inexact one (some record escaped), is never ruled out. func (b *columnarBlock) mayMatch(idx int, tests []zoneTest) bool { - z, ok := b.zones[idx] + z, ok := b.zone(idx) if !ok || z.escaped { return true } @@ -626,7 +668,7 @@ func (b *columnarBlock) mayMatch(idx int, tests []zoneTest) bool { // plus a bit test from every value read. Only numeric fields carry a zone, which is where the hot // scans are. func (b *columnarBlock) escapeFree(fieldIdx int) bool { - z, ok := b.zones[fieldIdx] + z, ok := b.zone(fieldIdx) return ok && !z.escaped } diff --git a/collections/colblock_layout_test.go b/collections/colblock_layout_test.go index 691dfbc3..ce026cdb 100644 --- a/collections/colblock_layout_test.go +++ b/collections/colblock_layout_test.go @@ -68,6 +68,12 @@ func TestColBlockLayoutSharedAcrossBlocks(t *testing.T) { if got.blocks[0].layout != got.blocks[1].layout { t.Error("reopened base blocks do not share one colLayout -- per-block layout has regressed") } + // The fixture has numeric fields (zones) and a low-cardinality Owner (strDict), so the slice + // checks below are not vacuous. + if len(got.blocks[0].zones) == 0 || len(got.blocks[0].strDict) == 0 { + t.Fatalf("fixture produced no zones (%d) or strDict (%d); the slice checks would be vacuous", + len(got.blocks[0].zones), len(got.blocks[0].strDict)) + } // Offsets aliased as packed u32: every record must reconstruct identically to the source block. for bi, b := range []*columnarBlock{b1, b2} { gb := got.blocks[bi] @@ -75,6 +81,28 @@ func TestColBlockLayoutSharedAcrossBlocks(t *testing.T) { t.Fatalf("block %d offset arrays not packed u32: strOffB=%d coldOffB=%d (n=%d)", bi, len(gb.strOffB), len(gb.coldOffB), gb.n) } + // zones and strDict are sorted-by-idx slices (not maps): the accessors binary-search, so the + // slices must be ordered, and every entry must be findable. + for i := 1; i < len(gb.zones); i++ { + if gb.zones[i-1].idx >= gb.zones[i].idx { + t.Fatalf("block %d zones not sorted by idx: %v", bi, gb.zones) + } + } + for _, e := range gb.zones { + if z, ok := gb.zone(e.idx); !ok || z != e.blockZone { + t.Errorf("block %d zone(%d) not found or mismatched", bi, e.idx) + } + } + for i := 1; i < len(gb.strDict); i++ { + if gb.strDict[i-1].idx >= gb.strDict[i].idx { + t.Fatalf("block %d strDict not sorted by idx: %v", bi, gb.strDict) + } + } + for _, e := range gb.strDict { + if info, ok := gb.strDictOf(e.idx); !ok || info != e.strDictField { + t.Errorf("block %d strDictOf(%d) not found or mismatched", bi, e.idx) + } + } for k := 0; k < b.n; k++ { a, err := b.reconstruct(k, nil) if err != nil { diff --git a/collections/colpersist.go b/collections/colpersist.go index c7c908a5..482ed52a 100644 --- a/collections/colpersist.go +++ b/collections/colpersist.go @@ -292,12 +292,12 @@ func appendColBlock(dst []byte, b *columnarBlock) []byte { dst = appendU32(dst, uint32(len(b.coldOffB)/4)) dst = append(dst, b.coldOffB...) dst = appendU32(dst, uint32(len(b.zones))) - for idx, z := range b.zones { - dst = appendU32(dst, uint32(idx)) - dst = appendU64(dst, math.Float64bits(z.Min)) - dst = appendU64(dst, math.Float64bits(z.Max)) + for _, e := range b.zones { // b.zones is sorted by idx + dst = appendU32(dst, uint32(e.idx)) + dst = appendU64(dst, math.Float64bits(e.Min)) + dst = appendU64(dst, math.Float64bits(e.Max)) esc := uint32(0) - if z.escaped { + if e.escaped { esc = 1 } dst = appendU32(dst, esc) @@ -325,20 +325,14 @@ func appendColBlock(dst []byte, b *columnarBlock) []byte { dst = appendBytes(dst, b.strDictComp) dst = appendBytes(dst, b.strCodeComp) dst = appendU32(dst, uint32(len(b.strDict))) - dictIdxs := make([]int, 0, len(b.strDict)) - for idx := range b.strDict { - dictIdxs = append(dictIdxs, idx) - } - sort.Ints(dictIdxs) - for _, idx := range dictIdxs { - info := b.strDict[idx] - dst = appendU32(dst, uint32(idx)) - dst = appendU32(dst, uint32(info.codeStart)) - dst = appendU32(dst, uint32(info.codeWidth)) - dst = appendU32(dst, uint32(info.dictStart)) - dst = appendU32(dst, uint32(info.count)) + for _, e := range b.strDict { // b.strDict is sorted by idx + dst = appendU32(dst, uint32(e.idx)) + dst = appendU32(dst, uint32(e.codeStart)) + dst = appendU32(dst, uint32(e.codeWidth)) + dst = appendU32(dst, uint32(e.dictStart)) + dst = appendU32(dst, uint32(e.count)) ne := uint32(0) - if info.noEscape { + if e.noEscape { ne = 1 } dst = appendU32(dst, ne) @@ -368,7 +362,7 @@ func readColBlock(c *cursor, s *adSchema, layout *colLayout, codec Codec) *colum if nz > len(s.fields) || !c.need(0) { return nil } - b.zones = make(map[int]blockZone, nz) + b.zones = make([]blockZoneEntry, 0, nz) for j := 0; j < nz; j++ { idx := int(c.u32()) z := blockZone{zoneRange: zoneRange{ @@ -379,8 +373,11 @@ func readColBlock(c *cursor, s *adSchema, layout *colLayout, codec Codec) *colum if idx < 0 || idx >= len(s.fields) { return nil } - b.zones[idx] = z + b.zones = append(b.zones, blockZoneEntry{idx: idx, blockZone: z}) } + // zone() binary-searches, so keep the slice ordered even if an older writer stored the + // entries in map-iteration order. + sort.Slice(b.zones, func(i, j int) bool { return b.zones[i].idx < b.zones[j].idx }) } b.escClass = c.bytes() if len(b.escClass) != 0 && len(b.escClass) != len(s.fields) { @@ -418,7 +415,7 @@ func readColBlock(c *cursor, s *adSchema, layout *colLayout, codec Codec) *colum if nd > len(s.fields) || !c.need(0) { return nil } - b.strDict = make(map[int]strDictField, nd) + b.strDict = make([]strDictEntry, 0, nd) for j := 0; j < nd; j++ { idx := int(c.u32()) info := strDictField{ @@ -431,8 +428,10 @@ func readColBlock(c *cursor, s *adSchema, layout *colLayout, codec Codec) *colum if idx < 0 || idx >= len(s.fields) || (info.codeWidth != 1 && info.codeWidth != 2) { return nil } - b.strDict[idx] = info + b.strDict = append(b.strDict, strDictEntry{idx: idx, strDictField: info}) } + // strDictOf binary-searches; the on-disk entries are already idx-sorted, but sort defensively. + sort.Slice(b.strDict, func(i, j int) bool { return b.strDict[i].idx < b.strDict[j].idx }) } if c.err != nil { return nil diff --git a/collections/colvec.go b/collections/colvec.go index c56fe8f6..473c9959 100644 --- a/collections/colvec.go +++ b/collections/colvec.go @@ -146,7 +146,7 @@ func (s *blockVecSource) loadStr(idx int, id uint32, dst *vm.Vec) bool { // block, then each record is a fixed-width code. if entries, ok := b.dictEntries(idx, s.bc, s.dictBufFor(idx)); ok { if codes, w, ok := b.dictCodes(idx, s.bc); ok { - info := b.strDict[idx] + info, _ := b.strDictOf(idx) // CODES, not strings, when every record has one. Then a comparison against a literal is an // integer range test in the executor and no string is materialized at all. With escapes present // the column is mixed -- an escaped value comes from the cold tail as a real string, and a code diff --git a/collections/strdict.go b/collections/strdict.go index 72b5da7a..68f92c17 100644 --- a/collections/strdict.go +++ b/collections/strdict.go @@ -8,6 +8,21 @@ import ( "github.com/PelicanPlatform/classad/classad" ) +// sortStrDict converts the encoder's per-field string-dictionary map into the sorted-by-idx slice a +// block holds, which strDictOf binary-searches -- the slice replaced a per-block map whose bucket +// overhead was a large share of columnar metadata heap. +func sortStrDict(m map[int]strDictField) []strDictEntry { + if len(m) == 0 { + return nil + } + out := make([]strDictEntry, 0, len(m)) + for idx, f := range m { + out = append(out, strDictEntry{idx: idx, strDictField: f}) + } + sort.Slice(out, func(i, j int) bool { return out[i].idx < out[j].idx }) + return out +} + // A per-block, per-field STRING DICTIONARY, so a string predicate compares integers. // // The string region is POSITIONAL -- uvarint(len)+bytes for each non-escaped string field in schema order @@ -233,7 +248,7 @@ func bytesToStr(b []byte) string { // buf is the caller's scratch, reused across blocks. Allocating it here made a 512-entry dictionary ~12 KB // of garbage per block -- across a segment, more garbage than the walk it replaces ever cost. func (b *columnarBlock) dictEntries(fieldIdx int, bc *blockCache, buf *[][]byte) ([][]byte, bool) { - info, ok := b.strDict[fieldIdx] + info, ok := b.strDictOf(fieldIdx) if !ok { return nil, false } @@ -260,7 +275,7 @@ func (b *columnarBlock) dictEntries(fieldIdx int, bc *blockCache, buf *[][]byte) // dictCodes returns field fieldIdx's code column, its width, and the decompressed code region. func (b *columnarBlock) dictCodes(fieldIdx int, bc *blockCache) ([]byte, int, bool) { - info, ok := b.strDict[fieldIdx] + info, ok := b.strDictOf(fieldIdx) if !ok { return nil, 0, false } @@ -301,7 +316,7 @@ func appendNonDictStrings(s *adSchema, r []byte, dicts map[int]strDictField, dst // dictOwns reports whether the dictionary is authoritative for this field, so a positional walk must skip it. func (b *columnarBlock) dictOwns(fieldIdx int) bool { - _, ok := b.strDict[fieldIdx] + _, ok := b.strDictOf(fieldIdx) return ok } @@ -347,7 +362,7 @@ func (b *columnarBlock) dictRange(fieldIdx int, lit string, bc *blockCache, buf // the cold tail where the dictionary cannot see it, so a partial dictionary must not prune. func (b *columnarBlock) dictPrunes(probes []strProbe, bc *blockCache, buf *[][]byte) bool { for _, p := range probes { - info, ok := b.strDict[p.fieldIdx] + info, ok := b.strDictOf(p.fieldIdx) if !ok || !info.noEscape { continue } diff --git a/collections/strdict_test.go b/collections/strdict_test.go index b5b03456..f7341192 100644 --- a/collections/strdict_test.go +++ b/collections/strdict_test.go @@ -68,7 +68,7 @@ func dictStats(t *testing.T, c *Collection, attr string) (withDict, blocks int) } for _, blk := range seg.blocks { blocks++ - if _, ok := blk.strDict[idx]; ok { + if _, ok := blk.strDictOf(idx); ok { withDict++ } } @@ -176,7 +176,7 @@ func BenchmarkStrDict(b *testing.B) { } if idx, ok := seg.schema().byID[id]; ok { for _, blk := range seg.blocks { - if _, ok := blk.strDict[idx]; ok { + if _, ok := blk.strDictOf(idx); ok { n++ } }