perf(core): Add a count min sketch for eq filter (#9218)

Author: Harshil Goel

algo/cm-sketch.go

@@ -0,0 +1,318 @@
+/*
+ * Copyright 2016-2025 Hypermode Inc. and Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package algo
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"hash"
+	"hash/fnv"
+	"io"
+	"math"
+)
+
+// This code is copied from https://www.github.com/tylertreat/BoomFilters/refs/heads/master/countmin.go
+// CountMinSketch implements a Count-Min Sketch as described by Cormode and
+// Muthukrishnan in An Improved Data Stream Summary: The Count-Min Sketch and
+// its Applications:
+//
+// http://dimacs.rutgers.edu/~graham/pubs/papers/cm-full.pdf
+//
+// A Count-Min Sketch (CMS) is a probabilistic data structure which
+// approximates the frequency of events in a data stream. Unlike a hash map, a
+// CMS uses sub-linear space at the expense of a configurable error factor.
+// Similar to Counting Bloom filters, items are hashed to a series of buckets,
+// which increment a counter. The frequency of an item is estimated by taking
+// the minimum of each of the item's respective counter values.
+//
+// Count-Min Sketches are useful for counting the frequency of events in
+// massive data sets or unbounded streams online. In these situations, storing
+// the entire data set or allocating counters for every event in memory is
+// impractical. It may be possible for offline processing, but real-time
+// processing requires fast, space-efficient solutions like the CMS. For
+// approximating set cardinality, refer to the HyperLogLog.
+type CountMinSketch struct {
+	matrix  [][]uint64  // count matrix
+	width   uint        // matrix width
+	depth   uint        // matrix depth
+	count   uint64      // number of items added
+	epsilon float64     // relative-accuracy factor
+	delta   float64     // relative-accuracy probability
+	hash    hash.Hash64 // hash function (kernel for all depth functions)
+}
+
+// hashKernel returns the upper and lower base hash values from which the k
+// hashes are derived.
+func hashKernel(data []byte, hash hash.Hash64) (uint32, uint32) {
+	hash.Write(data)
+	sum := hash.Sum64()
+	hash.Reset()
+	upper := uint32(sum & 0xffffffff)
+	lower := uint32((sum >> 32) & 0xffffffff)
+	return upper, lower
+}
+
+// NewCountMinSketch creates a new Count-Min Sketch whose relative accuracy is
+// within a factor of epsilon with probability delta. Both of these parameters
+// affect the space and time complexity.
+func NewCountMinSketch(epsilon, delta float64) *CountMinSketch {
+	var (
+		width  = uint(math.Ceil(math.E / epsilon))
+		depth  = uint(math.Ceil(math.Log(1 / delta)))
+		matrix = make([][]uint64, depth)
+	)
+
+	for i := uint(0); i < depth; i++ {
+		matrix[i] = make([]uint64, width)
+	}
+
+	return &CountMinSketch{
+		matrix:  matrix,
+		width:   width,
+		depth:   depth,
+		epsilon: epsilon,
+		delta:   delta,
+		hash:    fnv.New64(),
+	}
+}
+
+// Epsilon returns the relative-accuracy factor, epsilon.
+func (c *CountMinSketch) Epsilon() float64 {
+	return c.epsilon
+}
+
+// Delta returns the relative-accuracy probability, delta.
+func (c *CountMinSketch) Delta() float64 {
+	return c.delta
+}
+
+// TotalCount returns the number of items added to the sketch.
+func (c *CountMinSketch) TotalCount() uint64 {
+	return c.count
+}
+
+// AddInt will add the data to the set n times. Returns the CountMinSketch to allow for
+// chaining.
+func (c *CountMinSketch) AddInt(data []byte, n uint64) *CountMinSketch {
+	lower, upper := hashKernel(data, c.hash)
+	existingValue := uint64(math.MaxUint64)
+
+	// Increment count in each row.
+	for i := uint(0); i < c.depth; i++ {
+		index := (uint(lower) + uint(upper)*i) % c.width
+		val := c.matrix[i][index]
+		if val < existingValue {
+			existingValue = val
+		}
+		if c.matrix[i][index] < n {
+			c.matrix[i][index] = n
+		}
+	}
+
+	if n > existingValue {
+		c.count += n - existingValue
+	}
+	return c
+}
+
+// Add will add the data to the set. Returns the CountMinSketch to allow for
+// chaining.
+func (c *CountMinSketch) Add(data []byte) *CountMinSketch {
+	lower, upper := hashKernel(data, c.hash)
+
+	// Increment count in each row.
+	for i := uint(0); i < c.depth; i++ {
+		c.matrix[i][(uint(lower)+uint(upper)*i)%c.width]++
+	}
+
+	c.count++
+	return c
+}
+
+// Count returns the approximate count for the specified item, correct within
+// epsilon * total count with a probability of delta.
+func (c *CountMinSketch) Count(data []byte) uint64 {
+	var (
+		lower, upper = hashKernel(data, c.hash)
+		count        = uint64(math.MaxUint64)
+	)
+
+	for i := uint(0); i < c.depth; i++ {
+		count = uint64(math.Min(float64(count),
+			float64(c.matrix[i][(uint(lower)+uint(upper)*i)%c.width])))
+	}
+
+	return count
+}
+
+// Merge combines this CountMinSketch with another. Returns an error if the
+// matrix width and depth are not equal.
+func (c *CountMinSketch) Merge(other *CountMinSketch) error {
+	if c.depth != other.depth {
+		return errors.New("matrix depth must match")
+	}
+
+	if c.width != other.width {
+		return errors.New("matrix width must match")
+	}
+
+	for i := uint(0); i < c.depth; i++ {
+		for j := uint(0); j < c.width; j++ {
+			c.matrix[i][j] += other.matrix[i][j]
+		}
+	}
+
+	c.count += other.count
+	return nil
+}
+
+// Reset restores the CountMinSketch to its original state. It returns itself
+// to allow for chaining.
+func (c *CountMinSketch) Reset() *CountMinSketch {
+	for i := 0; i < len(c.matrix); i++ {
+		for j := 0; j < len(c.matrix[i]); j++ {
+			c.matrix[i][j] = 0
+		}
+	}
+
+	c.count = 0
+	return c
+}
+
+// SetHash sets the hashing function used.
+func (c *CountMinSketch) SetHash(h hash.Hash64) {
+	c.hash = h
+}
+
+// WriteDataTo writes a binary representation of the CMS data to
+// an io stream. It returns the number of bytes written and error
+func (c *CountMinSketch) WriteDataTo(stream io.Writer) (int, error) {
+	buf := new(bytes.Buffer)
+	// serialize epsilon and delta as cms configuration check
+	err := binary.Write(buf, binary.LittleEndian, c.epsilon)
+	if err != nil {
+		return 0, err
+	}
+	err = binary.Write(buf, binary.LittleEndian, c.delta)
+	if err != nil {
+		return 0, err
+	}
+	err = binary.Write(buf, binary.LittleEndian, c.count)
+	if err != nil {
+		return 0, err
+	}
+	// encode matrix
+	for i := range c.matrix {
+		err = binary.Write(buf, binary.LittleEndian, c.matrix[i])
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	return stream.Write(buf.Bytes())
+}
+
+// ReadDataFrom reads a binary representation of the CMS data written
+// by WriteDataTo() from io stream. It returns the number of bytes read
+// and error
+// If serialized CMS configuration is different it returns error with expected params
+func (c *CountMinSketch) ReadDataFrom(stream io.Reader) (int, error) {
+	var (
+		count          uint64
+		epsilon, delta float64
+	)
+
+	err := binary.Read(stream, binary.LittleEndian, &epsilon)
+	if err != nil {
+		return 0, err
+	}
+	err = binary.Read(stream, binary.LittleEndian, &delta)
+	if err != nil {
+		return 0, err
+	}
+
+	// check if serialized and target cms configurations are same
+	if c.epsilon != epsilon || c.delta != delta {
+		return 0, fmt.Errorf("expected cms values for epsilon %f and delta %f", epsilon, delta)
+	}
+
+	err = binary.Read(stream, binary.LittleEndian, &count)
+	if err != nil {
+		return 0, err
+	}
+
+	for i := uint(0); i < c.depth; i++ {
+		err = binary.Read(stream, binary.LittleEndian, c.matrix[i])
+	}
+	// count size of matrix and count
+	size := int(c.depth*c.width)*binary.Size(uint64(0)) + binary.Size(count) + 2*binary.Size(float64(0))
+
+	c.count = count
+
+	return size, err
+}
+
+// TestAndRemove attemps to remove n counts of data from the CMS. If
+// n is greater than the data count, TestAndRemove is a no-op and
+// returns false. Else, return true and decrement count by n.
+func (c *CountMinSketch) TestAndRemove(data []byte, n uint64) bool {
+	h, count := c.traverseDepth(data)
+
+	if n > count {
+		return false
+	}
+
+	for i := uint(0); i < c.depth; i++ {
+		*h[i] -= n
+	}
+
+	return true
+}
+
+// TestAndRemoveAll counts data frequency, performs TestAndRemove(data, count),
+// and returns true if count is positive. If count is 0, TestAndRemoveAll is a
+// no-op and returns false.
+func (c *CountMinSketch) TestAndRemoveAll(data []byte) bool {
+	h, count := c.traverseDepth(data)
+
+	if count == 0 {
+		return false
+	}
+
+	for i := uint(0); i < c.depth; i++ {
+		*h[i] -= count
+	}
+
+	return true
+}
+
+func (c *CountMinSketch) traverseDepth(data []byte) ([]*uint64, uint64) {
+	var (
+		lower, upper = hashKernel(data, c.hash)
+		count        = uint64(math.MaxUint64)
+		h            = make([]*uint64, c.depth)
+	)
+
+	for i := uint(0); i < c.depth; i++ {
+		h[i] = &c.matrix[i][(uint(lower)+uint(upper)*i)%c.width]
+		count = uint64(math.Min(float64(count), float64(*h[i])))
+	}
+
+	return h, count
+}

dgraph/cmd/alpha/run.go

@@ -668,7 +668,7 @@ func run() {
 		AuthToken:          security.GetString("token"),
 		Audit:              conf,
 		ChangeDataConf:     Alpha.Conf.GetString("cdc"),
-		TypeFilterUidLimit: x.Config.Limit.GetInt64("type-filter-uid-limit"),
+		TypeFilterUidLimit: x.Config.Limit.GetUint64("type-filter-uid-limit"),
 	}
 
 	keys, err := ee.GetKeys(Alpha.Conf)

go.mod

@@ -82,7 +82,7 @@ require (
 	github.com/blevesearch/upsidedown_store_api v1.0.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
-	github.com/chewxy/math32 v1.10.1 // indirect
+	github.com/chewxy/math32 v1.11.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/distribution/reference v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect

go.sum

@@ -113,8 +113,8 @@ github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XL
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/chewxy/math32 v1.10.1 h1:LFpeY0SLJXeaiej/eIp2L40VYfscTvKh/FSEZ68uMkU=
-github.com/chewxy/math32 v1.10.1/go.mod h1:dOB2rcuFrCn6UHrze36WSLVPKtzPMRAQvBvUwkSsLqs=
+github.com/chewxy/math32 v1.11.0 h1:8sek2JWqeaKkVnHa7bPVqCEOUPbARo4SGxs6toKyAOo=
+github.com/chewxy/math32 v1.11.0/go.mod h1:dOB2rcuFrCn6UHrze36WSLVPKtzPMRAQvBvUwkSsLqs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=

posting/mvcc.go

@@ -399,9 +399,14 @@ func (c *Cache) clear() {
 }
 
 type MemoryLayer struct {
+	// config
 	deleteOnUpdates bool
-	cache           *Cache
 
+	// data
+	cache *Cache
+
+	// metrics
+	statsHolder  *StatsHolder
 	numDisksRead int
 }
 
@@ -412,9 +417,14 @@ func (ml *MemoryLayer) del(key []byte) {
 	ml.cache.del(key)
 }
 
+func GetStatsHolder() *StatsHolder {
+	return memoryLayer.statsHolder
+}
+
 func initMemoryLayer(cacheSize int64, deleteOnUpdates bool) *MemoryLayer {
 	ml := &MemoryLayer{}
 	ml.deleteOnUpdates = deleteOnUpdates
+	ml.statsHolder = NewStatsHolder()
 	if cacheSize > 0 {
 		cache, err := ristretto.NewCache[[]byte, *CachePL](&ristretto.Config[[]byte, *CachePL]{
 			// Use 5% of cache memory for storing counters.