erigon-pulse/recsplit/recsplit.go

/*
   Copyright 2021 Erigon 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 recsplit

import (
	"encoding/binary"
	"fmt"
	"math"
	"math/bits"

	"github.com/ledgerwatch/erigon-lib/etl"
	"github.com/spaolacci/murmur3"
)

const RecSplitLogPrefix = "recsplit"

const MaxLeafSize = 24

/** David Stafford's (http://zimbry.blogspot.com/2011/09/better-bit-mixing-improving-on.html)
 * 13th variant of the 64-bit finalizer function in Austin Appleby's
 * MurmurHash3 (https://github.com/aappleby/smhasher).
 *
 * @param z a 64-bit integer.
 * @return a 64-bit integer obtained by mixing the bits of `z`.
 */

func remix(z uint64) uint64 {
	z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9
	z = (z ^ (z >> 27)) * 0x94d049bb133111eb
	return z ^ (z >> 31)
}

// RecSplit is the implementation of Recursive Split algorithm for constructing perfect hash mapping, described in
// https://arxiv.org/pdf/1910.06416.pdf Emmanuel Esposito, Thomas Mueller Graf, and Sebastiano Vigna.
// Recsplit: Minimal perfect hashing via recursive splitting. In 2020 Proceedings of the Symposium on Algorithm Engineering and Experiments (ALENEX),
// pages 175−185. SIAM, 2020.
type RecSplit struct {
	bucketSize       int
	keyExpectedCount uint64          // Number of keys in the hash table
	keysAdded        uint64          // Number of keys actually added to the recSplit (to check the match with keyExpectedCount)
	bucketCount      uint64          // Number of buckets
	hasher           murmur3.Hash128 // Salted hash function to use for splitting into initial buckets and mapping to 64-bit fingerprints
	collector        *etl.Collector
	built            bool       // Flag indicating that the hash function has been built and no more keys can be added
	currentBucketIdx uint64     // Current bucket being accumulated
	currentBucket    []uint64   // 64-bit fingerprints of keys in the current bucket accumulated before the recsplit is performed for that bucket
	gr               GolombRice // Helper object to encode the tree of hash function salts using Golomb-Rice code.
	// Helper object to encode the sequence of cumulative number of keys in the buckets
	// and the sequence of of cumulative bit offsets of buckets in the Golomb-Rice code.
	ef                 DoubleEliasFano
	bucketSizeAcc      []uint64 // Bucket size accumulator
	bucketPosAcc       []uint64 // Accumulator for position of every bucket in the encoding of the hash function
	leafSize           uint16   // Leaf size for recursive split algorithm
	primaryAggrBound   uint16   // The lower bound for primary key aggregation (computed from leafSize)
	secondaryAggrBound uint16   // The lower bound for secondary key aggregation (computed from leadSize)
	startSeed          []uint64
	golombRice         []uint32
	buffer             []uint64
	count              []uint16
	salt               uint32 // Murmur3 hash used for converting keys to 64-bit values and assigning to buckets
	collision          bool
	tmpDir             string
	trace              bool
}

type RecSplitArgs struct {
	KeyCount   int
	BucketSize int
	Salt       uint32 // Hash seed (salt) for the hash function used for allocating the initial buckets - need to be generated randomly
	LeafSize   uint16
	TmpDir     string
	StartSeed  []uint64 // For each level of recursive split, the hash seed (salt) used for that level - need to be generated randomly and be large enough to accomodate all the levels
}

