diff --git a/runtime/op/distinct.go b/runtime/op/distinct.go index 94994e09d..5465522d6 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 fcf73c8da..2ce6aa20d 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 000000000..4d6869b0a --- /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 000000000..13577fde4 --- /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 000000000..c810011a4 --- /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 000000000..4bf496894 --- /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 000000000..9484828a1 --- /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 000000000..3336f6164 --- /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 000000000..c1c4caeb9 --- /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 000000000..7bf8131ae --- /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 000000000..3eb143d06 --- /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 +}