Hashing and Hash Based Data Structures Questions

import java.util.Arrays;
import java.util.Objects;

public class LinearProbingHashSet<E> {
    private Object[] table;
    private byte[] state; // 0 EMPTY, 1 OCCUPIED, 2 TOMBSTONE
    private int capacity;
    private int size;    // number of OCCUPIED entries
    private int used;    // OCCUPIED + TOMBSTONE
    private static final double LOAD_FACTOR = 0.6;

    public LinearProbingHashSet(int initialCap) {
        capacity = Math.max(8, Integer.highestOneBit(initialCap - 1) << 1);
        table = new Object[capacity];
        state = new byte[capacity];
    }

    public LinearProbingHashSet() { this(16); }

    private int indexFor(Object key) {
        int h = Objects.hashCode(key);
        h ^= (h >>> 16);
        return (h & 0x7fffffff) % capacity;
    }

    private int findSlot(Object key) {
        int idx = indexFor(key);
        int firstTombstone = -1;
        for (int i = 0; i < capacity; i++) {
            int pos = (idx + i) % capacity;
            if (state[pos] == 0) { // EMPTY
                return firstTombstone != -1 ? firstTombstone : pos;
            } else if (state[pos] == 2) { // TOMBSTONE
                if (firstTombstone == -1) firstTombstone = pos;
            } else {
                if (Objects.equals(table[pos], key)) return pos;
            }
        }
        return firstTombstone != -1 ? firstTombstone : -1;
    }

    public boolean contains(E key) {
        int idx = indexFor(key);
        for (int i = 0; i < capacity; i++) {
            int pos = (idx + i) % capacity;
            if (state[pos] == 0) return false;
            if (state[pos] == 1 && Objects.equals(table[pos], key)) return true;
        }
        return false;
    }

    public boolean add(E key) {
        if ((used + 1) > capacity * LOAD_FACTOR) resize(capacity * 2);
        int slot = findSlot(key);
        if (slot == -1) throw new IllegalStateException("Table full");
        if (state[slot] == 1 && Objects.equals(table[slot], key)) return false; // already present
        boolean wasTomb = state[slot] == 2;
        table[slot] = key;
        state[slot] = 1;
        size++;
        if (!wasTomb) used++;
        return true;
    }

    public boolean remove(E key) {
        int idx = indexFor(key);
        for (int i = 0; i < capacity; i++) {
            int pos = (idx + i) % capacity;
            if (state[pos] == 0) return false;
            if (state[pos] == 1 && Objects.equals(table[pos], key)) {
                table[pos] = null;
                state[pos] = 2; // tombstone
                size--;
                // Optional: if tombstones dominate, rehash to reduce clustering
                if (used > capacity * 0.75) resize(capacity); // rehash in place to clean tombstones
                return true;
            }
        }
        return false;
    }

    private void resize(int newCap) {
        int target = Math.max(8, Integer.highestOneBit(newCap - 1) << 1);
        Object[] oldTable = table;
        byte[] oldState = state;
        table = new Object[target];
        state = new byte[target];
        int oldCap = capacity;
        capacity = target;
        size = 0;
        used = 0;
        for (int i = 0; i < oldCap; i++) {
            if (oldState[i] == 1) add((E) oldTable[i]);
        }
    }

    public int size() { return size; }
}

def rabin_karp_all(text: str, pattern: str):
    """Return list of start indices where pattern occurs in text."""
    n, m = len(text), len(pattern)
    if m == 0:
        return list(range(n + 1))
    if m > n:
        return []

    # Parameters: base and large prime modulus
    base = 256                       # alphabet size (bytes)
    mod = 2**61 - 1                  # large Mersenne prime reduces collisions

    # Precompute base^(m-1) % mod for rolling
    def mod_mul(a, b):
        return (a * b) % mod

    power = pow(base, m - 1, mod)

    # compute initial hashes
    h_p = 0
    h_t = 0
    for i in range(m):
        h_p = (h_p * base + ord(pattern[i])) % mod
        h_t = (h_t * base + ord(text[i])) % mod

    res = []
    for i in range(n - m + 1):
        if h_t == h_p:
            # verify to avoid false positive from collision
            if text[i:i + m] == pattern:
                res.append(i)
        if i < n - m:
            # remove leading char, add next char
            left = (ord(text[i]) * power) % mod
            h_t = (h_t - left) % mod
            h_t = (h_t * base + ord(text[i + m])) % mod

    return res

val labels = spark.read.parquet("labels")
val features = spark.read.parquet("features")
if (labelsSizeBytes < BROADCAST_THRESHOLD) {
  val bLabels = broadcast(labels)
  features.join(bLabels, "key")
} else {
  val p = optimalPartitions
  val fPart = features.repartition(p, col("key"))
  val lPart = labels.repartition(p, col("key"))
  fPart.join(lPart, "key")
}