// NewRecSplit creates a new RecSplit instance with given number of keys and given bucket size
// Typical bucket size is 100 - 2000, larger bucket sizes result in smaller representations of hash functions, at a cost of slower access
// salt parameters is used to randomise the hash function construction, to ensure that different Erigon instances (nodes)
// are likely to use different hash function, to collision attacks are unlikely to slow down any meaningful number of nodes at the same time
func NewRecSplit(args RecSplitArgs) (*RecSplit, error) {
	bucketCount := (args.KeyCount + args.BucketSize - 1) / args.BucketSize
	rs := &RecSplit{bucketSize: args.BucketSize, keyExpectedCount: uint64(args.KeyCount), bucketCount: uint64(bucketCount)}
	rs.salt = args.Salt
	rs.hasher = murmur3.New128WithSeed(rs.salt)
	rs.tmpDir = args.TmpDir
	rs.collector = etl.NewCollector(rs.tmpDir, etl.NewSortableBuffer(etl.BufferOptimalSize))
	rs.currentBucket = make([]uint64, 0, args.BucketSize)
	rs.bucketSizeAcc = make([]uint64, 1, bucketCount+1)
	rs.bucketPosAcc = make([]uint64, 1, bucketCount+1)
	if args.LeafSize > MaxLeafSize {
		return nil, fmt.Errorf("exceeded max leaf size %d: %d", MaxLeafSize, args.LeafSize)
	}
	rs.leafSize = args.LeafSize
	rs.primaryAggrBound = rs.leafSize * uint16(math.Max(2, math.Ceil(0.35*float64(rs.leafSize)+1./2.)))
	if rs.leafSize < 7 {
		rs.secondaryAggrBound = rs.primaryAggrBound * 2
	} else {
		rs.secondaryAggrBound = rs.primaryAggrBound * uint16(math.Ceil(0.21*float64(rs.leafSize)+9./10.))
	}
	rs.startSeed = args.StartSeed
	rs.count = make([]uint16, rs.secondaryAggrBound)
	return rs, nil
}

func (rs *RecSplit) SetTrace(trace bool) {
	rs.trace = trace
}

// remap converts the number x which is assumed to be uniformly distributed over the range [0..2^64) to the number that is uniformly
// distributed over the range [0..n)
func remap(x uint64, n uint64) uint64 {
	hi, _ := bits.Mul64(x, n)
	return hi
}

const mask48 uint64 = (1 << 48) - 1

// remap converts the number x which is assumed to be uniformly distributed over the range [0..2^64) to the number that is uniformly
// distributed over the range [0..n), under assumption that n is less than 2^16
func remap16(x uint64, n uint16) uint16 {
	return uint16(((x & mask48) * uint64(n)) >> 48)
}

// ResetNextSalt resets the RecSplit and uses the next salt value to try to avoid collisions
// when mapping keys to 64-bit values
func (rs *RecSplit) ResetNextSalt() {
	rs.collision = false
	rs.keysAdded = 0
	rs.salt++
	rs.hasher = murmur3.New128WithSeed(rs.salt)
	rs.collector = etl.NewCollector(rs.tmpDir, etl.NewSortableBuffer(etl.BufferOptimalSize))
	rs.currentBucket = rs.currentBucket[:0]
	rs.bucketSizeAcc = rs.bucketSizeAcc[:1] // First entry is always zero
	rs.bucketPosAcc = rs.bucketPosAcc[:0]   // First entry is always zero
}

func (rs *RecSplit) splitParams(m uint16) (fanout, unit uint16) {
	if m > rs.secondaryAggrBound { // High-level aggregation (fanout 2)
		unit = rs.secondaryAggrBound * (((m+1)/2 + rs.secondaryAggrBound - 1) / rs.secondaryAggrBound)
		fanout = 2
	} else if m > rs.primaryAggrBound { // Second-level aggregation
		unit = rs.primaryAggrBound
		fanout = (m + rs.primaryAggrBound - 1) / rs.primaryAggrBound
	} else { // First-level aggregation
		unit = rs.leafSize
		fanout = (m + rs.leafSize - 1) / rs.leafSize
	}
	return
}

func (rs *RecSplit) computeGolombRice(m uint16, table []uint32) {
	fanout, unit := rs.splitParams(m)
	k := make([]uint16, fanout)
	k[fanout-1] = m
	for i := uint16(0); i < fanout-1; i++ {
		k[i] = unit
		k[fanout-1] -= k[i]
	}
	sqrt_prod := float64(1)
	for i := uint16(0); i < fanout; i++ {
		sqrt_prod *= math.Sqrt(float64(k[i]))
	}
	p := math.Sqrt(float64(m)) / (math.Pow(2*math.Pi, (float64(fanout)-1.)/2.0) * sqrt_prod)
	golombRiceLength := uint32(math.Ceil(math.Log2(-math.Log((math.Sqrt(5)+1.0)/2.0) / math.Log1p(-p)))) // log2 Golomb modulus
	if golombRiceLength > 0x1F {
		panic("golombRiceLength > 0x1F")
	}
	table[m] = golombRiceLength << 27
	for i := uint16(0); i < fanout; i++ {
		golombRiceLength += table[k[i]] & 0xFFFF
	}
	if golombRiceLength > 0xFFFF {
		panic("golombRiceLength > 0xFFFF")
	}
	table[m] |= golombRiceLength // Sum of Golomb-Rice codeslengths in the subtree, stored in the lower 16 bits
	nodes := uint32(1)
	for i := uint16(0); i < fanout; i++ {
		nodes += (table[k[i]] >> 16) & 0x7FF
	}
	if rs.leafSize >= 3 && nodes > 0x7FF {
		panic("rs.leafSize >= 3 && nodes > 0x7FF")
	}
	table[m] |= nodes << 16
}

