From a6e5ee1a1c94b736787c1fff386326ffda1c6839 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 16:57:55 +0700 Subject: [PATCH 01/11] db/etl: split the sortable buffer into pooled 1MB chunks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sortableBuffer held every key/value byte in one slice, so a Collect that outgrew it re-allocated and copied everything collected so far - for a 256MB buffer that is ~256MB of memmove and ~512MB live at the peak. Store the bytes in 1MB chunks drawn from a shared sync.Pool: Put never copies what is already there, Reset hands the chunks back, and an idle pooled buffer no longer pins its busiest run's RAM. entryLoc.offset packs the chunk index with the offset inside the chunk; an entry never straddles a chunk, so Get still returns direct references. An entry larger than one chunk gets a private chunk of its own. BenchmarkCollect (no Prealloc, the path collectors actually take): 10k_largebuf 518.8µs -> 116.8µs -77% 8.7MB/op -> 0.7MB/op 100k_largebuf 3.286ms -> 1.137ms -65% 87MB/op -> 8.5MB/op Cost of the extra indirection, both on prealloc'd buffers that never grow: Put +11%, Sort +8..24%. --- db/etl/buffers.go | 159 ++++++++++++++++++++++++++++++++++---------- db/etl/collector.go | 4 +- db/etl/etl_test.go | 86 +++++++++++++++++++++++- 3 files changed, 208 insertions(+), 41 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 844e15770ef..bfa75cee44d 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -26,6 +26,7 @@ import ( "sort" "strconv" "sync" + "unsafe" "github.com/c2h5oh/datasize" @@ -81,9 +82,9 @@ var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var // etlSmallBufRAM (BufferOptimalSize/8) bounds the flush threshold so a full // set of domain/history/index flush collectors (~17 per batch writer) stays -// around 512 MB when all run full. Pooled buffers start empty and grow with -// the data they actually see; grown capacity survives reuse (Reset preserves -// cap), so hot collectors amortize growth while never-full ones stay small. +// around 512 MB when all run full. Pooled buffers start empty and take chunks +// as they fill; Reset returns those chunks to the shared dataChunks pool, so an +// idle buffer doesn't pin the RAM its busiest run needed. var etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) var SmallSortableBuffers = NewAllocator(&sync.Pool{ New: func() any { @@ -97,6 +98,31 @@ var LargeSortableBuffers = NewAllocator(&sync.Pool{ }, }) +const ( + // sortableBuffer stores key/value bytes in dataChunkSize blocks. entryLoc.offset + // packs the chunk index and the offset inside the chunk, so the index range is + // what limits one buffer to maxDataChunks. + dataChunkBits = 20 + dataChunkSize = 1 << dataChunkBits + maxDataChunks = math.MaxInt32>>dataChunkBits + 1 +) + +// dataChunks are shared by all sortableBuffer instances: a buffer takes chunks as +// it fills and gives them back on Reset, instead of pinning its peak size forever. +var dataChunks = sync.Pool{New: func() any { + c := make([]byte, dataChunkSize) + return &c +}} + +func getDataChunk() []byte { return *dataChunks.Get().(*[]byte) } + +func putDataChunk(c []byte) { + if len(c) != dataChunkSize { // private chunk of an oversized entry + return + } + dataChunks.Put(&c) +} + type Buffer interface { // Put does copy `k` and `v` Put(k, v []byte) @@ -123,9 +149,10 @@ var ( _ Buffer = &oldestEntrySortableBuffer{} ) -// entryLoc stores the location of a key/value pair within sortableBuffer.data. -// Key occupies data[offset : offset+keyLen], value follows at data[offset+max(0,keyLen) : ...+valLen]. -// keyLen/valLen of -1 indicates nil. +// entryLoc stores the location of a key/value pair inside sortableBuffer. +// offset packs the chunk index and the offset inside that chunk: +// idx<= maxDataChunks { + panic(fmt.Sprintf("etl: sortableBuffer exceeded %d chunks", maxDataChunks)) + } + if n > dataChunkSize { + b.cur = make([]byte, n) + } else { + b.cur = getDataChunk() + } + b.chunks = append(b.chunks, b.cur) + b.bases = append(b.bases, unsafe.Pointer(&b.cur[0])) + b.curBase = int32(len(b.chunks)-1) << dataChunkBits //nolint:gosec + b.curOff = 0 + b.chunkBytes += len(b.cur) +} + +// entryData points at e's first byte: the key, immediately followed by the value. +// bases[i] is chunks[i]'s first byte - one load per lookup instead of a slice +// header, which the sort comparator does twice per comparison. +func (b *sortableBuffer) entryData(e *entryLoc) unsafe.Pointer { + return unsafe.Add(b.bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) +} + // Put adds key and value to the buffer. These slices will not be accessed later, // so no copying is necessary func (b *sortableBuffer) Put(k, v []byte) { e := entryLoc{ - offset: int32(len(b.data)), //nolint:gosec keyLen: int32(len(k)), //nolint:gosec valLen: int32(len(v)), //nolint:gosec insertionOrder: int32(len(b.entries)), //nolint:gosec @@ -163,11 +223,26 @@ func (b *sortableBuffer) Put(k, v []byte) { if v == nil { e.valLen = -1 } + if n := len(k) + len(v); n > 0 { + off := b.curOff + if int(off)+n > len(b.cur) { + b.nextChunk(n) + off = 0 + } + data := b.cur[off:] + copy(data, k) + copy(data[len(k):], v) + e.offset = b.curBase | off + b.curOff = off + int32(n) //nolint:gosec + } b.entries = append(b.entries, e) - b.data = append(append(b.data, k...), v...) } -func (b *sortableBuffer) Size() int { return len(b.data) + len(b.entries)*entryLocSize } +// Size counts the bytes of every chunk taken so far, minus the unused tail of +// the chunk being filled - so it tracks RAM held, not just bytes stored. +func (b *sortableBuffer) Size() int { + return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize +} func (b *sortableBuffer) Len() int { return len(b.entries) @@ -176,22 +251,24 @@ func (b *sortableBuffer) Len() int { func (b *sortableBuffer) Get(i int) ([]byte, []byte) { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - keyOffset := int(e.offset) - valOffset := keyOffset - if kLen > 0 { - valOffset += kLen - } var key, val []byte - if kLen > 0 { - key = b.data[keyOffset : keyOffset+kLen] - } else if kLen == 0 { + if kLen == 0 { key = []byte{} } - if vLen > 0 { - val = b.data[valOffset : valOffset+vLen] - } else if vLen == 0 { + if vLen == 0 { val = []byte{} } + if kLen <= 0 && vLen <= 0 { + return key, val + } + p := b.entryData(e) + if kLen > 0 { + key = unsafe.Slice((*byte)(p), kLen) + p = unsafe.Add(p, kLen) + } + if vLen > 0 { + val = unsafe.Slice((*byte)(p), vLen) + } return key, val } @@ -199,23 +276,35 @@ func (b *sortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer if cap(b.entries) < predictKeysAmount { b.entries = make([]entryLoc, 0, predictKeysAmount) } - if cap(b.data) < predictDataSize { - b.data = make([]byte, 0, predictDataSize) + if n := predictDataSize/dataChunkSize + 1; cap(b.chunks) < n { + b.chunks = slices.Grow(b.chunks, n) + b.bases = slices.Grow(b.bases, n) } return b } func (b *sortableBuffer) Reset() { b.entries = b.entries[:0] - b.data = b.data[:0] + for i, c := range b.chunks { + putDataChunk(c) + b.chunks[i], b.bases[i] = nil, nil + } + b.chunks, b.bases = b.chunks[:0], b.bases[:0] + b.cur, b.curBase, b.curOff = nil, 0, 0 + b.chunkBytes = 0 } func (b *sortableBuffer) SizeLimit() int { return b.optimalSize } func (b *sortableBuffer) Sort() { - data := b.data + bases := b.bases + key := func(e entryLoc) []byte { + if e.keyLen <= 0 { + return nil + } + p := unsafe.Add(bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) + return unsafe.Slice((*byte)(p), e.keyLen) + } cmp := func(a, b entryLoc) int { - aKey := data[a.offset : a.offset+max(a.keyLen, 0)] - bKey := data[b.offset : b.offset+max(b.keyLen, 0)] - if c := bytes.Compare(aKey, bKey); c != 0 { + if c := bytes.Compare(key(a), key(b)); c != 0 { return c } return int(a.insertionOrder - b.insertionOrder) // StableSort: preserve insertion order for duplicate keys @@ -235,10 +324,9 @@ func (b *sortableBuffer) Write(w io.Writer) error { for i := range b.entries { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - keyOffset := int(e.offset) - valOffset := keyOffset - if kLen > 0 { - valOffset += kLen + var p unsafe.Pointer + if kLen > 0 || vLen > 0 { + p = b.entryData(e) } // write key n := binary.PutVarint(numBuf[:], int64(e.keyLen)) @@ -246,9 +334,10 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if kLen > 0 { - if _, err := w.Write(b.data[keyOffset : keyOffset+kLen]); err != nil { + if _, err := w.Write(unsafe.Slice((*byte)(p), kLen)); err != nil { return err } + p = unsafe.Add(p, kLen) } // write value n = binary.PutVarint(numBuf[:], int64(e.valLen)) @@ -256,7 +345,7 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if vLen > 0 { - if _, err := w.Write(b.data[valOffset : valOffset+vLen]); err != nil { + if _, err := w.Write(unsafe.Slice((*byte)(p), vLen)); err != nil { return err } } diff --git a/db/etl/collector.go b/db/etl/collector.go index f1ceab82cd0..7fd9663dec8 100644 --- a/db/etl/collector.go +++ b/db/etl/collector.go @@ -45,9 +45,7 @@ func (a *Allocator) Put(b Buffer) { if b == nil { return } - //if cast, ok := b.(*sortableBuffer); ok { - // log.Warn("[dbg] return buf", "cap(cast.data)", cap(cast.data), "cap(cast.lens)", cap(cast.lens)) - //} + b.Reset() // release the data chunks now: an idle pooled buffer must not pin them a.p.Put(b) } func (a *Allocator) Get() Buffer { diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 3887b1a1662..eb6ee2860e0 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -495,10 +495,10 @@ func TestReuseCollectorAfterLoad(t *testing.T) { require.Equal(t, 1, see) c.Close() - // buffers are not lost - require.Empty(t, buf.data) + // buffers are not lost: entries keep their cap, data chunks went back to the pool + require.Empty(t, buf.chunks) require.Empty(t, buf.entries) - require.NotZero(t, cap(buf.data)) + require.Zero(t, buf.Size()) require.NotZero(t, cap(buf.entries)) // teset that no data visible @@ -1574,3 +1574,83 @@ func TestCollectorWithAllocatorDrawsBufferLazily(t *testing.T) { require.NoError(err) require.Equal([]byte{1}, v) } + +// TestSortableBufferChunks pins the chunked layout: key/value bytes live in +// fixed-size chunks, so a growing buffer never re-allocates and copies the +// bytes it already holds. +func TestSortableBufferChunks(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + + const entries = 512 + val := bytes.Repeat([]byte{0xAB}, 16*1024) // 512*16KB = 8MB of values + key := make([]byte, 8) + for i := range entries { + binary.BigEndian.PutUint64(key, uint64(i)) + buf.Put(key, val) + } + + require.Equal(t, entries, buf.Len()) + require.Greater(t, len(buf.chunks), 1, "data must be split into chunks") + for i, c := range buf.chunks { + require.Equal(t, dataChunkSize, cap(c), "chunk %d", i) + } + + for i := range entries { + binary.BigEndian.PutUint64(key, uint64(i)) + k, v := buf.Get(i) + require.Equal(t, key, k, "entry %d", i) + require.Equal(t, val, v, "entry %d", i) + } +} + +// TestSortableBufferOversizedEntry: an entry bigger than one chunk gets a chunk +// of its own - Get must still return one contiguous slice per key and value. +func TestSortableBufferOversizedEntry(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + + big := bytes.Repeat([]byte{0xCD}, dataChunkSize+7) + buf.Put([]byte{0x01}, []byte("small")) + buf.Put([]byte{0x02}, big) + buf.Put([]byte{0x03}, []byte("after")) + + k, v := buf.Get(1) + require.Equal(t, []byte{0x02}, k) + require.Equal(t, big, v) + k, v = buf.Get(2) + require.Equal(t, []byte{0x03}, k) + require.Equal(t, []byte("after"), v) + + w := bytes.NewBuffer(nil) + require.NoError(t, buf.Write(w)) + m := &mmapBytesReader{data: w.Bytes()} + for i := range buf.Len() { + wantK, wantV := buf.Get(i) + gotK, err := readField(m) + require.NoError(t, err) + gotV, err := readField(m) + require.NoError(t, err) + require.Equal(t, wantK, gotK) + require.Equal(t, wantV, gotV) + } +} + +// TestSortableBufferResetReleasesChunks: Reset hands the chunks back to the +// shared pool, so an idle pooled buffer doesn't pin the RAM it once needed. +func TestSortableBufferResetReleasesChunks(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + val := bytes.Repeat([]byte{0xEF}, 16*1024) + for i := range 512 { + buf.Put(binary.BigEndian.AppendUint64(nil, uint64(i)), val) + } + require.NotEmpty(t, buf.chunks) + + buf.Reset() + require.Empty(t, buf.chunks) + require.Zero(t, buf.Size()) + require.Zero(t, buf.Len()) + + buf.Put([]byte{0x01}, []byte("reused")) + k, v := buf.Get(0) + require.Equal(t, []byte{0x01}, k) + require.Equal(t, []byte("reused"), v) +} From 04b416d0adbd477968a3fd9b7e67d4b57046df47 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:03:11 +0700 Subject: [PATCH 02/11] db/etl: say what Size counts --- db/etl/buffers.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index bfa75cee44d..a89ee2b4efd 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -238,8 +238,7 @@ func (b *sortableBuffer) Put(k, v []byte) { b.entries = append(b.entries, e) } -// Size counts the bytes of every chunk taken so far, minus the unused tail of -// the chunk being filled - so it tracks RAM held, not just bytes stored. +// Size counts the stored bytes plus the tails wasted by the chunks already filled. func (b *sortableBuffer) Size() int { return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize } From b6b69c6ea8a20a6c0885bc3c73dcf0e89e272fc5 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:20:01 +0700 Subject: [PATCH 03/11] db/etl: dispose providers before recycling the buffer Close released the buffer first, and a KeepInRAM provider reads straight from it. That was harmless while Reset only truncated the buffer's own slice; now Reset hands the chunks to a pool other collectors draw from, so any read after Close would race a different goroutine's writes. --- db/etl/collector.go | 14 ++++++++------ db/etl/etl_test.go | 28 ++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/db/etl/collector.go b/db/etl/collector.go index 7fd9663dec8..5d802fd592c 100644 --- a/db/etl/collector.go +++ b/db/etl/collector.go @@ -261,6 +261,14 @@ func (c *Collector) Load(db kv.RwTx, toBucket string, loadFunc LoadFunc, args Tr } func (c *Collector) Close() { + // Providers first: a KeepInRAM one reads straight from `buf`, whose chunks + // Reset hands to a pool that other collectors draw from. + if c.dataProviders != nil { //idempotency + for _, p := range c.dataProviders { + p.Dispose() + } + c.dataProviders = nil + } if c.buf != nil { //idempotency if c.allocator != nil { c.allocator.Put(c.buf) @@ -269,12 +277,6 @@ func (c *Collector) Close() { c.buf.Reset() } } - if c.dataProviders != nil { //idempotency - for _, p := range c.dataProviders { - p.Dispose() - } - c.dataProviders = nil - } c.allFlushed = false } diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index eb6ee2860e0..59a4bf0949e 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -1654,3 +1654,31 @@ func TestSortableBufferResetReleasesChunks(t *testing.T) { require.Equal(t, []byte{0x01}, k) require.Equal(t, []byte("reused"), v) } + +// disposeProbe records whether the collector still owned its data chunks when +// the provider was disposed. +type disposeProbe struct { + buf *sortableBuffer + sawOwnChunks bool +} + +func (p *disposeProbe) Next() ([]byte, []byte, error) { return nil, nil, io.EOF } +func (p *disposeProbe) Wait() error { return nil } +func (p *disposeProbe) String() string { return "disposeProbe" } +func (p *disposeProbe) Dispose() { p.sawOwnChunks = len(p.buf.chunks) > 0 } + +// TestCloseDisposesProvidersBeforeBuffer: KeepInRAM hands out a provider backed +// by the collector's own buffer, and Reset gives that buffer's chunks to a pool +// other collectors draw from. So Close must be done with every provider before +// it recycles the buffer. +func TestCloseDisposesProvidersBeforeBuffer(t *testing.T) { + allocator := NewAllocator(&sync.Pool{New: func() any { return NewSortableBuffer(BufferOptimalSize) }}) + c := NewCollectorWithAllocator(t.Name(), t.TempDir(), allocator, log.New()) + require.NoError(t, c.Collect([]byte{1}, []byte{2})) + + probe := &disposeProbe{buf: c.buf.(*sortableBuffer)} + c.dataProviders = append(c.dataProviders, probe) + c.Close() + + require.True(t, probe.sawOwnChunks, "buffer was recycled before its providers were disposed") +} From ab9de1211c333ef892463395516f78c4014b5a50 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:29:07 +0700 Subject: [PATCH 04/11] db/etl: drop the unsafe base-pointer table Look up chunk bytes through the chunks slice itself. The parallel []unsafe.Pointer bought ~12 points of Sort geomean by saving a slice-header load per key, but it duplicated ownership of every chunk to buy back part of a cost the chunking itself creates. --- db/etl/buffers.go | 40 +++++++++++++++++----------------------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index a89ee2b4efd..13fd8331e2b 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -26,7 +26,6 @@ import ( "sort" "strconv" "sync" - "unsafe" "github.com/c2h5oh/datasize" @@ -176,7 +175,6 @@ type sortableBuffer struct { // far. All chunks are dataChunkSize, except the private chunk an entry // larger than that gets. cur is the chunk being filled. chunks [][]byte - bases []unsafe.Pointer cur []byte curBase int32 // packed location of cur's first byte: curIdx<>dataChunkBits], e.offset&(dataChunkSize-1)) +// entryData returns e's bytes: the key, immediately followed by the value. +func (b *sortableBuffer) entryData(e *entryLoc) []byte { + return b.chunks[e.offset>>dataChunkBits][e.offset&(dataChunkSize-1):] } // Put adds key and value to the buffer. These slices will not be accessed later, @@ -260,13 +255,13 @@ func (b *sortableBuffer) Get(i int) ([]byte, []byte) { if kLen <= 0 && vLen <= 0 { return key, val } - p := b.entryData(e) + data := b.entryData(e) if kLen > 0 { - key = unsafe.Slice((*byte)(p), kLen) - p = unsafe.Add(p, kLen) + key = data[:kLen:kLen] + data = data[kLen:] } if vLen > 0 { - val = unsafe.Slice((*byte)(p), vLen) + val = data[:vLen:vLen] } return key, val } @@ -277,7 +272,6 @@ func (b *sortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer } if n := predictDataSize/dataChunkSize + 1; cap(b.chunks) < n { b.chunks = slices.Grow(b.chunks, n) - b.bases = slices.Grow(b.bases, n) } return b } @@ -286,21 +280,21 @@ func (b *sortableBuffer) Reset() { b.entries = b.entries[:0] for i, c := range b.chunks { putDataChunk(c) - b.chunks[i], b.bases[i] = nil, nil + b.chunks[i] = nil } - b.chunks, b.bases = b.chunks[:0], b.bases[:0] + b.chunks = b.chunks[:0] b.cur, b.curBase, b.curOff = nil, 0, 0 b.chunkBytes = 0 } func (b *sortableBuffer) SizeLimit() int { return b.optimalSize } func (b *sortableBuffer) Sort() { - bases := b.bases + chunks := b.chunks key := func(e entryLoc) []byte { if e.keyLen <= 0 { return nil } - p := unsafe.Add(bases[e.offset>>dataChunkBits], e.offset&(dataChunkSize-1)) - return unsafe.Slice((*byte)(p), e.keyLen) + off := e.offset & (dataChunkSize - 1) + return chunks[e.offset>>dataChunkBits][off : off+e.keyLen] } cmp := func(a, b entryLoc) int { if c := bytes.Compare(key(a), key(b)); c != 0 { @@ -323,9 +317,9 @@ func (b *sortableBuffer) Write(w io.Writer) error { for i := range b.entries { e := &b.entries[i] kLen, vLen := int(e.keyLen), int(e.valLen) - var p unsafe.Pointer + var data []byte if kLen > 0 || vLen > 0 { - p = b.entryData(e) + data = b.entryData(e) } // write key n := binary.PutVarint(numBuf[:], int64(e.keyLen)) @@ -333,10 +327,10 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if kLen > 0 { - if _, err := w.Write(unsafe.Slice((*byte)(p), kLen)); err != nil { + if _, err := w.Write(data[:kLen]); err != nil { return err } - p = unsafe.Add(p, kLen) + data = data[kLen:] } // write value n = binary.PutVarint(numBuf[:], int64(e.valLen)) @@ -344,7 +338,7 @@ func (b *sortableBuffer) Write(w io.Writer) error { return err } if vLen > 0 { - if _, err := w.Write(unsafe.Slice((*byte)(p), vLen)); err != nil { + if _, err := w.Write(data[:vLen]); err != nil { return err } } From ff0ef48fe606bb10c75b2934e7d8a7ed98af4c0f Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:36:37 +0700 Subject: [PATCH 05/11] db/etl: say what the chunk constants mean 20 and 1<<20 read as magic: the power of two is what lets entryLoc.offset pack the chunk index with the in-chunk offset, and the index range is what caps a buffer at 2GB. --- db/etl/buffers.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 13fd8331e2b..5f6cbda8620 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -98,11 +98,15 @@ var LargeSortableBuffers = NewAllocator(&sync.Pool{ }) const ( - // sortableBuffer stores key/value bytes in dataChunkSize blocks. entryLoc.offset - // packs the chunk index and the offset inside the chunk, so the index range is - // what limits one buffer to maxDataChunks. + // sortableBuffer stores key/value bytes in chunks of a power-of-two size, so + // entryLoc.offset can pack the chunk index with the offset inside the chunk + // and splitting the two is a shift and a mask. 1MB is also the least a + // collector can hold once it takes a chunk at all. dataChunkBits = 20 - dataChunkSize = 1 << dataChunkBits + dataChunkSize = 1 << dataChunkBits // 1MB + + // The chunk index takes what is left of a positive int32, so one buffer + // addresses 2GB - the ceiling NewSortableBuffer already puts on optimalSize. maxDataChunks = math.MaxInt32>>dataChunkBits + 1 ) From 18a81256f5efd3e4e032d041001d984c6773eef4 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Sun, 23 Aug 2026 17:58:37 +0700 Subject: [PATCH 06/11] save --- db/etl/buffers.go | 41 ++++++++++++++++++++++++++--------------- 1 file changed, 26 insertions(+), 15 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 5f6cbda8620..57f23770c9f 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -33,15 +33,15 @@ import ( ) const ( - //SliceBuffer - just simple slice w + // SliceBuffer - just simple slice w SortableSliceBuffer = iota - //SortableAppendBuffer - map[k] [v1 v2 v3] + // SortableAppendBuffer - map[k] [v1 v2 v3] SortableAppendBuffer // SortableOldestAppearedBuffer - buffer that keeps only the oldest entries. // if first v1 was added under key K, then v2; only v1 will stay SortableOldestAppearedBuffer - //BufIOSize - 128 pages | default is 1 page | increasing over `64 * 4096` doesn't show speedup on SSD/NVMe, but show speedup in cloud drives + // BufIOSize - 128 pages | default is 1 page | increasing over `64 * 4096` doesn't show speedup on SSD/NVMe, but show speedup in cloud drives BufIOSize = 128 * 4096 entryLocSize = 16 // sizeof(entryLoc): insertionOrder(4) + offset(4) + keyLen(4) + valLen(4) @@ -84,18 +84,23 @@ var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var // around 512 MB when all run full. Pooled buffers start empty and take chunks // as they fill; Reset returns those chunks to the shared dataChunks pool, so an // idle buffer doesn't pin the RAM its busiest run needed. -var etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) -var SmallSortableBuffers = NewAllocator(&sync.Pool{ - New: func() any { - return NewSortableBuffer(etlSmallBufRAM) - }, -}) -var etlLargeBufRAM = BufferOptimalSize -var LargeSortableBuffers = NewAllocator(&sync.Pool{ - New: func() any { - return NewSortableBuffer(etlLargeBufRAM) - }, -}) +var ( + etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) + SmallSortableBuffers = NewAllocator(&sync.Pool{ + New: func() any { + return NewSortableBuffer(etlSmallBufRAM).Prealloc(int(etlSmallBufRAM/512), int(etlSmallBufRAM)) // SortableBuffer does Prealloc only metadata slices - not buffers itself + }, + }) +) + +var ( + etlLargeBufRAM = BufferOptimalSize + LargeSortableBuffers = NewAllocator(&sync.Pool{ + New: func() any { + return NewSortableBuffer(etlLargeBufRAM) + }, + }) +) const ( // sortableBuffer stores key/value bytes in chunks of a power-of-two size, so @@ -380,6 +385,7 @@ func (b *appendSortableBuffer) SizeLimit() int { return b.optimalSize } func (b *appendSortableBuffer) Len() int { return len(b.entries) } + func (b *appendSortableBuffer) Sort() { b.sortedBuf = b.sortedBuf[:0] if cap(b.sortedBuf) < len(b.entries) { @@ -402,11 +408,13 @@ func (b *appendSortableBuffer) Swap(i, j int) { func (b *appendSortableBuffer) Get(i int) ([]byte, []byte) { return b.sortedBuf[i].key, b.sortedBuf[i].value } + func (b *appendSortableBuffer) Reset() { b.sortedBuf = nil b.entries = make(map[string][]byte) b.size = 0 } + func (b *appendSortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer { b.entries = make(map[string][]byte, predictKeysAmount) // maps have no cap(), always recreate if cap(b.sortedBuf) < predictKeysAmount { @@ -478,11 +486,13 @@ func (b *oldestEntrySortableBuffer) Swap(i, j int) { func (b *oldestEntrySortableBuffer) Get(i int) ([]byte, []byte) { return b.sortedBuf[i].key, b.sortedBuf[i].value } + func (b *oldestEntrySortableBuffer) Reset() { b.sortedBuf = nil b.entries = make(map[string][]byte) b.size = 0 } + func (b *oldestEntrySortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer { b.entries = make(map[string][]byte, predictKeysAmount) // maps have no cap(), always recreate if cap(b.sortedBuf) < predictKeysAmount { @@ -494,6 +504,7 @@ func (b *oldestEntrySortableBuffer) Prealloc(predictKeysAmount, predictDataSize func (b *oldestEntrySortableBuffer) Write(w io.Writer) error { return writeSortedEntries(w, b.sortedBuf) } + func (b *oldestEntrySortableBuffer) CheckFlushSize() bool { return b.size >= b.optimalSize } From 801893ae079e556e8e3813161db32baf0a44c9df Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Mon, 24 Aug 2026 09:21:21 +0700 Subject: [PATCH 07/11] save --- db/etl/buffers.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 57f23770c9f..9ee28d184f1 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -88,7 +88,7 @@ var ( etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) SmallSortableBuffers = NewAllocator(&sync.Pool{ New: func() any { - return NewSortableBuffer(etlSmallBufRAM).Prealloc(int(etlSmallBufRAM/512), int(etlSmallBufRAM)) // SortableBuffer does Prealloc only metadata slices - not buffers itself + return NewSortableBuffer(etlSmallBufRAM) }, }) ) From 616397dfbb89c1e39fdfa178e925a09112eff057 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Mon, 24 Aug 2026 13:11:25 +0700 Subject: [PATCH 08/11] db/etl: address CR round-2 comments on the chunked sortable buffer - Correct maxDataChunks' comment: the panic is reachable near the top of optimalSize's range, not fully ruled out by NewSortableBuffer's bound - Consolidate the pooling rationale at dataChunks; restore the 17-collector derivation dropped from etlSmallBufRAM's comment - Complete Size()'s comment with the entryLoc metadata term - Fix stale sortableBuffer.data/GetRef references in test docstrings - Add TestDataChunkPoolRoundTrip and TestPutDataChunkRejectsOversized to cover putDataChunk's oversized-chunk guard, which had zero test coverage --- db/etl/buffers.go | 17 ++++++++++------- db/etl/collector.go | 2 +- db/etl/etl_test.go | 38 ++++++++++++++++++++++++++++++-------- 3 files changed, 41 insertions(+), 16 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 9ee28d184f1..6a9a1722ea4 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -79,11 +79,10 @@ func writeSortedEntries(w io.Writer, entries []sortableBufferEntry) error { var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var because we want to sometimes change it from tests or command-line flags */ -// etlSmallBufRAM (BufferOptimalSize/8) bounds the flush threshold so a full -// set of domain/history/index flush collectors (~17 per batch writer) stays -// around 512 MB when all run full. Pooled buffers start empty and take chunks -// as they fill; Reset returns those chunks to the shared dataChunks pool, so an -// idle buffer doesn't pin the RAM its busiest run needed. +// etlSmallBufRAM (BufferOptimalSize/8) bounds the flush threshold: +// 3_domains * 2 + 3_history * 1 + 4_indices * 2 = 17 etl collectors, +// 17*(256Mb/8) = 512Mb for all collectors combined. Buffers pool their +// chunks — see dataChunks below. var ( etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) SmallSortableBuffers = NewAllocator(&sync.Pool{ @@ -111,7 +110,10 @@ const ( dataChunkSize = 1 << dataChunkBits // 1MB // The chunk index takes what is left of a positive int32, so one buffer - // addresses 2GB - the ceiling NewSortableBuffer already puts on optimalSize. + // addresses at most maxDataChunks*dataChunkSize bytes (~2GB); nextChunk + // panics past that. NewSortableBuffer's MaxInt32 bound on optimalSize + // does not fully rule this out, since Put can grow the buffer past + // optimalSize before CheckFlushSize is checked. maxDataChunks = math.MaxInt32>>dataChunkBits + 1 ) @@ -242,7 +244,8 @@ func (b *sortableBuffer) Put(k, v []byte) { b.entries = append(b.entries, e) } -// Size counts the stored bytes plus the tails wasted by the chunks already filled. +// Size counts the stored bytes, the tail wasted by the chunk still filling, and +// entryLocSize bytes of metadata per entry. func (b *sortableBuffer) Size() int { return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize } diff --git a/db/etl/collector.go b/db/etl/collector.go index 5d802fd592c..6371dce78ae 100644 --- a/db/etl/collector.go +++ b/db/etl/collector.go @@ -45,7 +45,7 @@ func (a *Allocator) Put(b Buffer) { if b == nil { return } - b.Reset() // release the data chunks now: an idle pooled buffer must not pin them + b.Reset() // return the buffer's chunks to the pool now — see dataChunks in buffers.go a.p.Put(b) } func (a *Allocator) Get() Buffer { diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 59a4bf0949e..88b0804feb4 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -495,7 +495,7 @@ func TestReuseCollectorAfterLoad(t *testing.T) { require.Equal(t, 1, see) c.Close() - // buffers are not lost: entries keep their cap, data chunks went back to the pool + // buffer state resets for reuse: entries keep their cap, chunks are cleared require.Empty(t, buf.chunks) require.Empty(t, buf.entries) require.Zero(t, buf.Size()) @@ -624,9 +624,8 @@ func TestAppendAcrossProviders(t *testing.T) { } // TestAppendAcrossMemProviders tests that value concatenation works correctly -// when multiple memoryDataProviders have the same key. GetRef returns zero-copy -// slices into sortableBuffer.data — appending to prevV without copying would -// corrupt adjacent entries in the buffer. +// when multiple memoryDataProviders have the same key, across providers backed +// by different buffer types (file-flushed and in-memory). func TestAppendAcrossMemProviders(t *testing.T) { tmpdir := t.TempDir() @@ -917,7 +916,7 @@ func TestMixedProvidersInterleavedKeys(t *testing.T) { } // TestMixedProvidersZeroCopyIntegrity verifies that zero-copy slices from -// memoryDataProvider (GetRef) are not corrupted by subsequent Next() calls. +// memoryDataProvider (Get) are not corrupted by subsequent Next() calls. func TestMixedProvidersZeroCopyIntegrity(t *testing.T) { tmpdir := t.TempDir() @@ -927,7 +926,7 @@ func TestMixedProvidersZeroCopyIntegrity(t *testing.T) { fileProvider, err := FlushToDisk("test", fileBuf, tmpdir, log.LvlInfo) require.NoError(t, err) - // Memory provider with multiple keys - GetRef returns slices into sortableBuffer.data + // Memory provider with multiple keys - Get returns slices into sortableBuffer.chunks memBuf := NewSortableBuffer(BufferOptimalSize) memBuf.Put([]byte("bbb"), []byte("mem-bbb")) memBuf.Put([]byte("ccc"), []byte("mem-ccc")) @@ -1634,8 +1633,9 @@ func TestSortableBufferOversizedEntry(t *testing.T) { } } -// TestSortableBufferResetReleasesChunks: Reset hands the chunks back to the -// shared pool, so an idle pooled buffer doesn't pin the RAM it once needed. +// TestSortableBufferResetReleasesChunks: Reset drops the buffer's own chunk +// slice and size bookkeeping so it can be reused immediately. Pool round-tripping +// is TestDataChunkPoolRoundTrip's job. func TestSortableBufferResetReleasesChunks(t *testing.T) { buf := NewSortableBuffer(256 * datasize.MB) val := bytes.Repeat([]byte{0xEF}, 16*1024) @@ -1655,6 +1655,28 @@ func TestSortableBufferResetReleasesChunks(t *testing.T) { require.Equal(t, []byte("reused"), v) } +// TestDataChunkPoolRoundTrip: a chunk released via putDataChunk comes back on +// the next getDataChunk, instead of a fresh allocation. +func TestDataChunkPoolRoundTrip(t *testing.T) { + c := getDataChunk() + c[0] = 0xAA + putDataChunk(c) + + got := getDataChunk() + require.Equal(t, byte(0xAA), got[0]) +} + +// TestPutDataChunkRejectsOversized: an entry's private chunk (bigger than +// dataChunkSize) must never enter the shared pool — a later getDataChunk +// handing it out under a normal chunk index would corrupt an unrelated buffer. +func TestPutDataChunkRejectsOversized(t *testing.T) { + oversized := make([]byte, dataChunkSize+7) + putDataChunk(oversized) + + got := getDataChunk() + require.Len(t, got, dataChunkSize) +} + // disposeProbe records whether the collector still owned its data chunks when // the provider was disposed. type disposeProbe struct { From 3ebbb549529b9289dd53f0f45bdf644a255a7e98 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Mon, 24 Aug 2026 13:38:30 +0700 Subject: [PATCH 09/11] db/etl: drop the flaky sync.Pool round-trip test TestDataChunkPoolRoundTrip assumed a same-goroutine Put/Get pair on sync.Pool round-trips deterministically. It doesn't: under -race CI it failed even pinned to GOMAXPROCS(1) with GC disabled, so the pool is dropping or missing the item for reasons outside our control (Go's own sync.Pool tests reach for an internal-only pin hook to make this reliable, which isn't available outside the sync package). TestPutDataChunkRejectsOversized stays: since a rejected oversized chunk never reaches dataChunks.Put, whatever getDataChunk returns afterwards is necessarily dataChunkSize-length regardless of how the pool internally schedules retrieval, so that assertion is not subject to the same flake. --- db/etl/etl_test.go | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 88b0804feb4..6595df1c020 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -1634,8 +1634,7 @@ func TestSortableBufferOversizedEntry(t *testing.T) { } // TestSortableBufferResetReleasesChunks: Reset drops the buffer's own chunk -// slice and size bookkeeping so it can be reused immediately. Pool round-tripping -// is TestDataChunkPoolRoundTrip's job. +// slice and size bookkeeping so it can be reused immediately. func TestSortableBufferResetReleasesChunks(t *testing.T) { buf := NewSortableBuffer(256 * datasize.MB) val := bytes.Repeat([]byte{0xEF}, 16*1024) @@ -1655,17 +1654,6 @@ func TestSortableBufferResetReleasesChunks(t *testing.T) { require.Equal(t, []byte("reused"), v) } -// TestDataChunkPoolRoundTrip: a chunk released via putDataChunk comes back on -// the next getDataChunk, instead of a fresh allocation. -func TestDataChunkPoolRoundTrip(t *testing.T) { - c := getDataChunk() - c[0] = 0xAA - putDataChunk(c) - - got := getDataChunk() - require.Equal(t, byte(0xAA), got[0]) -} - // TestPutDataChunkRejectsOversized: an entry's private chunk (bigger than // dataChunkSize) must never enter the shared pool — a later getDataChunk // handing it out under a normal chunk index would corrupt an unrelated buffer. From 9887f23656e5c641d747b7aff841dbd09d7d8ebb Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 26 Aug 2026 14:25:05 +0700 Subject: [PATCH 10/11] db/etl: fold key extraction into the sort comparator pdqsortCmpFunc calls the comparator indirectly, so a separate key closure never inlines there and costs a real call per key, twice per comparison. Inlining it into cmp on n5: Sort/random_100k -14.38%, Sort/random_500k -12.47%, geomean -6.99%. Also fix Size's docstring, which named the one term Size subtracts: the tails of the chunks already filled are counted, the tail of the chunk still filling is not. And say in Prealloc that predictDataSize only sizes the chunk-pointer slice, since the chunks themselves are still taken one at a time. --- db/etl/buffers.go | 29 ++++++++++++++++++----------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 6a9a1722ea4..0571609be2f 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -244,8 +244,8 @@ func (b *sortableBuffer) Put(k, v []byte) { b.entries = append(b.entries, e) } -// Size counts the stored bytes, the tail wasted by the chunk still filling, and -// entryLocSize bytes of metadata per entry. +// Size counts the stored bytes, the tails wasted by the chunks already filled, +// and entryLocSize bytes of metadata per entry. func (b *sortableBuffer) Size() int { return b.chunkBytes - (len(b.cur) - int(b.curOff)) + len(b.entries)*entryLocSize } @@ -278,6 +278,9 @@ func (b *sortableBuffer) Get(i int) ([]byte, []byte) { return key, val } +// Prealloc sizes the entries slice. predictDataSize only reserves room in the +// chunks slice for the chunk pointers; the chunks themselves are still taken +// one at a time, which is what keeps an idle buffer from holding its peak. func (b *sortableBuffer) Prealloc(predictKeysAmount, predictDataSize int) Buffer { if cap(b.entries) < predictKeysAmount { b.entries = make([]entryLoc, 0, predictKeysAmount) @@ -301,18 +304,22 @@ func (b *sortableBuffer) Reset() { func (b *sortableBuffer) SizeLimit() int { return b.optimalSize } func (b *sortableBuffer) Sort() { chunks := b.chunks - key := func(e entryLoc) []byte { - if e.keyLen <= 0 { - return nil + // Key extraction stays inside cmp: pdqsortCmpFunc calls the comparator + // indirectly, so a separate closure never inlines and costs a call per key. + cmp := func(x, y entryLoc) int { + var xk, yk []byte + if x.keyLen > 0 { + off := x.offset & (dataChunkSize - 1) + xk = chunks[x.offset>>dataChunkBits][off : off+x.keyLen] } - off := e.offset & (dataChunkSize - 1) - return chunks[e.offset>>dataChunkBits][off : off+e.keyLen] - } - cmp := func(a, b entryLoc) int { - if c := bytes.Compare(key(a), key(b)); c != 0 { + if y.keyLen > 0 { + off := y.offset & (dataChunkSize - 1) + yk = chunks[y.offset>>dataChunkBits][off : off+y.keyLen] + } + if c := bytes.Compare(xk, yk); c != 0 { return c } - return int(a.insertionOrder - b.insertionOrder) // StableSort: preserve insertion order for duplicate keys + return int(x.insertionOrder - y.insertionOrder) // StableSort: preserve insertion order for duplicate keys } if slices.IsSortedFunc(b.entries, cmp) { return From 6d66504200fc18f6edbf7489cc0938ef0433a1c7 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 26 Aug 2026 17:13:12 +0700 Subject: [PATCH 11/11] db/etl: cover cross-chunk sort, make Dispose join before it reads p.file Review follow-ups: correct the combined ETL budget to 544Mb, test the sort comparator past chunk 0, assert the pool-eligibility decision instead of pool internals, and wait for the async flush before Dispose returns on a nil file. --- db/etl/buffers.go | 9 +++++++-- db/etl/dataprovider.go | 5 +++-- db/etl/etl_test.go | 44 +++++++++++++++++++++++++++++++++++++----- 3 files changed, 49 insertions(+), 9 deletions(-) diff --git a/db/etl/buffers.go b/db/etl/buffers.go index 0571609be2f..80487604ff5 100644 --- a/db/etl/buffers.go +++ b/db/etl/buffers.go @@ -81,7 +81,7 @@ var BufferOptimalSize = dbg.EnvDataSize("ETL_OPTIMAL", 256*datasize.MB) /* var // etlSmallBufRAM (BufferOptimalSize/8) bounds the flush threshold: // 3_domains * 2 + 3_history * 1 + 4_indices * 2 = 17 etl collectors, -// 17*(256Mb/8) = 512Mb for all collectors combined. Buffers pool their +// 17*(256Mb/8) = 544Mb for all collectors combined. Buffers pool their // chunks — see dataChunks below. var ( etlSmallBufRAM = dbg.EnvDataSize("ETL_SMALL", BufferOptimalSize/8) @@ -126,8 +126,13 @@ var dataChunks = sync.Pool{New: func() any { func getDataChunk() []byte { return *dataChunks.Get().(*[]byte) } +// isPooledChunk reports whether c came from the pool. An oversized entry gets a +// private chunk instead, and handing that one back would let a later +// getDataChunk give an unrelated buffer a chunk of the wrong size. +func isPooledChunk(c []byte) bool { return len(c) == dataChunkSize } + func putDataChunk(c []byte) { - if len(c) != dataChunkSize { // private chunk of an oversized entry + if !isPooledChunk(c) { return } dataChunks.Put(&c) diff --git a/db/etl/dataprovider.go b/db/etl/dataprovider.go index c4a35b2fc14..2dd50763782 100644 --- a/db/etl/dataprovider.go +++ b/db/etl/dataprovider.go @@ -189,12 +189,13 @@ func readField(m *mmapBytesReader) ([]byte, error) { func (p *fileDataProvider) Wait() error { return p.wg.Wait() } func (p *fileDataProvider) Dispose() { + // Wait first: the async flush assigns p.file from its own goroutine, so + // reading it before joining both races and can leak a file created after. + p.Wait() if p.file == nil { return } - p.Wait() - if p.mmapData != nil { _ = p.mmapData.Unmap() p.mmapData = nil diff --git a/db/etl/etl_test.go b/db/etl/etl_test.go index 6595df1c020..07c8844e282 100644 --- a/db/etl/etl_test.go +++ b/db/etl/etl_test.go @@ -1602,6 +1602,32 @@ func TestSortableBufferChunks(t *testing.T) { } } +// TestSortableBufferSortAcrossChunks: the sort comparator has to split a +// packed offset back into a chunk index and an offset inside it, so entries +// must still order correctly once they live past chunk 0. +func TestSortableBufferSortAcrossChunks(t *testing.T) { + buf := NewSortableBuffer(256 * datasize.MB) + + const entries = 512 + val := bytes.Repeat([]byte{0xCD}, 16*1024) // 512*16KB = 8MB of values + key := make([]byte, 8) + for i := range entries { + // Scrambled, so IsSortedFunc cannot short-circuit and pdqsort really runs. + // 313 is odd, so it permutes a power-of-two range. + binary.BigEndian.PutUint64(key, uint64(i*313%entries)) + buf.Put(key, val) + } + require.Greater(t, len(buf.chunks), 1, "data must be split into chunks") + + buf.Sort() + for i := range entries { + binary.BigEndian.PutUint64(key, uint64(i)) + k, v := buf.Get(i) + require.Equal(t, key, k, "entry %d", i) + require.Equal(t, val, v, "entry %d", i) + } +} + // TestSortableBufferOversizedEntry: an entry bigger than one chunk gets a chunk // of its own - Get must still return one contiguous slice per key and value. func TestSortableBufferOversizedEntry(t *testing.T) { @@ -1658,11 +1684,19 @@ func TestSortableBufferResetReleasesChunks(t *testing.T) { // dataChunkSize) must never enter the shared pool — a later getDataChunk // handing it out under a normal chunk index would corrupt an unrelated buffer. func TestPutDataChunkRejectsOversized(t *testing.T) { - oversized := make([]byte, dataChunkSize+7) - putDataChunk(oversized) - - got := getDataChunk() - require.Len(t, got, dataChunkSize) + for _, tc := range []struct { + name string + length int + pooled bool + }{ + {"short", dataChunkSize - 1, false}, + {"exact", dataChunkSize, true}, + {"oversized", dataChunkSize + 7, false}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.pooled, isPooledChunk(make([]byte, tc.length))) + }) + } } // disposeProbe records whether the collector still owned its data chunks when