Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 56 additions & 2 deletions streamvbyte_coder.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,56 @@ func init() {
}
}

// decodeStreamVByteFull decodes exactly len(dst) values from control/data.
// The SIMD kernel needs a full 16-byte load window per group; on groups
// without one its tail paths are unreliable (observed on linux/arm64:
// zero-filled output with decoded/dataConsumed still reporting success), so
// it is only ever given the provably safe group prefix — full-window groups
// writing 4 whole dst lanes. The remainder is decoded scalar here, and if
// the kernel deviates from the plan everything is redone scalar.
func decodeStreamVByteFull(control, data []byte, dst []uint32) error {
if len(dst) == 0 {
return nil
}

group, dataPos := 0, 0
for group < len(control) && dataPos+16 <= len(data) && (group+1)*4 <= len(dst) {
dataPos += int(streamVByteControlByteLengths[control[group]])
group++
}
if group > 0 {
_, consumed := varint.DecodeStreamVByte32Into(control[:group], data, dst[:group*4])
if consumed != dataPos {
group, dataPos = 0, 0
}
}

dstPos := group * 4
if dstPos >= len(dst) {
return nil
}
for ; group < len(control) && dstPos < len(dst); group++ {
ctrl := control[group]
for lane := 0; lane < 4 && dstPos < len(dst); lane++ {
size := int((ctrl>>(uint(lane)*2))&0x03) + 1
if dataPos+size > len(data) {
return fmt.Errorf("StreamVByte: truncated data at value %d of %d", dstPos, len(dst))
}
var v uint32
for b := size - 1; b >= 0; b-- {
v = v<<8 | uint32(data[dataPos+b])
}
dst[dstPos] = v
dstPos++
dataPos += size
}
}
if dstPos < len(dst) {
return fmt.Errorf("StreamVByte: decoded %d of %d values", dstPos, len(dst))
}
return nil
}

// SeparatedLocFormatMarker is a 2-byte marker indicating separated field ID encoding.
// The sequence 0xFF 0x00 can't be a valid varint because:
// - 0xFF means continuation (value=127, more bytes follow)
Expand Down Expand Up @@ -649,7 +699,9 @@ func (d *streamVByteChunkedIntDecoder) loadChunk(chunk int) error {
} else {
d.values = make([]uint32, numValues)
}
varint.DecodeStreamVByte32Into(d.control, dataBytes, d.values)
if err := decodeStreamVByteFull(d.control, dataBytes, d.values); err != nil {
return err
}

// Apply delta decoding if this chunk was delta encoded
if d.format == ChunkFormatStreamVByteDelta && len(d.values) > 1 {
Expand Down Expand Up @@ -732,7 +784,9 @@ func (d *streamVByteChunkedIntDecoder) loadChunkColumnar() error {
} else {
buf = make([]uint32, numValues)
}
varint.DecodeStreamVByte32Into(ctrl, dataBytes, buf)
if err := decodeStreamVByteFull(ctrl, dataBytes, buf); err != nil {
return nil, err
}
return buf, nil
}

Expand Down
41 changes: 41 additions & 0 deletions streamvbyte_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -430,3 +430,44 @@ func TestMergeFormatUpgrade(t *testing.T) {

t.Logf("Format upgrade test passed: merged segment has %d docs with readable locations", merged.Count())
}

// TestDecodeStreamVByteFullTailCompletion exercises the ≤15-byte data tails
// the SIMD batch kernel leaves undecoded (it stops at the last full 16-byte
// load window). Sizes cover: below one SIMD window, exact group boundaries,
// maximal tails, and multi-window inputs; widths cover 1-4 byte values.
func TestDecodeStreamVByteFullTailCompletion(t *testing.T) {
widths := map[string]func(i int) uint32{
"1byte": func(i int) uint32 { return uint32(i % 250) },
"2byte": func(i int) uint32 { return uint32(260 + i) },
"4byte": func(i int) uint32 { return uint32(0x01000000 + i*7919) },
"mixed": func(i int) uint32 { return uint32(i) << (uint(i%4) * 8) },
}
for name, gen := range widths {
for _, n := range []int{1, 3, 4, 5, 15, 16, 17, 31, 32, 33, 100, 1000} {
values := make([]uint32, n)
for i := range values {
values[i] = gen(i)
}
control, data := varint.EncodeStreamVByte32Into(values, nil, nil)
got := make([]uint32, n)
if err := decodeStreamVByteFull(control, data, got); err != nil {
t.Fatalf("%s/n=%d: decode error: %v", name, n, err)
}
for i := range values {
if got[i] != values[i] {
t.Fatalf("%s/n=%d: value %d mismatch: got %d, want %d", name, n, i, got[i], values[i])
}
}
}
}

// Truncated data must error, not silently zero-fill.
values := []uint32{300, 300, 300, 300, 300}
control, data := varint.EncodeStreamVByte32Into(values, nil, nil)
got := make([]uint32, len(values))
// last group = one 2-byte value + three 1-byte zero pads; cut past the
// padding into the value's bytes so real data is actually missing
if err := decodeStreamVByteFull(control, data[:len(data)-4], got); err == nil {
t.Fatal("expected error for truncated data, got nil")
}
}
Loading