// golombParam returns the optimal Golomb parameter to use for encoding
// salt for the part of the hash function separating m elements. It is based on
// calculations with assumptions that we draw hash functions at random
func (rs *RecSplit) golombParam(m uint16) int {
	s := uint16(len(rs.golombRice))
	for m >= s {
		rs.golombRice = append(rs.golombRice, 0)
		// For the case where bucket is larger than planned
		if s == 0 {
			rs.golombRice[0] = (bijMemo[0] << 27) | bijMemo[0]
		} else if s <= rs.leafSize {
			rs.golombRice[s] = (bijMemo[s] << 27) | (uint32(1) << 16) | bijMemo[s]
		} else {
			rs.computeGolombRice(s, rs.golombRice)
		}
		s++
	}
	return int(rs.golombRice[m] >> 27)
}

// Add key to the RecSplit. There can be many more keys than what fits in RAM, and RecSplit
// spills data onto disk to accomodate that. The key gets copied by the collector, therefore
// the slice underlying key is not getting accessed by RecSplit after this invocation.
func (rs *RecSplit) AddKey(key []byte) error {
	if rs.built {
		return fmt.Errorf("cannot add keys after perfect hash function had been built")
	}
	rs.hasher.Reset()
	rs.hasher.Write(key) //nolint:errcheck
	hi, lo := rs.hasher.Sum128()
	var bucketKey [16]byte
	binary.BigEndian.PutUint64(bucketKey[:], remap(hi, rs.bucketCount))
	binary.BigEndian.PutUint64(bucketKey[8:], lo)
	rs.keysAdded++
	return rs.collector.Collect(bucketKey[:], []byte{})
}

func (rs *RecSplit) recsplitCurrentBucket() error {
	// Extend rs.bucketSizeAcc to accomodate current bucket index + 1
	for len(rs.bucketSizeAcc) <= int(rs.currentBucketIdx)+1 {
		rs.bucketSizeAcc = append(rs.bucketSizeAcc, rs.bucketSizeAcc[len(rs.bucketSizeAcc)-1])
	}
	rs.bucketSizeAcc[int(rs.currentBucketIdx)+1] += uint64(len(rs.currentBucket))
	if len(rs.currentBucket) > 1 {
		for i, key := range rs.currentBucket[1:] {
			if key == rs.currentBucket[i] {
				rs.collision = true
				return fmt.Errorf("duplicate key %x", key)
			}
		}
		bitPos := rs.gr.bitCount
		if rs.buffer == nil {
			rs.buffer = make([]uint64, len(rs.currentBucket))
		} else {
			for len(rs.buffer) < len(rs.currentBucket) {
				rs.buffer = append(rs.buffer, 0)
			}
		}
		unary := rs.recsplit(0 /* level */, rs.currentBucket, nil /* unary */)
		rs.gr.appendUnaryAll(unary)
		if rs.trace {
			fmt.Printf("recsplitBucket(%d, %d, bitsize = %d)\n", rs.currentBucketIdx, len(rs.currentBucket), rs.gr.bitCount-bitPos)
		}
	}
	// Extend rs.bucketPosAcc to accomodate current bucket index + 1
	for len(rs.bucketPosAcc) <= int(rs.currentBucketIdx)+1 {
		rs.bucketPosAcc = append(rs.bucketPosAcc, rs.bucketPosAcc[len(rs.bucketPosAcc)-1])
	}
	rs.bucketPosAcc[int(rs.currentBucketIdx)+1] = uint64(rs.gr.Bits())
	// clear for the next buckey
	rs.currentBucket = rs.currentBucket[:0]
	return nil
}

