vector/vhash: add vectorized hash table and use it for distinct

runtime/op/distinct.go

@@ -1,50 +1,84 @@
 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"
 )
 
 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)
 	}
 }

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}

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)
+}