From caf2dcddd4b270001f08a61b268f86ee1317e424 Mon Sep 17 00:00:00 2001 From: Matthew Nibecker Date: Wed, 4 Nov 2026 13:00:41 -0800 Subject: [PATCH] vector/vhash: add vectorized hash table and use it for distinct Add vector/vhash, a vectorized hash table that maps tuples of key vectors to dense key IDs. It uses open addressing with linear probing, where each slot packs a 32-bit hash salt with a key ID. Keys are kept in vector format so they can be returned directly as vectors. It is meant to be shared by distinct, hash join, and aggregations. Switch the distinct operator from a map of serialized values to vhash, keeping one table per key type. --- runtime/op/distinct.go | 72 +++++++--- runtime/ztests/op/distinct.yaml | 101 ++++++++------ vector/vhash/bytes.go | 109 +++++++++++++++ vector/vhash/comparable.go | 119 ++++++++++++++++ vector/vhash/hash.go | 26 ++++ vector/vhash/key.go | 67 +++++++++ vector/vhash/named.go | 19 +++ vector/vhash/numeric.go | 124 +++++++++++++++++ vector/vhash/scode.go | 72 ++++++++++ vector/vhash/table.go | 237 ++++++++++++++++++++++++++++++++ vector/vhash/table_test.go | 111 +++++++++++++++ 11 files changed, 998 insertions(+), 59 deletions(-) create mode 100644 vector/vhash/bytes.go create mode 100644 vector/vhash/comparable.go create mode 100644 vector/vhash/hash.go create mode 100644 vector/vhash/key.go create mode 100644 vector/vhash/named.go create mode 100644 vector/vhash/numeric.go create mode 100644 vector/vhash/scode.go create mode 100644 vector/vhash/table.go create mode 100644 vector/vhash/table_test.go diff --git a/runtime/op/distinct.go b/runtime/op/distinct.go index 94994e09da..5465522d6a 100644 --- a/runtime/op/distinct.go +++ b/runtime/op/distinct.go @@ -1,12 +1,12 @@ package op import ( - "encoding/binary" + "slices" "github.com/superdb/super" "github.com/superdb/super/runtime/expr" - "github.com/superdb/super/scode" "github.com/superdb/super/vector" + "github.com/superdb/super/vector/vhash" "github.com/superdb/super/vector/vio" ) @@ -14,37 +14,71 @@ type Distinct struct { sctx *super.Context parent vio.Puller expr expr.Evaluator - - blocked map[string]struct{} - defuse *expr.SuperDefuse - key []byte + defuse *expr.SuperDefuse + tables map[super.Type]*vhash.Table + // scratch + index []uint32 } func NewDistinct(sctx *super.Context, parent vio.Puller, e expr.Evaluator) *Distinct { - return &Distinct{sctx, parent, e, map[string]struct{}{}, expr.NewSuperDefuse(sctx), nil} + return &Distinct{ + sctx: sctx, + parent: parent, + expr: e, + defuse: expr.NewSuperDefuse(sctx), + tables: map[super.Type]*vhash.Table{}, + } } func (d *Distinct) Pull(done bool) (vector.Any, error) { for { vec, err := d.parent.Pull(done) if vec == nil || err != nil { - clear(d.blocked) + clear(d.tables) return nil, err } - var sb scode.Builder - var index []uint32 keyVec := d.defuse.Eval(d.expr.Eval(vec)) - for i := range keyVec.Len() { - keyVal := vector.ValueAt(&sb, keyVec, i) - d.key = binary.LittleEndian.AppendUint32(d.key[:0], uint32(keyVal.Type().ID())) - d.key = append(d.key, keyVal.Bytes()...) - if _, ok := d.blocked[string(d.key)]; !ok { - d.blocked[string(d.key)] = struct{}{} - index = append(index, i) + d.index = d.index[:0] + d.insert(keyVec, nil) + if len(d.index) > 0 { + // vhash reports new keys in creation order, not row order + // (salt collisions defer rows to a later pass), and Dynamic + // keys are inserted per type. + // XXX There are use cases (e.g., distinct before an aggregation) + // where we don't care about maintaining order so we should have a + // way to skip this when not needed. + if !slices.IsSorted(d.index) { + slices.Sort(d.index) + } + return vector.Pick(vec, d.index), nil + } + } +} + +func (d *Distinct) insert(vec vector.Any, rows []uint32) { + if dynamic, ok := vec.(*vector.Dynamic); ok { + for tag, slots := range dynamic.ReverseTagMap() { + if len(slots) == 0 { + continue } + d.insert(dynamic.Values[tag], slots) } - if len(index) > 0 { - return vector.Pick(vec, index), nil + return + } + if vec.Len() == 0 { + return + } + typ := vec.Type() + table, ok := d.tables[typ] + if !ok { + table = vhash.NewTable([]super.Type{typ}) + d.tables[typ] = table + } + _, created := table.FindOrInsert([]vector.Any{vec}) + for _, slot := range created { + if rows != nil { + slot = rows[slot] } + d.index = append(d.index, slot) } } diff --git a/runtime/ztests/op/distinct.yaml b/runtime/ztests/op/distinct.yaml index fcf73c8da7..2ce6aa20d4 100644 --- a/runtime/ztests/op/distinct.yaml +++ b/runtime/ztests/op/distinct.yaml @@ -1,40 +1,61 @@ -script: | - super -s -c 'from "1.sup" | distinct x.ok()' - echo === - super -s -c 'from "2.sup" | distinct abs(this)' - echo === - super -s -c 'from "3.sup" | unnest this into (distinct this | count())' - -inputs: - - name: 1.sup - data: | - {x:1} - {x:2} - {x:2} - {y:3} - type named={y:int64} - {y:3}::named - - name: 2.sup - data: | - 1 - -2 - 2 - 3 - - name: 3.sup - data: | - [1,2,2,3] - [2,2,3] - -outputs: - - name: stdout - data: | - {x:1} - {x:2} - {y:3} - === - 1 - -2 - 3 - === - 3 - 2 +spq: distinct x.ok() + +input: | + {x:1} + {x:2} + {x:2} + {y:3} + type named={y:int64} + {y:3}::named + +output: | + {x:1} + {x:2} + {y:3} + +--- + +spq: distinct abs(this) + +input: | + 1 + -2 + 2 + 3 + +output: | + 1 + -2 + 3 + +--- + +spq: unnest this into (distinct this | count()) + +input: | + [1,2,2,3] + [2,2,3] + +output: | + 3 + 2 + +--- + +# NaN equals NaN and -0 differs from +0. SUP formats -0 as 0., so 1./x +# shows which zero each row is. +spq: distinct x | values {x, inv:1./x} + +input: | + {x:NaN} + {x:0.} + {x:-0.} + {x:NaN} + {x:1.5} + {x:0.} + +output: | + {x:NaN,inv:NaN} + {x:0.,inv:+Inf} + {x:0.,inv:-Inf} + {x:1.5,inv:0.6666666666666666} diff --git a/vector/vhash/bytes.go b/vector/vhash/bytes.go new file mode 100644 index 0000000000..4d6869b0a2 --- /dev/null +++ b/vector/vhash/bytes.go @@ -0,0 +1,109 @@ +package vhash + +import ( + "bytes" + "hash/maphash" + + "github.com/superdb/super" + "github.com/superdb/super/vector" +) + +type bytesKey struct { + typ super.Type + vals vector.BytesTable + + // Bound input. + in vector.BytesTable + index []uint32 + scratch []uint32 + baseH []uint64 +} + +func newBytesKey(typ super.Type) *bytesKey { + return &bytesKey{typ: typ, vals: vector.NewBytesTableEmpty(0)} +} + +func (k *bytesKey) bind(vec vector.Any) { + var base vector.Any + base, k.index, k.scratch = resolveVec(vec, k.scratch) + switch base := base.(type) { + case *vector.String: + k.in = base.Table() + case *vector.Bytes: + k.in = base.Table() + default: + panic(base) + } +} + +func (k *bytesKey) hash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = k.baseH[slot] + } + case k.index == nil: + for i := range k.in.Len() { + hashes[i] = maphash.Bytes(seed, k.in.Bytes(i)) + } + default: + for i, slot := range k.index { + hashes[i] = maphash.Bytes(seed, k.in.Bytes(slot)) + } + } +} + +func (k *bytesKey) combineHash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = combine(hashes[i], k.baseH[slot]) + } + case k.index == nil: + for i := range k.in.Len() { + hashes[i] = combine(hashes[i], maphash.Bytes(seed, k.in.Bytes(i))) + } + default: + for i, slot := range k.index { + hashes[i] = combine(hashes[i], maphash.Bytes(seed, k.in.Bytes(slot))) + } + } +} + +func (k *bytesKey) hashBase() bool { + n := k.in.Len() + if k.index == nil || int(n) >= len(k.index) { + return false + } + k.baseH = resize(k.baseH, int(n)) + for i := range n { + k.baseH[i] = maphash.Bytes(seed, k.in.Bytes(i)) + } + return true +} + +func (k *bytesKey) equal(rows, keyIDs []uint32, eq []bool) { + for i, row := range rows { + eq[i] = eq[i] && bytes.Equal(k.at(row), k.vals.Bytes(keyIDs[i])) + } +} + +func (k *bytesKey) append(rows []uint32) { + for _, row := range rows { + k.vals.Append(k.at(row)) + } +} + +func (k *bytesKey) at(row uint32) []byte { + if k.index != nil { + row = k.index[row] + } + return k.in.Bytes(row) +} + +func (k *bytesKey) vector(*super.Context) vector.Any { + if k.typ.ID() == super.IDString { + return vector.NewString(k.vals) + } + return vector.NewBytes(k.vals) +} diff --git a/vector/vhash/comparable.go b/vector/vhash/comparable.go new file mode 100644 index 0000000000..13577fde42 --- /dev/null +++ b/vector/vhash/comparable.go @@ -0,0 +1,119 @@ +package vhash + +import ( + "hash/maphash" + "net/netip" + "uuid" + + "github.com/superdb/super" + "github.com/superdb/super/vector" +) + +// comparableKey stores fixed-width values of a Go comparable type (UUIDs, +// IP addresses, and networks) and hashes them with maphash.Comparable. +type comparableKey[E comparable] struct { + vals []E + valuesOf func(vector.Any) []E + build func([]E) vector.Any + + // Bound input. + in []E + index []uint32 + scratch []uint32 + baseH []uint64 +} + +func newUUIDKey() *comparableKey[uuid.UUID] { + return &comparableKey[uuid.UUID]{ + valuesOf: func(vec vector.Any) []uuid.UUID { return vec.(*vector.UUID).Values }, + build: func(vals []uuid.UUID) vector.Any { return vector.NewUUID(vals) }, + } +} + +func newIPKey() *comparableKey[netip.Addr] { + return &comparableKey[netip.Addr]{ + valuesOf: func(vec vector.Any) []netip.Addr { return vec.(*vector.IP).Values }, + build: func(vals []netip.Addr) vector.Any { return vector.NewIP(vals) }, + } +} + +func newNetKey() *comparableKey[netip.Prefix] { + return &comparableKey[netip.Prefix]{ + valuesOf: func(vec vector.Any) []netip.Prefix { return vec.(*vector.Net).Values }, + build: func(vals []netip.Prefix) vector.Any { return vector.NewNet(vals) }, + } +} + +func (k *comparableKey[E]) bind(vec vector.Any) { + var base vector.Any + base, k.index, k.scratch = resolveVec(vec, k.scratch) + k.in = k.valuesOf(base) +} + +func (k *comparableKey[E]) at(row uint32) E { + if k.index != nil { + row = k.index[row] + } + return k.in[row] +} + +func (k *comparableKey[E]) hash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = k.baseH[slot] + } + case k.index == nil: + for i, v := range k.in { + hashes[i] = maphash.Comparable(seed, v) + } + default: + for i, slot := range k.index { + hashes[i] = maphash.Comparable(seed, k.in[slot]) + } + } +} + +func (k *comparableKey[E]) combineHash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = combine(hashes[i], k.baseH[slot]) + } + case k.index == nil: + for i, v := range k.in { + hashes[i] = combine(hashes[i], maphash.Comparable(seed, v)) + } + default: + for i, slot := range k.index { + hashes[i] = combine(hashes[i], maphash.Comparable(seed, k.in[slot])) + } + } +} + +func (k *comparableKey[E]) hashBase() bool { + if k.index == nil || len(k.in) >= len(k.index) { + return false + } + k.baseH = resize(k.baseH, len(k.in)) + for i, v := range k.in { + k.baseH[i] = maphash.Comparable(seed, v) + } + return true +} + +func (k *comparableKey[E]) equal(rows, keyIDs []uint32, eq []bool) { + for i, row := range rows { + eq[i] = eq[i] && k.at(row) == k.vals[keyIDs[i]] + } +} + +func (k *comparableKey[E]) append(rows []uint32) { + for _, row := range rows { + k.vals = append(k.vals, k.at(row)) + } +} + +func (k *comparableKey[E]) vector(*super.Context) vector.Any { + return k.build(k.vals) +} diff --git a/vector/vhash/hash.go b/vector/vhash/hash.go new file mode 100644 index 0000000000..c810011a4f --- /dev/null +++ b/vector/vhash/hash.go @@ -0,0 +1,26 @@ +package vhash + +import "hash/maphash" + +var seed = maphash.MakeSeed() + +// fmix64 is the 64-bit finalizer from MurmurHash3. It is used to hash +// fixed-width integer values so that the low bits (used for the slot offset) +// and the high bits (used for the salt) are both well distributed. +func fmix64(h uint64) uint64 { + h ^= h >> 33 + h *= 0xff51afd7ed558ccd + h ^= h >> 33 + h *= 0xc4ceb9fe1a85ec53 + h ^= h >> 33 + return h +} + +// combine folds the hash of the next key position into the running hash h. +// h is scrambled before the XOR so the result depends on key order, and next +// is already a well-mixed hash, so no further finalization is needed. +func combine(h, next uint64) uint64 { + h ^= h >> 32 + h *= 0xd6e8feb86659fd93 + return h ^ next +} diff --git a/vector/vhash/key.go b/vector/vhash/key.go new file mode 100644 index 0000000000..4bf4968947 --- /dev/null +++ b/vector/vhash/key.go @@ -0,0 +1,67 @@ +package vhash + +import ( + "github.com/superdb/super" + "github.com/superdb/super/vector" +) + +type key interface { + // bind attaches vec as the current input batch. + bind(vec vector.Any) + // hash writes the hash of each input row into hashes. + hash(hashes []uint64) + // combineHash folds the hash of each input row into hashes, which + // already holds the hashes of the preceding key positions. + combineHash(hashes []uint64) + // equal clears eq[i] when input row rows[i] differs from stored key keyIDs[i]. + equal(rows, keyIDs []uint32, eq []bool) + // append stores input rows as new keys. + append(rows []uint32) + vector(sctx *super.Context) vector.Any +} + +func newKey(typ super.Type) key { + switch typ := typ.(type) { + case *super.TypeNamed: + return newNamedKey(typ) + case *super.TypeOfInt8, *super.TypeOfInt16, *super.TypeOfInt32, *super.TypeOfInt64, + *super.TypeOfDuration, *super.TypeOfTimestamp: + return newIntKey(typ) + case *super.TypeOfUint8, *super.TypeOfUint16, *super.TypeOfUint32, *super.TypeOfUint64: + return newUintKey(typ) + case *super.TypeOfFloat16, *super.TypeOfFloat32, *super.TypeOfFloat64: + return newFloatKey(typ) + case *super.TypeOfUUID: + return newUUIDKey() + case *super.TypeOfIP: + return newIPKey() + case *super.TypeOfNet: + return newNetKey() + case *super.TypeOfString, *super.TypeOfBytes: + return newBytesKey(typ) + default: + // Everything else (bool, null, type values, records, arrays, maps, + // unions, enums, errors, options, ...) is keyed by its serialized + // form for now. + return newScodeKey(typ) + } +} + +func resolveVec(vec vector.Any, scratch []uint32) (vector.Any, []uint32, []uint32) { + switch v := vector.Under(vec).(type) { + case *vector.View: + return v.Any, v.Index, scratch + case *vector.Dict: + scratch = resize(scratch, len(v.Index)) + for i, slot := range v.Index { + scratch[i] = uint32(slot) + } + return v.Any, scratch, scratch + case *vector.Const: + scratch = resize(scratch, int(v.Len())) + clear(scratch) + return v.Any, scratch, scratch + default: + return v, nil, scratch + } +} diff --git a/vector/vhash/named.go b/vector/vhash/named.go new file mode 100644 index 0000000000..9484828a15 --- /dev/null +++ b/vector/vhash/named.go @@ -0,0 +1,19 @@ +package vhash + +import ( + "github.com/superdb/super" + "github.com/superdb/super/vector" +) + +type namedKey struct { + typ *super.TypeNamed + key +} + +func newNamedKey(typ *super.TypeNamed) *namedKey { + return &namedKey{typ: typ, key: newKey(typ.Type)} +} + +func (k *namedKey) vector(sctx *super.Context) vector.Any { + return vector.NewNamed(k.typ, k.key.vector(sctx)) +} diff --git a/vector/vhash/numeric.go b/vector/vhash/numeric.go new file mode 100644 index 0000000000..3336f61644 --- /dev/null +++ b/vector/vhash/numeric.go @@ -0,0 +1,124 @@ +package vhash + +import ( + "github.com/superdb/super" + "github.com/superdb/super/pkg/byteconv" + "github.com/superdb/super/vector" +) + +type numericKey struct { + vals []uint64 + valuesOf func(vector.Any) []uint64 + build func([]uint64) vector.Any + + // Bound input. + in []uint64 + index []uint32 + scratch []uint32 + baseH []uint64 +} + +func newIntKey(typ super.Type) *numericKey { + return &numericKey{ + valuesOf: func(vec vector.Any) []uint64 { + return byteconv.ReinterpretSlice[uint64](vec.(*vector.Int).Values) + }, + build: func(vals []uint64) vector.Any { + return vector.NewInt(typ, byteconv.ReinterpretSlice[int64](vals)) + }, + } +} + +func newUintKey(typ super.Type) *numericKey { + return &numericKey{ + valuesOf: func(vec vector.Any) []uint64 { return vec.(*vector.Uint).Values }, + build: func(vals []uint64) vector.Any { return vector.NewUint(typ, vals) }, + } +} + +func newFloatKey(typ super.Type) *numericKey { + return &numericKey{ + valuesOf: func(vec vector.Any) []uint64 { + return byteconv.ReinterpretSlice[uint64](vec.(*vector.Float).Values) + }, + build: func(vals []uint64) vector.Any { + return vector.NewFloat(typ, byteconv.ReinterpretSlice[float64](vals)) + }, + } +} + +func (k *numericKey) bind(vec vector.Any) { + var base vector.Any + base, k.index, k.scratch = resolveVec(vec, k.scratch) + k.in = k.valuesOf(base) +} + +func (k *numericKey) hash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = k.baseH[slot] + } + case k.index == nil: + for i, v := range k.in { + hashes[i] = fmix64(v) + } + default: + for i, slot := range k.index { + hashes[i] = fmix64(k.in[slot]) + } + } +} + +func (k *numericKey) combineHash(hashes []uint64) { + switch { + case k.hashBase(): + for i, slot := range k.index { + hashes[i] = combine(hashes[i], k.baseH[slot]) + } + case k.index == nil: + for i, v := range k.in { + hashes[i] = combine(hashes[i], fmix64(v)) + } + default: + for i, slot := range k.index { + hashes[i] = combine(hashes[i], fmix64(k.in[slot])) + } + } +} + +func (k *numericKey) hashBase() bool { + if k.index == nil || len(k.in) >= len(k.index) { + return false + } + k.baseH = resize(k.baseH, len(k.in)) + for i, v := range k.in { + k.baseH[i] = fmix64(v) + } + return true +} + +func (k *numericKey) equal(rows, keyIDs []uint32, eq []bool) { + if k.index != nil { + for i, row := range rows { + eq[i] = eq[i] && k.in[k.index[row]] == k.vals[keyIDs[i]] + } + } else { + for i, row := range rows { + eq[i] = eq[i] && k.in[row] == k.vals[keyIDs[i]] + } + } +} + +func (k *numericKey) append(rows []uint32) { + for _, row := range rows { + if k.index != nil { + row = k.index[row] + } + k.vals = append(k.vals, k.in[row]) + } +} + +func (k *numericKey) vector(*super.Context) vector.Any { + return k.build(k.vals) +} diff --git a/vector/vhash/scode.go b/vector/vhash/scode.go new file mode 100644 index 0000000000..c1c4caeb93 --- /dev/null +++ b/vector/vhash/scode.go @@ -0,0 +1,72 @@ +package vhash + +import ( + "hash/maphash" + + "github.com/superdb/super" + "github.com/superdb/super/scode" + "github.com/superdb/super/vector" +) + +// scodeKey is the fallback for types without specialized storage. Each +// value is stored in its tagged, serialized form (so null and empty bodies +// remain distinct) and compared byte-wise. +type scodeKey struct { + typ super.Type + vals vector.BytesTable + + // Bound input, serialized. + in vector.BytesTable + b scode.Builder +} + +func newScodeKey(typ super.Type) *scodeKey { + return &scodeKey{ + typ: typ, + vals: vector.NewBytesTableEmpty(0), + in: vector.NewBytesTableEmpty(0), + } +} + +func (k *scodeKey) bind(vec vector.Any) { + k.in.Reset() + for i := range vec.Len() { + k.b.Truncate() + vec.Serialize(&k.b, i) + k.in.Append(k.b.Bytes()) + } +} + +func (k *scodeKey) hash(hashes []uint64) { + for i := range k.in.Len() { + hashes[i] = maphash.Bytes(seed, k.in.Bytes(i)) + } +} + +func (k *scodeKey) combineHash(hashes []uint64) { + for i := range k.in.Len() { + hashes[i] = combine(hashes[i], maphash.Bytes(seed, k.in.Bytes(i))) + } +} + +func (k *scodeKey) equal(rows, keyIDs []uint32, eq []bool) { + for i, row := range rows { + if eq[i] && string(k.in.Bytes(row)) != string(k.vals.Bytes(keyIDs[i])) { + eq[i] = false + } + } +} + +func (k *scodeKey) append(rows []uint32) { + for _, row := range rows { + k.vals.Append(k.in.Bytes(row)) + } +} + +func (k *scodeKey) vector(sctx *super.Context) vector.Any { + b := vector.NewValueBuilder(k.typ) + for i := range k.vals.Len() { + b.Write(scode.Bytes(k.vals.Bytes(i)).Body()) + } + return b.Build(sctx) +} diff --git a/vector/vhash/table.go b/vector/vhash/table.go new file mode 100644 index 0000000000..7bf8131ae0 --- /dev/null +++ b/vector/vhash/table.go @@ -0,0 +1,237 @@ +// Package vhash provides a vectorized hash table that maps tuples of key +// values to dense key IDs. It is the shared core for operators that need +// to look up keys in bulk, e.g., distinct, hash join, and aggregation. +package vhash + +import ( + "fmt" + "math" + "slices" + + "github.com/superdb/super" + "github.com/superdb/super/sup" + "github.com/superdb/super/vector" +) + +// Miss is the key ID that Find reports for keys not in the table. +const Miss = math.MaxUint32 + +const ( + saltMask = 0xffffffff << 32 + keyIDMask = 1<<32 - 1 + minSlots = 256 +) + +type Table struct { + types []super.Type + keys []key + slots []uint64 // salt | (keyID+1); 0 means empty + mask uint64 + hashes []uint64 // hash of each stored key, by key ID + + // Per-call scratch. + h []uint64 + offsets []uint64 + keyIDs []uint32 + created []uint32 + remaining []uint32 + newRows []uint32 + candRows []uint32 + candKeyIDs []uint32 + eq []bool +} + +// NewTable returns an empty table for keys of the given types. It panics if +// types is empty. +func NewTable(types []super.Type) *Table { + if len(types) == 0 { + panic("vhash.NewTable: no key types") + } + t := &Table{types: types} + for _, typ := range types { + t.keys = append(t.keys, newKey(typ)) + } + t.resizeTable(minSlots) + return t +} + +// Len returns the number of distinct keys in the table. +func (t *Table) Len() uint32 { + return uint32(len(t.hashes)) +} + +// Types returns the key types of the table. +func (t *Table) Types() []super.Type { + return t.types +} + +// FindOrInsert looks up each row of keys, inserting the keys not already +// present. keys must have one vector per key type, each of the matching type +// and all of the same length. It returns the key ID of each row and the +// rows that inserted new keys. Key IDs are assigned densely in creation +// order, which is not necessarily row order (a row whose probe hits a salt +// collision is resolved in a later pass). The returned slices are owned by t +// and are valid until the next call. +func (t *Table) FindOrInsert(keys []vector.Any) (keyIDs, created []uint32) { + return t.lookup(keys, true) +} + +// Find looks up each row of keys and returns its key ID, or Miss if the key +// is not in the table. The returned slice is owned by t and is valid until +// the next call. +func (t *Table) Find(keys []vector.Any) []uint32 { + keyIDs, _ := t.lookup(keys, false) + return keyIDs +} + +func (t *Table) Keys(sctx *super.Context) []vector.Any { + var out []vector.Any + for _, k := range t.keys { + out = append(out, k.vector(sctx)) + } + return out +} + +func (t *Table) lookup(vecs []vector.Any, insert bool) ([]uint32, []uint32) { + if len(vecs) != len(t.keys) { + panic(fmt.Sprintf("vhash: got %d key vectors, table expects %d", len(vecs), len(t.keys))) + } + n := int(vecs[0].Len()) + for i, vec := range vecs { + if int(vec.Len()) != n { + panic("vhash: key vectors have different lengths") + } + if _, ok := vec.(*vector.Dynamic); ok { + panic(fmt.Sprintf("vhash: key %d is a vector.Dynamic; split it by type first", i)) + } + if typ := vec.Type(); typ != t.types[i] { + panic(fmt.Sprintf("vhash: key %d has type %s, table expects %s", + i, sup.FormatType(typ), sup.FormatType(t.types[i]))) + } + } + t.keyIDs = resize(t.keyIDs, n) + t.created = t.created[:0] + if n == 0 { + return t.keyIDs, t.created + } + for i, k := range t.keys { + k.bind(vecs[i]) + } + t.h = resize(t.h, n) + hashes := t.h + t.keys[0].hash(hashes) + for _, k := range t.keys[1:] { + k.combineHash(hashes) + } + if insert { + t.reserve(uint64(len(t.hashes)) + uint64(n)) + } + t.offsets = resize(t.offsets, n) + for i, h := range hashes { + t.offsets[i] = h & t.mask + } + t.remaining = resize(t.remaining, n) + for i := range t.remaining { + t.remaining[i] = uint32(i) + } + for len(t.remaining) > 0 { + t.newRows = t.newRows[:0] + t.candRows = t.candRows[:0] + t.candKeyIDs = t.candKeyIDs[:0] + // Probe each remaining row until it finds an empty slot or a + // slot with a matching salt. + for _, row := range t.remaining { + h := hashes[row] + salt := h & saltMask + off := t.offsets[row] + for { + e := t.slots[off] + if e == 0 { + if insert { + keyID := uint32(len(t.hashes)) + t.slots[off] = salt | uint64(keyID+1) + t.hashes = append(t.hashes, h) + t.keyIDs[row] = keyID + t.newRows = append(t.newRows, row) + } else { + t.keyIDs[row] = Miss + } + break + } + if e&saltMask == salt { + t.candRows = append(t.candRows, row) + t.candKeyIDs = append(t.candKeyIDs, uint32(e&keyIDMask-1)) + break + } + // Occupied by another key; try the next slot, wrapping at the + // end + off = (off + 1) & t.mask + } + t.offsets[row] = off + } + // Store the new keys before comparing candidates since + // a candidate may be a later duplicate of a new key in + // this same batch. + if len(t.newRows) > 0 { + for _, k := range t.keys { + k.append(t.newRows) + } + t.created = append(t.created, t.newRows...) + } + // Compare candidates' keys. Matches are resolved; mismatches + // (salt collisions) continue probing at the next slot. + t.remaining = t.remaining[:0] + if len(t.candRows) > 0 { + t.eq = resize(t.eq, len(t.candRows)) + for i := range t.eq { + t.eq[i] = true + } + for _, k := range t.keys { + k.equal(t.candRows, t.candKeyIDs, t.eq) + } + for i, row := range t.candRows { + if t.eq[i] { + t.keyIDs[row] = t.candKeyIDs[i] + } else { + t.offsets[row] = (t.offsets[row] + 1) & t.mask + t.remaining = append(t.remaining, row) + } + } + } + } + return t.keyIDs, t.created +} + +// reserve grows the slot array if needed so that it can hold n keys at a +// load factor of at most 3/4. +func (t *Table) reserve(n uint64) { + // Key IDs are uint32 and Miss is reserved, so there can be at most + // Miss keys (IDs 0 through Miss-1). + if n > Miss { + panic("vhash: too many keys") + } + size := uint64(len(t.slots)) + for n*4 > size*3 { + size *= 2 + } + if size != uint64(len(t.slots)) { + t.resizeTable(size) + } +} + +// resize returns s with length n, reusing its capacity when possible. +func resize[E any](s []E, n int) []E { + return slices.Grow(s[:0], n)[:n] +} + +func (t *Table) resizeTable(size uint64) { + t.slots = make([]uint64, size) + t.mask = size - 1 + for keyID, h := range t.hashes { + off := h & t.mask + for t.slots[off] != 0 { + off = (off + 1) & t.mask + } + t.slots[off] = h&saltMask | uint64(keyID+1) + } +} diff --git a/vector/vhash/table_test.go b/vector/vhash/table_test.go new file mode 100644 index 0000000000..3eb143d064 --- /dev/null +++ b/vector/vhash/table_test.go @@ -0,0 +1,111 @@ +package vhash + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/superdb/super" + "github.com/superdb/super/vector" +) + +func TestTableFindOrInsert(t *testing.T) { + tab := NewTable([]super.Type{super.TypeInt64}) + keyIDs, created := tab.FindOrInsert([]vector.Any{ints(5, 7, 5, 9, 7, 5)}) + assert.Equal(t, []uint32{0, 1, 0, 2, 1, 0}, keyIDs) + assert.Equal(t, []uint32{0, 1, 3}, created) + + keyIDs, created = tab.FindOrInsert([]vector.Any{ints(9, 11, 5, 11)}) + assert.Equal(t, []uint32{2, 3, 0, 3}, keyIDs) + assert.Equal(t, []uint32{1}, created) + assert.Equal(t, uint32(4), tab.Len()) + + keys := tab.Keys(super.NewContext()) + assert.Equal(t, []int64{5, 7, 9, 11}, keys[0].(*vector.Int).Values) + + assert.Equal(t, []uint32{3, Miss, 0}, tab.Find([]vector.Any{ints(11, 12, 5)})) + assert.Equal(t, uint32(4), tab.Len()) +} + +// TestTableTypeMismatch checks that keys whose type differs from the +// table's are rejected even when they share a vector representation. +func TestTableTypeMismatch(t *testing.T) { + tab := NewTable([]super.Type{super.TypeInt64}) + assert.Panics(t, func() { + tab.FindOrInsert([]vector.Any{vector.NewInt(super.TypeInt32, []int64{1})}) + }) +} + +// TestTableSaltCollision forces keys that share a home slot and a salt but +// differ, which exercises the mismatch-and-retry path of the probe loop. +func TestTableSaltCollision(t *testing.T) { + const salt = 0xabcd1234 << 32 + k1 := int64(unmix(salt | 5)) // home slot 5 + k2 := int64(unmix(salt | 1<<16 | 5)) // home slot 5, same salt, different key + k3 := int64(unmix(0x11111111<<32 | 50)) + require.Equal(t, uint64(salt|5), fmix64(uint64(k1))) + + tab := NewTable([]super.Type{super.TypeInt64}) + keyIDs, created := tab.FindOrInsert([]vector.Any{ints(k1, k2, k3, k1, k2)}) + // k2 collides with k1 in the first pass and is inserted in the second, + // so it gets a higher key ID than k3 even though it comes first. + assert.Equal(t, []uint32{0, 2, 1, 0, 2}, keyIDs) + assert.Equal(t, []uint32{0, 2, 1}, created) + assert.Equal(t, []uint32{0, 2, 1, Miss}, tab.Find([]vector.Any{ints(k1, k2, k3, int64(unmix(salt|1<<17|5)))})) +} + +// unmix inverts fmix64 so tests can construct int64 keys with chosen hashes. +func unmix(h uint64) uint64 { + // x ^= x >> 33 is its own inverse since 2*33 >= 64. + h ^= h >> 33 + h *= inverse(0xc4ceb9fe1a85ec53) + h ^= h >> 33 + h *= inverse(0xff51afd7ed558ccd) + h ^= h >> 33 + return h +} + +// inverse returns the multiplicative inverse of odd a modulo 2^64. +func inverse(a uint64) uint64 { + x := a // correct to 3 bits; each Newton step doubles that + for range 5 { + x *= 2 - a*x + } + return x +} + +// TestTableResize inserts enough keys across batches to grow the slot array +// several times and checks that every key is still found with its key ID. +func TestTableResize(t *testing.T) { + tab := NewTable([]super.Type{super.TypeInt64}) + var vals []int64 + var keyIDs []uint32 + for b := range 10 { + batch := make([]int64, 1000) + for i := range batch { + batch[i] = int64(b*1000 + i) + } + ids, created := tab.FindOrInsert([]vector.Any{ints(batch...)}) + require.Len(t, created, 1000) + vals = append(vals, batch...) + keyIDs = append(keyIDs, ids...) + } + require.Equal(t, uint32(len(vals)), tab.Len()) + assert.Equal(t, keyIDs, tab.Find([]vector.Any{ints(vals...)})) + keys := tab.Keys(super.NewContext()) + for i, keyID := range keyIDs { + require.Equal(t, vals[i], vector.IntValue(keys[0], keyID)) + } +} + +func ints(vals ...int64) vector.Any { + return vector.NewInt(super.TypeInt64, vals) +} + +func strs(vals ...string) vector.Any { + b := vector.NewStringEmpty(0) + for _, v := range vals { + b.Append(v) + } + return b +}