// recsplit applies recSplit algorithm to the given bucket
func (rs *RecSplit) recsplit(level int, bucket []uint64, unary []uint64) []uint64 {
	if rs.trace {
		fmt.Printf("recsplit(%d, %d, %x)\n", level, len(bucket), bucket)
	}
	// Pick initial salt for this level of recursive split
	salt := rs.startSeed[level]
	m := uint16(len(bucket))
	if m <= rs.leafSize {
		// No need to build aggregation levels - just find find bijection
		var mask uint32
		for {
			mask = 0
			var fail bool
			for i := uint16(0); !fail && i < m; i++ {
				bit := uint32(1) << remap16(remix(bucket[i]+salt), m)
				if mask&bit != 0 {
					fail = true
				} else {
					mask |= bit
				}
			}
			if !fail {
				break
			}
			salt++
		}
		salt -= rs.startSeed[level]
		log2golomb := rs.golombParam(m)
		if rs.trace {
			fmt.Printf("encode bij %d with log2golomn %d at p = %d\n", salt, log2golomb, rs.gr.bitCount)
		}
		rs.gr.appendFixed(salt, log2golomb)
		unary = append(unary, salt>>log2golomb)
	} else {
		fanout, unit := rs.splitParams(m)
		count := rs.count
		for {
			for i := uint16(0); i < fanout-1; i++ {
				count[i] = 0
			}
			var fail bool
			for i := uint16(0); i < m; i++ {
				count[remap16(remix(bucket[i]+salt), m)/unit]++
			}
			for i := uint16(0); i < fanout-1; i++ {
				fail = fail || (count[i] != unit)
			}
			if !fail {
				break
			}
			salt++
		}
		for i, c := uint16(0), uint16(0); i < fanout; i++ {
			count[i] = c
			c += unit
		}
		for i := uint16(0); i < m; i++ {
			j := remap16(remix(bucket[i]+salt), m) / unit
			rs.buffer[count[j]] = bucket[i]
			count[j]++
		}
		copy(bucket, rs.buffer)
		salt -= rs.startSeed[level]
		log2golomb := rs.golombParam(m)
		if rs.trace {
			fmt.Printf("encode fanout %d: %d with log2golomn %d at p = %d\n", fanout, salt, log2golomb, rs.gr.bitCount)
		}
		rs.gr.appendFixed(salt, log2golomb)
		unary = append(unary, salt>>log2golomb)
		var i uint16
		for i = 0; i < m-unit; i += unit {
			unary = rs.recsplit(level+1, bucket[i:i+unit], unary)
		}
		if m-i > 1 {
			unary = rs.recsplit(level+1, bucket[i:], unary)
		}
	}
	return unary
}

// loadFunc is required to satisfy the type etl.LoadFunc type, to use with collector.Load
func (rs *RecSplit) loadFunc(k, v []byte, table etl.CurrentTableReader, next etl.LoadNextFunc) error {
	// k is the BigEndian encoding of the bucket number, and the v is the key that is assigned into that bucket
	bucketIdx := binary.BigEndian.Uint64(k)
	if rs.currentBucketIdx != bucketIdx {
		if rs.currentBucketIdx != math.MaxUint64 {
			if err := rs.recsplitCurrentBucket(); err != nil {
				return err
			}
		}
		rs.currentBucketIdx = bucketIdx
	}
	rs.currentBucket = append(rs.currentBucket, binary.BigEndian.Uint64(k[8:]))
	return nil
}

// Build has to be called after all the keys have been added, and it initiates the process
// of building the perfect hash function.
func (rs *RecSplit) Build() error {
	if rs.built {
		return fmt.Errorf("already built")
	}
	if rs.keysAdded != rs.keyExpectedCount {
		return fmt.Errorf("expected keys %d, got %d", rs.keyExpectedCount, rs.keysAdded)
	}
	rs.currentBucketIdx = math.MaxUint64 // To make sure 0 bucket is detected
	defer rs.collector.Close(RecSplitLogPrefix)
	if err := rs.collector.Load(RecSplitLogPrefix, nil /* db */, "" /* toBucket */, rs.loadFunc, etl.TransformArgs{}); err != nil {
		return err
	}
	if len(rs.currentBucket) > 0 {
		if err := rs.recsplitCurrentBucket(); err != nil {
			return err
		}
	}
	rs.gr.appendFixed(1, 1) // Sentinel (avoids checking for parts of size 1)
	// Construct Elias Fano index
	rs.ef.Build(rs.bucketSizeAcc, rs.bucketPosAcc)
	rs.built = true
	return nil
}

