collections: drop byOff map + alias offsets; block zones/strDict maps -> slices

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
 					}

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
 }
 

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 {
@@ -41,13 +68,41 @@ 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]
 		if len(gb.strOffB) != (gb.n+1)*4 || len(gb.coldOffB) != (gb.n+1)*4 {
 			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 {

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

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 {

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)
 		}

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 " +

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) }
 

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

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 {

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