diff --git a/internal/quick_select.go b/internal/quick_select.go
index ab53a17..360f280 100644
--- a/internal/quick_select.go
+++ b/internal/quick_select.go
@@ -61,3 +61,51 @@ func partition[T cmp.Ordered](arr []T, lo int, hi int) int {
 	arr[lo], arr[j] = arr[j], arr[lo]
 	return j
 }
+
+// QuickSelectFunc finds the k-th smallest element in a slice using the Quickselect algorithm with a custom comparator.
+// The slice is partially partitioned and may not maintain full order.
+// It modifies the input slice in-place.
+// T is a generic type, and the comparison logic is provided by the `compare` function.
+// The `lo` and `hi` parameters define the range in the slice to consider for the selection.
+func QuickSelectFunc[T any](arr []T, lo int, hi int, pivot int, compare func(a, b T) int) T {
+	for hi > lo {
+		j := partitionFunc(arr, lo, hi, compare)
+		if j == pivot {
+			return arr[pivot]
+		}
+		if j > pivot {
+			hi = j - 1
+		} else {
+			lo = j + 1
+		}
+	}
+	return arr[pivot]
+}
+
+func partitionFunc[T any](arr []T, lo int, hi int, compare func(a, b T) int) int {
+	i := lo
+	j := hi + 1
+	v := arr[lo]
+	for {
+		for compare(arr[i+1], v) < 0 {
+			i++
+			if i == hi {
+				break
+			}
+		}
+		i++
+		for compare(v, arr[j-1]) < 0 {
+			j--
+			if j == lo {
+				break
+			}
+		}
+		j--
+		if i >= j {
+			break
+		}
+		arr[i], arr[j] = arr[j], arr[i]
+	}
+	arr[lo], arr[j] = arr[j], arr[lo]
+	return j
+}
diff --git a/internal/quick_select_test.go b/internal/quick_select_test.go
index f90781e..72f8553 100644
--- a/internal/quick_select_test.go
+++ b/internal/quick_select_test.go
@@ -149,3 +149,93 @@ func TestQuickSelectString(t *testing.T) {
 
 	assert.Equal(t, expected, result, "want: %v\ngot: %v", expected, result)
 }