func (rs *RecSplit) skipBits(m uint16) int {
	return int(rs.golombRice[m] & 0xffff)
}

func (rs *RecSplit) skipNodes(m uint16) int {
	return int(rs.golombRice[m]>>16) & 0x7FF
}

func (rs *RecSplit) Lookup(key []byte, trace bool) int {
	rs.hasher.Reset()
	rs.hasher.Write(key) //nolint:errcheck
	bucketHash, fingerprint := rs.hasher.Sum128()
	if trace {
		fmt.Printf("lookup key %x, fingerprint %x\n", key, fingerprint)
	}
	bucket := remap(bucketHash, rs.bucketCount)
	cumKeys, cumKeysNext, bitPos := rs.ef.Get3(bucket)
	m := uint16(cumKeysNext - cumKeys) // Number of keys in this bucket
	if trace {
		fmt.Printf("bucket: %d, m = %d, bitPos = %d, unaryOffset = %d\n", bucket, m, bitPos, rs.skipBits(m))
	}
	rs.gr.ReadReset(int(bitPos), rs.skipBits(m))
	var level int
	var p int
	for m > rs.secondaryAggrBound { // fanout = 2
		if trace {
			p = rs.gr.currFixedOffset
		}
		d := rs.gr.ReadNext(rs.golombParam(m))
		if trace {
			fmt.Printf("level %d, p = %d, d = %d golomb %d\n", level, p, d, rs.golombParam(m))
		}
		hmod := remap16(remix(fingerprint+rs.startSeed[level]+d), m)
		split := (((m+1)/2 + rs.secondaryAggrBound - 1) / rs.secondaryAggrBound) * rs.secondaryAggrBound
		if hmod < split {
			m = split
		} else {
			rs.gr.SkipSubtree(rs.skipNodes(split), rs.skipBits(split))
			m -= split
			cumKeys += uint64(split)
		}
		level++
	}
	if m > rs.primaryAggrBound {
		if trace {
			p = rs.gr.currFixedOffset
		}
		d := rs.gr.ReadNext(rs.golombParam(m))
		if trace {
			fmt.Printf("level %d, p = %d, d = %d golomb %d\n", level, p, d, rs.golombParam(m))
		}
		hmod := remap16(remix(fingerprint+rs.startSeed[level]+d), m)
		part := hmod / rs.primaryAggrBound
		if rs.primaryAggrBound < m-part*rs.primaryAggrBound {
			m = rs.primaryAggrBound
		} else {
			m = m - part*rs.primaryAggrBound
		}
		cumKeys += uint64(rs.primaryAggrBound * part)
		if part != 0 {
			rs.gr.SkipSubtree(rs.skipNodes(rs.primaryAggrBound)*int(part), rs.skipBits(rs.primaryAggrBound)*int(part))
		}
		level++
	}
	if m > rs.leafSize {
		if trace {
			p = rs.gr.currFixedOffset
		}
		d := rs.gr.ReadNext(rs.golombParam(m))
		if trace {
			fmt.Printf("level %d, p = %d, d = %d, golomb %d\n", level, p, d, rs.golombParam(m))
		}
		hmod := remap16(remix(fingerprint+rs.startSeed[level]+d), m)
		part := hmod / rs.leafSize
		if rs.leafSize < m-part*rs.leafSize {
			m = rs.leafSize
		} else {
			m = m - part*rs.leafSize
		}
		cumKeys += uint64(rs.leafSize * part)
		if part != 0 {
			rs.gr.SkipSubtree(int(part), rs.skipBits(rs.leafSize)*int(part))
		}
		level++
	}
	if trace {
		p = rs.gr.currFixedOffset
	}
	b := rs.gr.ReadNext(rs.golombParam(m))
	if trace {
		fmt.Printf("level %d, p = %d, b = %d, golomn = %d\n", level, p, b, rs.golombParam(m))
	}
	return int(cumKeys) + int(remap16(remix(fingerprint+rs.startSeed[level]+b), m))
}

// Stats returns the size of golomb rice encoding and ellias fano encoding
func (rs RecSplit) Stats() (int, int) {
	return len(rs.gr.Data()), len(rs.ef.Data())
}

// Collision returns true if there was a collision detected during mapping of keys
// into 64-bit values
// RecSplit needs to be reset, re-populated with keys, and rebuilt
func (rs RecSplit) Collision() bool {
	return rs.collision
}