zzg
6cfb5b0319
A shard file is a sequence of length-prefixed CRC'd chunks, each carrying its own dictionary and complete column data. Nothing already written is ever revised, so appending is a write at the end of the file and a torn tail costs exactly the chunk that was in flight. Large text goes to a per-shard .blob sidecar so scans touch timestamps and dictionary codes rather than megabytes of prose. Tests cover round-trip, encoding determinism under Go's randomised map iteration, truncation at every byte offset, random bit flips, torn-tail and bad-CRC recovery, midnight-straddling flushes, and both retention policies.
101 lines
2.7 KiB
Go
101 lines
2.7 KiB
Go
package store
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
"io"
|
|
)
|
|
|
|
// Dict is a string dictionary local to a single chunk.
|
|
//
|
|
// Scoping it to the chunk rather than the shard or the store is the whole
|
|
// trick: a chunk is written once and never revised, so its dictionary is
|
|
// finished the moment it is flushed. There is no shared dictionary to lock, to
|
|
// grow, or to merge — the problem Scuba solved with per-leaf dictionaries,
|
|
// which on one machine simply does not arise. The cost is that a string
|
|
// repeated across chunks is stored once per chunk, which for the strings that
|
|
// actually repeat (span names, service names, attribute keys) is a handful of
|
|
// bytes per flush.
|
|
type Dict struct {
|
|
strs []string
|
|
idx map[string]uint32
|
|
}
|
|
|
|
func NewDict() *Dict {
|
|
return &Dict{idx: make(map[string]uint32)}
|
|
}
|
|
|
|
// Put interns s and returns its code.
|
|
func (d *Dict) Put(s string) uint32 {
|
|
if c, ok := d.idx[s]; ok {
|
|
return c
|
|
}
|
|
c := uint32(len(d.strs))
|
|
d.strs = append(d.strs, s)
|
|
d.idx[s] = c
|
|
return c
|
|
}
|
|
|
|
// Get resolves a code. An out-of-range code means a corrupt chunk; returning
|
|
// empty rather than panicking keeps one bad chunk from taking down a query that
|
|
// spans a hundred good ones.
|
|
func (d *Dict) Get(c uint32) string {
|
|
if int(c) >= len(d.strs) {
|
|
return ""
|
|
}
|
|
return d.strs[c]
|
|
}
|
|
|
|
// Lookup finds an existing code without interning. Queries use this to turn a
|
|
// literal into a code once per chunk, after which comparison is integer work.
|
|
// A miss means no row in this chunk can match, which lets the scanner skip the
|
|
// whole chunk for an equality predicate.
|
|
func (d *Dict) Lookup(s string) (uint32, bool) {
|
|
c, ok := d.idx[s]
|
|
return c, ok
|
|
}
|
|
|
|
func (d *Dict) Len() int { return len(d.strs) }
|
|
|
|
func (d *Dict) encode(w io.Writer) error {
|
|
var buf [binary.MaxVarintLen64]byte
|
|
n := binary.PutUvarint(buf[:], uint64(len(d.strs)))
|
|
if _, err := w.Write(buf[:n]); err != nil {
|
|
return err
|
|
}
|
|
for _, s := range d.strs {
|
|
n := binary.PutUvarint(buf[:], uint64(len(s)))
|
|
if _, err := w.Write(buf[:n]); err != nil {
|
|
return err
|
|
}
|
|
if _, err := io.WriteString(w, s); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func decodeDict(r *reader) (*Dict, error) {
|
|
n, err := r.uvarint()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if n > maxDictEntries {
|
|
return nil, fmt.Errorf("dictionary claims %d entries, over the %d cap", n, maxDictEntries)
|
|
}
|
|
d := &Dict{strs: make([]string, 0, n), idx: make(map[string]uint32, n)}
|
|
for i := uint64(0); i < n; i++ {
|
|
s, err := r.str()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
d.strs = append(d.strs, s)
|
|
d.idx[s] = uint32(i)
|
|
}
|
|
return d, nil
|
|
}
|
|
|
|
// maxDictEntries bounds allocation from a corrupt length prefix. A chunk holds
|
|
// at most a few thousand rows, so a dictionary this large is already nonsense.
|
|
const maxDictEntries = 1 << 22
|