+
+type testEntry struct {
+	hash    uint64
+	summary any
+}
+
+func TestQuickSelectFunc(t *testing.T) {
+	testCases := []struct {
+		name     string
+		arr      []testEntry
+		lo       int
+		hi       int
+		pivot    int
+		expected uint64
+	}{
+		{
+			name:     "two elements first smaller",
+			arr:      []testEntry{{hash: 50}, {hash: 100}},
+			lo:       0,
+			hi:       1,
+			pivot:    1,
+			expected: 100,
+		},
+		{
+			name:     "find median",
+			arr:      []testEntry{{hash: 3}, {hash: 1}, {hash: 4}, {hash: 1}, {hash: 5}, {hash: 9}, {hash: 2}, {hash: 6}},
+			lo:       0,
+			hi:       7,
+			pivot:    4,
+			expected: 4,
+		},
+		{
+			name:     "find minimum",
+			arr:      []testEntry{{hash: 3}, {hash: 1}, {hash: 4}, {hash: 1}, {hash: 5}, {hash: 9}, {hash: 2}, {hash: 6}},
+			lo:       0,
+			hi:       7,
+			pivot:    0,
+			expected: 1,
+		},
+		{
+			name:     "find maximum",
+			arr:      []testEntry{{hash: 3}, {hash: 1}, {hash: 4}, {hash: 1}, {hash: 5}, {hash: 9}, {hash: 2}, {hash: 6}},
+			lo:       0,
+			hi:       7,
+			pivot:    7,
+			expected: 9,
+		},
+		{
+			name:     "single element",
+			arr:      []testEntry{{hash: 42}},
+			lo:       0,
+			hi:       0,
+			pivot:    0,
+			expected: 42,
+		},
+		{
+			name:     "two elements descending",
+			arr:      []testEntry{{hash: 5}, {hash: 3}},
+			lo:       0,
+			hi:       1,
+			pivot:    0,
+			expected: 3,
+		},
+		{
+			name:     "with summary data",
+			arr:      []testEntry{{hash: 30, summary: "a"}, {hash: 10, summary: "b"}, {hash: 20, summary: "c"}},
+			lo:       0,
+			hi:       2,
+			pivot:    1,
+			expected: 20,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			arrCopy := make([]testEntry, len(tc.arr))
+			copy(arrCopy, tc.arr)
+
+			result := QuickSelectFunc(arrCopy, tc.lo, tc.hi, tc.pivot, func(a, b testEntry) int {
+				if a.hash < b.hash {
+					return -1
+				} else if a.hash > b.hash {
+					return 1
+				}
+				return 0
+			})
+
+			assert.Equal(t, tc.expected, result.hash, "want: %v\ngot: %v", tc.expected, result.hash)
+		})
+	}
+}
diff --git a/tuple/hashtable.go b/tuple/hashtable.go
new file mode 100644
index 0000000..e0b5fef
--- /dev/null
+++ b/tuple/hashtable.go
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tuple
+
+import (
+	"errors"
+	"fmt"
+	"math"
+
+	"github.com/apache/datasketches-go/internal"
+	"github.com/apache/datasketches-go/theta"
+)
+
+const (
+	resizeThreshold  = 0.5
+	rebuildThreshold = 15.0 / 16.0
+)
+
+const (
+	strideHashBits = 7
+	strideMask     = (1 << strideHashBits) - 1
+)
+
+var (
+	ErrKeyNotFound                = errors.New("key not found")
+	ErrKeyNotFoundAndNoEmptySlots = errors.New("key not found and no empty slots")
+	// ErrZeroHashValue is used to indicate that the hash value is zero.
+	// Zero is a reserved value for empty slots in the hash table.
+	ErrZeroHashValue = errors.New("zero hash value")
+)
+
+type entry[S Summary] struct {
+	Hash    uint64
+	Summary S
+}
+
+func (e *entry[S]) reset() {
+	if e.Hash != 0 {
+		e.Summary.Reset()
+	}
+	e.Hash = 0
+}
+
+type hashtable[S Summary] struct {
+	entries       []entry[S]
+	entryLessFunc func(a, b entry[S]) int
+	theta         uint64
+	seed          uint64
+	numEntries    uint32
+	p             float32
+	lgCurSize     uint8
+	lgNomSize     uint8
+	rf            theta.ResizeFactor
+	isEmpty       bool
+}
+
+func newHashtable[S Summary](lgCurSize, lgNomSize uint8, rf theta.ResizeFactor, p float32, theta, seed uint64, isEmpty bool) *hashtable[S] {
+	sketch := &hashtable[S]{
+		isEmpty:    isEmpty,
+		lgCurSize:  lgCurSize,
+		lgNomSize:  lgNomSize,
+		rf:         rf,
+		p:          p,
+		numEntries: 0,
+		theta:      theta,
+		seed:       seed,
+		entries:    nil,
+		entryLessFunc: func(a, b entry[S]) int {
+			if a.Hash < b.Hash {
+				return -1
+			} else if a.Hash > b.Hash {
+				return 1
+			}
+			return 0
+		},
+	}
+
+	if lgCurSize > 0 {
+		size := 1 << lgCurSize
+		sketch.entries = make([]entry[S], size)
+	}
+
+	return sketch
+}
+
+// HashStringAndScreen computes the hash of string and checks if it passes theta threshold
+func (t *hashtable[S]) HashStringAndScreen(data string) (uint64, error) {
+	t.isEmpty = false
+	h1, _ := internal.HashCharSliceMurmur3([]byte(data), 0, len(data), t.seed)
+	hash := h1 >> 1
+	if hash >= t.theta {
+		return 0, fmt.Errorf("hash %d is greater than or equal to theta %d", hash, t.theta)
+	}
+	if hash == 0 {
+		return 0, ErrZeroHashValue
+	}
+	return hash, nil
+}
+
+// HashInt32AndScreen computes the hash of int32 and checks if it passes theta threshold
+func (t *hashtable[S]) HashInt32AndScreen(data int32) (uint64, error) {
+	t.isEmpty = false
+	h1, _ := internal.HashInt32SliceMurmur3([]int32{data}, 0, 1, t.seed)
+	hash := h1 >> 1
+	if hash >= t.theta {
+		return 0, fmt.Errorf("hash %d is greater than or equal to theta %d", hash, t.theta)
+	}
+	if hash == 0 {
+		return 0, ErrZeroHashValue
+	}
+	return hash, nil
+}
+
+// HashInt64AndScreen computes the hash of int64 and checks if it passes theta threshold
+func (t *hashtable[S]) HashInt64AndScreen(data int64) (uint64, error) {
+	t.isEmpty = false
+	h1, _ := internal.HashInt64SliceMurmur3([]int64{data}, 0, 1, t.seed)
+	hash := h1 >> 1
+	if hash >= t.theta {
+		return 0, fmt.Errorf("hash %d is greater than or equal to theta %d", hash, t.theta)
+	}
+	if hash == 0 {
+		return 0, ErrZeroHashValue
+	}
+	return hash, nil
+}
+
+// HashBytesAndScreen computes the hash of bytes and checks if it passes theta threshold
+func (t *hashtable[S]) HashBytesAndScreen(data []byte) (uint64, error) {
+	t.isEmpty = false
+	h1, _ := internal.HashByteArrMurmur3(data, 0, len(data), t.seed)
+	hash := h1 >> 1
+	if hash >= t.theta {
+		return 0, fmt.Errorf("hash %d is greater than or equal to theta %d", hash, t.theta)
+	}
+	if hash == 0 {
+		return 0, ErrZeroHashValue
+	}
+	return hash, nil
+}
+
+// Find searches for an entry in the hash table and returns the index if found,
+// or an error if not found
+func (t *hashtable[S]) Find(key uint64) (int, error) {
+	return find(t.entries, t.lgCurSize, key)
+}
+
+func find[S Summary](entries []entry[S], lgSize uint8, key uint64) (int, error) {
+	size := uint32(1 << lgSize)
+	mask := size - 1
+	stride := computeStride(key, lgSize)
+	index := uint32(key) & mask
+
+	loopIndex := index
+	for {
+		probe := entries[index]
+		if probe.Hash == 0 {
+			return int(index), ErrKeyNotFound
+		} else if probe.Hash == key {
+			return int(index), nil
+		}
+
+		index = (index + stride) & mask
+		if index == loopIndex {
+			return 0, ErrKeyNotFoundAndNoEmptySlots
+		}
+	}
+}
+
+// computeStride computes the stride for probing
+func computeStride(key uint64, lgSize uint8) uint32 {
+	// odd and independent of the index assuming lg_size lowest bits of the key were used for the index
+	return (2 * uint32((key>>lgSize)&strideMask)) + 1
+}
+
+// Insert inserts an entry at the given index
+func (t *hashtable[S]) Insert(index int, entry entry[S]) {
+	t.entries[index] = entry
+	t.numEntries++
+
+	if t.numEntries > computeCapacity(t.lgCurSize, t.lgNomSize) {
+		if t.lgCurSize <= t.lgNomSize {
+			t.resize()
+		} else {
+			t.rebuild()
+		}
+	}
+}
+
+func computeCapacity(lgCurSize, lgNomSize uint8) uint32 {
+	var fraction float64
+	if lgCurSize <= lgNomSize {
+		fraction = resizeThreshold
+	} else {
+		fraction = rebuildThreshold
+	}
+	return uint32(math.Floor(fraction * float64(uint32(1)<<lgCurSize)))
+}
+
+func (t *hashtable[S]) resize() {
+	oldSize := 1 << t.lgCurSize
+	lgNewSize := min(t.lgCurSize+uint8(t.rf), t.lgNomSize+1)
+	newSize := 1 << lgNewSize
+	newEntries := make([]entry[S], newSize)
+
+	for i := 0; i < oldSize; i++ {
+		e := t.entries[i]
+		if e.Hash != 0 {
+			// always finds an empty slot in a larger table
+			index, _ := find(newEntries, lgNewSize, e.Hash)
+			newEntries[index] = e
+		}
+	}
+
+	t.entries = newEntries
+	t.lgCurSize = lgNewSize
+}
+
+func (t *hashtable[S]) rebuild() {
+	size := 1 << t.lgCurSize
+	nominalSize := 1 << t.lgNomSize
+
+	// empty entries have uninitialized payloads
+	consolidateNonEmpty(t.entries, size, int(t.numEntries))
+
+	internal.QuickSelectFunc[entry[S]](t.entries[:t.numEntries], 0, int(t.numEntries)-1, nominalSize, t.entryLessFunc)
+	t.theta = t.entries[nominalSize].Hash
+
+	oldEntries := t.entries
+	t.entries = make([]entry[S], size)
+	t.numEntries = uint32(nominalSize)
+
+	// reinsert entries below new theta
+	for i := 0; i < nominalSize; i++ {
+		index, _ := find(t.entries, t.lgCurSize, oldEntries[i].Hash)
+		t.entries[index] = oldEntries[i]
+	}
+}
+
+// Trim reduces the sketch to nominal size if needed
+func (t *hashtable[S]) Trim() {
+	if t.numEntries > uint32(1<<t.lgNomSize) {
+		t.rebuild()
+	}
+}
+
+// Reset clears the sketch
+func (t *hashtable[S]) Reset() {
+	startingLgSize := startingSubMultiple(t.lgNomSize+1, theta.MinLgK, uint8(t.rf))
+
+	if startingLgSize != t.lgCurSize {
+		t.lgCurSize = startingLgSize
+		newSize := 1 << startingLgSize
+		t.entries = make([]entry[S], newSize)
+	} else {
+		// just clear existing entries
+		for i := range t.entries {
+			t.entries[i].reset()
+		}
+	}
+
+	t.numEntries = 0
+	t.theta = startingThetaFromP(t.p)
+	t.isEmpty = true
+}
+
+func consolidateNonEmpty[S Summary](entries []entry[S], size, num int) {
+	// find the first empty slot
+	i := 0
+	for i < size && entries[i].Hash != 0 {
+		i++
+	}
+
+	// scan the rest and move non-empty entries to the front
+	for j := i + 1; j < size; j++ {
+		if entries[j].Hash != 0 {
+			entries[i] = entries[j]
+			entries[j] = entry[S]{}
+			i++
+			if i == num {
+				break
+			}
+		}
+	}
+}
+
+// startingThetaFromP returns the starting theta value from probability p
+// Consistent way of initializing theta from p
+// Avoids multiplication if p == 1 since it might not yield MAX_THETA exactly
+func startingThetaFromP(p float32) uint64 {
+	if p < 1 {
+		return uint64(float64(theta.MaxTheta) * float64(p))
+	}
+	return theta.MaxTheta
+}
+
+// startingSubMultiple calculates the starting sub-multiple
+func startingSubMultiple(lgTgt, lgMin, lgRf uint8) uint8 {
+	if lgTgt <= lgMin {
+		return lgMin
+	}
+	if lgRf == 0 {
+		return lgTgt
+	}
+	return ((lgTgt - lgMin) % lgRf) + lgMin
+}
diff --git a/tuple/hashtable_test.go b/tuple/hashtable_test.go
new file mode 100644
index 0000000..bbe501b
--- /dev/null
+++ b/tuple/hashtable_test.go
@@ -0,0 +1,595 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tuple
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/apache/datasketches-go/theta"
+)
+
+func TestHashtable_HashStringAndScreen(t *testing.T) {
+	testCases := []struct {
+		name       string
+		data       string
+		theta      uint64
+		seed       uint64
+		wantErrMsg string
+	}{
+		{
+			name:       "normal string with max theta",
+			data:       "hello world",
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "empty string",
+			data:       "",
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "string with special characters",
+			data:       "test@#$%^&*()",
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "unicode string",
+			data:       "가나다라마바사",
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "with low theta (likely filtered)",
+			data:       "test",
+			theta:      1,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "is greater than or equal to theta 1",
+		},
+		{
+			name:       "different seed",
+			data:       "test",
+			theta:      theta.MaxTheta,
+			seed:       99999,
+			wantErrMsg: "",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, tc.theta, tc.seed, true)
+			hash, err := ht.HashStringAndScreen(tc.data)
+
+			assert.False(t, ht.isEmpty)
+			if tc.wantErrMsg != "" {
+				assert.ErrorContains(t, err, tc.wantErrMsg)
+			} else {
+				assert.NotZero(t, hash, "Expected non-zero hash for data: %s", tc.data)
+			}
+		})
+	}
+}
+
+func TestHashtable_HashStringAndScreenConsistency(t *testing.T) {
+	ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+	hash1, err := ht.HashStringAndScreen("test")
+	assert.NoError(t, err)
+	hash2, err := ht.HashStringAndScreen("test")
+	assert.NoError(t, err)
+
+	assert.Equal(t, hash1, hash2, "Same string should produce same hash")
+}
+
+func TestHashtable_HashInt32AndScreen(t *testing.T) {
+	testCases := []struct {
+		name       string
+		data       int32
+		theta      uint64
+		seed       uint64
+		wantErrMsg string
+	}{
+		{
+			name:       "positive integer",
+			data:       12345,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "negative integer",
+			data:       -12345,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "zero",
+			data:       0,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "max int32",
+			data:       2147483647,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "min int32",
+			data:       -2147483648,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "with very low theta (likely filtered)",
+			data:       12345,
+			theta:      1,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "is greater than or equal to theta 1",
+		},
+		{
+			name:       "different seed",
+			data:       12345,
+			theta:      theta.MaxTheta,
+			seed:       99999,
+			wantErrMsg: "",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, tc.theta, tc.seed, true)
+			hash, err := ht.HashInt32AndScreen(tc.data)
+
+			assert.False(t, ht.isEmpty)
+
+			if tc.wantErrMsg != "" {
+				assert.ErrorContains(t, err, tc.wantErrMsg)
+			} else {
+				assert.NotZero(t, hash, "Expected non-zero hash for data: %d", tc.data)
+			}
+		})
+	}
+}
+
+func TestHashtable_HashInt32AndScreenConsistency(t *testing.T) {
+	ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+	hash1, err := ht.HashInt32AndScreen(42)
+	assert.NoError(t, err)
+	hash2, err := ht.HashInt32AndScreen(42)
+	assert.NoError(t, err)
+	assert.Equal(t, hash1, hash2, "Same int32 should produce same hash")
+}
+
+func TestHashtable_HashInt64AndScreen(t *testing.T) {
+	testCases := []struct {
+		name       string
+		data       int64
+		theta      uint64
+		seed       uint64
+		wantErrMsg string
+	}{
+		{
+			name:       "positive integer",
+			data:       1234567890,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "negative integer",
+			data:       -1234567890,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "zero",
+			data:       0,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "max int64",
+			data:       9223372036854775807,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "min int64",
+			data:       -9223372036854775808,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "large positive value",
+			data:       9876543210123456,
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "with very low theta (likely filtered)",
+			data:       1234567890,
+			theta:      1,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "is greater than or equal to theta 1",
+		},
+		{
+			name:       "different seed",
+			data:       1234567890,
+			theta:      theta.MaxTheta,
+			seed:       55555,
+			wantErrMsg: "",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, tc.theta, tc.seed, true)
+			hash, err := ht.HashInt64AndScreen(tc.data)
+
+			assert.False(t, ht.isEmpty)
+
+			if tc.wantErrMsg != "" {
+				assert.ErrorContains(t, err, tc.wantErrMsg)
+			} else {
+				assert.NotZero(t, hash, "Expected non-zero hash for data: %d", tc.data)
+			}
+		})
+	}
+}
+
+func TestHashtable_HashInt64AndScreenConsistency(t *testing.T) {
+	ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+	hash1, err := ht.HashInt64AndScreen(123456789)
+	assert.NoError(t, err)
+	hash2, err := ht.HashInt64AndScreen(123456789)
+	assert.NoError(t, err)
+	assert.Equal(t, hash1, hash2, "Same int64 should produce same hash")
+}
+
+func TestHashtable_HashBytesAndScreen(t *testing.T) {
+	testCases := []struct {
+		name       string
+		data       []byte
+		theta      uint64
+		seed       uint64
+		wantErrMsg string
+	}{
+		{
+			name:       "normal byte array",
+			data:       []byte{1, 2, 3, 4, 5},
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "empty byte array",
+			data:       []byte{},
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "single byte",
+			data:       []byte{42},
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "byte array from string",
+			data:       []byte("hello world"),
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "byte array with zeros",
+			data:       []byte{0, 0, 0, 0},
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "byte array with max values",
+			data:       []byte{255, 255, 255, 255},
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "large byte array",
+			data:       make([]byte, 1000),
+			theta:      theta.MaxTheta,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "",
+		},
+		{
+			name:       "with very low theta (likely filtered)",
+			data:       []byte{1, 2, 3, 4, 5},
+			theta:      100,
+			seed:       theta.DefaultSeed,
+			wantErrMsg: "is greater than or equal to theta 1",
+		},
+		{
+			name:       "different seed",
+			data:       []byte{1, 2, 3, 4, 5},
+			theta:      theta.MaxTheta,
+			seed:       77777,
+			wantErrMsg: "",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, tc.theta, tc.seed, true)
+			hash, err := ht.HashBytesAndScreen(tc.data)
+
+			assert.False(t, ht.isEmpty)
+
+			if tc.wantErrMsg != "" {
+				assert.ErrorContains(t, err, tc.wantErrMsg)
+			} else {
+				assert.NotZero(t, hash, "Expected non-zero hash for data: %v", tc.data)
+			}
+		})
+	}
+}
+
+func TestHashtable_HashBytesAndScreenConsistency(t *testing.T) {
+	ht := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+	hash1, err := ht.HashBytesAndScreen([]byte{1, 2, 3, 4, 5})
+	assert.NoError(t, err)
+	hash2, err := ht.HashBytesAndScreen([]byte{1, 2, 3, 4, 5})
+	assert.NoError(t, err)
+	assert.Equal(t, hash1, hash2, "Same byte array should produce same hash")
+}
+
+func TestHashTable_Find(t *testing.T) {
+	sketch := newHashtable[*float64Summary](2, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+	e := entry[*float64Summary]{
+		Hash: uint64(12345),
+	}
+
+	// Find an empty table
+	index, err := sketch.Find(e.Hash)
+	assert.ErrorIs(t, err, ErrKeyNotFound)
+
+	sketch.entries[index] = e
+	sketch.numEntries++
+
+	// Find the inserted key
+	index2, err := sketch.Find(e.Hash)
+	assert.NoError(t, err)
+	assert.Equal(t, index, index2)
+
+	// Table is full
+	size := 1 << sketch.lgCurSize
+	for i := 0; i < size; i++ {
+		sketch.entries[i] = entry[*float64Summary]{
+			Hash: uint64(i + 1000),
+		}
+	}
+	sketch.numEntries = uint32(size)
+
+	index, err = sketch.Find(e.Hash)
+	assert.ErrorIs(t, err, ErrKeyNotFoundAndNoEmptySlots)
+}
+
+func TestHashtable_Insert(t *testing.T) {
+	t.Run("Without resizing & rebuilding", func(t *testing.T) {
+		sketch := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+		e := entry[*float64Summary]{
+			Hash: uint64(12345),
+		}
+		index, err := sketch.Find(e.Hash)
+		assert.ErrorIs(t, err, ErrKeyNotFound)
+
+		sketch.Insert(index, e)
+		assert.Equal(t, 1, int(sketch.numEntries))
+
+		index2, err := sketch.Find(e.Hash)
+		assert.NoError(t, err)
+		assert.Equal(t, sketch.entries[index2], e)
+	})
+
+	t.Run("With resizing", func(t *testing.T) {
+		lgCurSize := uint8(2)
+		lgNomSize := uint8(4)
+		sketch := newHashtable[*float64Summary](lgCurSize, lgNomSize, theta.ResizeX2, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+		initialSize := sketch.lgCurSize
+
+		insertedEntries := make([]entry[*float64Summary], 0)
+		numToInsert := 10 // Insert enough to trigger resize
+		for i := 0; i < numToInsert; i++ {
+			e := entry[*float64Summary]{
+				Hash: uint64(i + 1000),
+			}
+			index, err := sketch.Find(e.Hash)
+			if err == nil {
+				continue
+			}
+
+			sketch.Insert(index, e)
+			insertedEntries = append(insertedEntries, e)
+		}
+
+		assert.Greater(t, sketch.lgCurSize, initialSize, "Table should have been resized")
+		assert.Equal(t, numToInsert, len(insertedEntries), "Should have inserted all keys")
+
+		for _, insertedEntry := range insertedEntries {
+			index, err := sketch.Find(insertedEntry.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, insertedEntry, sketch.entries[index], "Key value should match")
+		}
+	})
+
+	t.Run("With rebuilding", func(t *testing.T) {
+		lgNomSize := uint8(3)
+		lgCurSize := uint8(4)
+		sketch := newHashtable[*float64Summary](lgCurSize, lgNomSize, theta.ResizeX2, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+		numToInsert := 100
+		insertedEntries := make([]entry[*float64Summary], 0)
+		rebuildOccurred := false
+
+		for i := 0; i < numToInsert; i++ {
+			e := entry[*float64Summary]{
+				Hash: uint64(i + 1000),
+			}
+			index, err := sketch.Find(e.Hash)
+			if err == nil {
+				continue
+			}
+			if index == -1 {
+				// Table is full, cannot insert more
+				break
+			}
+
+			prevTheta := sketch.theta
+			sketch.Insert(index, e)
+			insertedEntries = append(insertedEntries, e)
+
+			// Rebuild is detected when theta decreases
+			if sketch.theta < prevTheta {
+				rebuildOccurred = true
+				nominalSize := uint32(1 << lgNomSize)
+				assert.Equal(t, nominalSize, sketch.numEntries, "After rebuild, entries should equal nominal size")
+				assert.Less(t, sketch.theta, theta.MaxTheta, "Theta should decrease after rebuild")
+				break
+			}
+		}
+
+		assert.True(t, rebuildOccurred, "Rebuild should have occurred")
+
+		foundCount := 0
+		for _, insertedEntry := range insertedEntries {
+			index, err := sketch.Find(insertedEntry.Hash)
+			if err == nil && index >= 0 && sketch.entries[index] == insertedEntry {
+				foundCount++
+			}
+		}
+
+		assert.Greater(t, foundCount, 0, "Some entries should still be accessible after rebuild")
+	})
+}
+
+func TestHashtable_Trim(t *testing.T) {
+	t.Run("rebuild", func(t *testing.T) {
+		lgNomSize := uint8(3)
+		lgCurSize := uint8(5)
+		sketch := newHashtable[*float64Summary](lgCurSize, lgNomSize, theta.ResizeX2, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+		// Insert entries exceeding nominal size
+		numToInsert := 20
+		for i := 0; i < numToInsert; i++ {
+			e := entry[*float64Summary]{
+				Hash: uint64(i + 5000),
+			}
+			index, err := sketch.Find(e.Hash)
+			if err == nil {
+				continue
+			}
+
+			sketch.entries[index] = e
+			sketch.numEntries++
+		}
+
+		initialNumEntries := sketch.numEntries
+		nominalSize := uint32(1 << lgNomSize)
+
+		assert.Greater(t, initialNumEntries, nominalSize, "numEntries should exceed nominal size before Trim")
+
+		sketch.Trim()
+
+		assert.Equal(t, nominalSize, sketch.numEntries, "After Trim, numEntries should equal nominal size")
+		assert.Less(t, sketch.theta, theta.MaxTheta, "Theta should decrease after Trim")
+	})
+
+	t.Run("no op", func(t *testing.T) {
+		lgNomSize := uint8(4)
+		lgCurSize := uint8(4)
+		sketch := newHashtable[*float64Summary](lgCurSize, lgNomSize, theta.ResizeX2, 1.0, theta.MaxTheta, theta.DefaultSeed, true)
+
+		// Insert fewer entries than the nominal size
+		numToInsert := 5
+		for i := 0; i < numToInsert; i++ {
+			e := entry[*float64Summary]{
+				Hash: uint64(i + 6000),
+			}
+			index, err := sketch.Find(e.Hash)
+			if err == nil {
+				continue
+			}
+
+			sketch.entries[index] = e
+			sketch.numEntries++
+		}
+
+		initialNumEntries := sketch.numEntries
+		initialTheta := sketch.theta
+		nominalSize := uint32(1 << lgNomSize)
+
+		assert.Less(t, initialNumEntries, nominalSize, "numEntries should be less than nominal size")
+
+		sketch.Trim()
+
+		assert.Equal(t, initialNumEntries, sketch.numEntries, "numEntries should not change when less than nominal size")
+		assert.Equal(t, initialTheta, sketch.theta, "Theta should not change when entries <= nominal size")
+	})
+}
+
+func TestHashtable_Reset(t *testing.T) {
+	sketch := newHashtable[*float64Summary](4, 4, theta.ResizeX1, 0.5, theta.MaxTheta, theta.DefaultSeed, false)
+
+	sketch.entries[0] = entry[*float64Summary]{
+		Hash: uint64(100),
+	}
+	sketch.entries[5] = entry[*float64Summary]{
+		Hash: uint64(200),
+	}
+	sketch.numEntries = 2
+	sketch.isEmpty = false
+
+	sketch.Reset()
+
+	assert.True(t, sketch.isEmpty)
+	assert.Zero(t, sketch.numEntries)
+	// Verify all entries are zero
+	for i, e := range sketch.entries {
+		assert.Zero(t, e, "entry at index %d should be zero after reset", i)
+	}
+
+	expectedTheta := startingThetaFromP(sketch.p)
+	assert.Equal(t, expectedTheta, sketch.theta, "theta should be %d after reset", expectedTheta)
+}
diff --git a/tuple/sketch.go b/tuple/sketch.go
new file mode 100644
index 0000000..822fe62
--- /dev/null
+++ b/tuple/sketch.go
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tuple
+
+import (
+	"iter"
+)
+
+// Summary is the base interface for all summary types used in tuple sketches.
+// A summary holds aggregate data associated with each retained hash key.
+type Summary interface {
+	// Reset clears the content of the summary, restoring it to its initial state.
+	Reset()
+	// Clone creates and returns a deep copy of the current Summary instance.
+	Clone() Summary
+}
+
+// Sketch is the base interface for tuple sketches.
+// It extends Theta sketch to associate arbitrary summaries with each retained key.
+type Sketch[S Summary] interface {
+	// IsEmpty reports whether this sketch represents an empty set.
+	// Note: this is not the same as having no retained hashes.
+	IsEmpty() bool
+
+	// Estimate returns the estimated distinct count of the input stream.
+	Estimate() float64
+
+	// LowerBoundFromSubset returns the approximate lower error bound for
+	// the given number of standard deviations over a subset of retained hashes.
+	// numStdDevs specifies the confidence level (1, 2, or 3) corresponding to
+	// approximately 67%, 95%, or 99% confidence intervals.
+	// numSubsetEntries specifies number of items from {0, 1, ..., get_num_retained()}
+	// over which to estimate the bound.
+	LowerBoundFromSubset(numStdDevs uint8, numSubsetEntries uint32) (float64, error)
+
+	// LowerBound returns the approximate lower error bound for the given
+	// number of standard deviations. numStdDevs should be 1, 2, or 3 for
+	// approximately 67%, 95%, or 99% confidence intervals.
+	LowerBound(numStdDevs uint8) (float64, error)
+
+	// UpperBoundFromSubset returns the approximate upper error bound for
+	// the given number of standard deviations over a subset of retained hashes.
+	// numStdDevs specifies the confidence level (1, 2, or 3) corresponding to
+	// approximately 67%, 95%, or 99% confidence intervals.
+	// numSubsetEntries specifies number of items from {0, 1, ..., get_num_retained()}
+	// over which to estimate the bound.
+	UpperBoundFromSubset(numStdDevs uint8, numSubsetEntries uint32) (float64, error)
+
+	// UpperBound returns the approximate upper error bound for the given
+	// number of standard deviations. numStdDevs should be 1, 2, or 3 for
+	// approximately 67%, 95%, or 99% confidence intervals.
+	UpperBound(numStdDevs uint8) (float64, error)
+
+	// IsEstimationMode reports whether the sketch is in estimation mode,
+	// as opposed to exact mode.
+	IsEstimationMode() bool
+
+	// Theta returns theta as a fraction from 0 to 1, representing the
+	// effective sampling rate.
+	Theta() float64
+
+	// Theta64 returns theta as a positive integer between 0 and math.MaxUint64.
+	Theta64() uint64
+
+	// NumRetained returns the number of hashes retained in the sketch.
+	NumRetained() uint32
+
+	// SeedHash returns the hash of the seed used to hash the input.
+	SeedHash() (uint16, error)
+
+	// IsOrdered reports whether retained hashes are sorted by hash value.
+	IsOrdered() bool
+
+	// String returns a human-readable summary of this sketch.
+	// If printItems is true, the output includes all retained hashes.
+	String(shouldPrintItems bool) string
+
+	// All returns an iterator over all hash-summary pairs in the sketch.
+	All() iter.Seq2[uint64, S]
+}
diff --git a/tuple/testing.go b/tuple/testing.go
new file mode 100644
index 0000000..ff05207
--- /dev/null
+++ b/tuple/testing.go
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 tuple
+
+type int32Summary struct {
+	value int32
+}
+
+func (s *int32Summary) Reset() {
+	s.value = 0
+}
+
+func (s *int32Summary) Clone() Summary {
+	return &int32Summary{
+		value: s.value,
+	}
+}
+
+func (s *int32Summary) Update(value int32) {
+	s.value += value
+}
+
+func newInt32Summary() *int32Summary {
+	return &int32Summary{}
+}
+
+type float64Summary struct {
+	value float64
+}
+
+func (s *float64Summary) Reset() {
+	s.value = 0
+}
+
+func (s *float64Summary) Clone() Summary {
+	return &float64Summary{value: s.value}
+}
+
+func (s *float64Summary) Update(value float64) {
+	s.value += value
+}
+
+func newFloat64Summary() *float64Summary {
+	return &float64Summary{}
+}