1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import math
 
def analyze_load_factor(alpha: float) -> dict:
    """
    Analyze expected performance metrics for a given load factor.
    """
    # Expected probes for successful lookup (average chain position)
    successful_probes = (1 + alpha) / 2
    
    # Expected probes for unsuccessful lookup (full chain)
    unsuccessful_probes = alpha
    
    # Probability of bucket having more than k records
    def prob_overflow(k: int) -> float:
        cumulative = sum(
            (alpha**i * math.exp(-alpha)) / math.factorial(i)
            for i in range(k + 1)
        )
        return 1 - cumulative
    
    return {
        "load_factor": alpha,
        "expected_chain_length": alpha,
        "successful_lookup_probes": successful_probes,
        "unsuccessful_lookup_probes": unsuccessful_probes,
        "prob_empty_bucket": math.exp(-alpha),
        "prob_overflow_3": prob_overflow(3),
        "prob_overflow_5": prob_overflow(5),
        "space_utilization": min(alpha, 1.0)  # Approximate
    }
 
# Example analysis
for alpha in [0.5, 0.7, 0.8, 1.0, 1.5]:
    metrics = analyze_load_factor(alpha)
    print(f"α={alpha}: {metrics['successful_lookup_probes']:.2f} probes, "
          f"{metrics['prob_overflow_3']*100:.1f}% overflow >3")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

def analyze_io_cost(
    num_records: int,
    page_size: int = 4096,
    record_size: int = 100,
    load_factor: float = 0.75
) -> dict:
    """
    Analyze expected I/O costs for disk-based linear hashing.
    """
    # Bucket capacity (records per primary page)
    header_size = 64  # Page header
    records_per_page = (page_size - header_size) // record_size
    
    # Number of buckets needed
    num_buckets = int(num_records / (load_factor * records_per_page))
    
    # Expected records per bucket
    records_per_bucket = num_records / num_buckets
    
    # Expected overflow pages per bucket
    if records_per_bucket <= records_per_page:
        overflow_per_bucket = 0
    else:
        overflow_per_bucket = (records_per_bucket - records_per_page) / records_per_page
    
    # I/O costs
    lookup_io = 1 + overflow_per_bucket
    insert_io = lookup_io  # Find bucket + possible write
    
    # Split I/O: read old bucket + write old + write new
    split_io = 3 + 2 * max(0, overflow_per_bucket)
    
    return {
        "num_buckets": num_buckets,
        "records_per_bucket": records_per_bucket,
        "records_per_page": records_per_page,
        "overflow_pages_per_bucket": overflow_per_bucket,
        "expected_lookup_io": lookup_io,
        "expected_insert_io": insert_io,
        "split_io": split_io,
        "total_pages": num_buckets * (1 + overflow_per_bucket),
        "space_utilization": num_records * record_size / (num_buckets * (1 + overflow_per_bucket) * page_size)
    }
 
# Example: 1 million records
analysis = analyze_io_cost(1_000_000)
print(f"1M records: {analysis['expected_lookup_io']:.2f} I/Os per lookup")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

class TunedLinearHash:
    """
    Linear hash table with tunable parameters.
    """
    
    def __init__(
        self,
        initial_buckets: int = 64,      # Start larger for known large datasets
        target_load_factor: float = 0.75,
        min_load_factor: float = 0.25,   # Trigger contract if too sparse
        bucket_capacity: int = 50,       # Match to page size
    ):
        self.N0 = initial_buckets
        self.L = 0
        self.S = 0
        self.buckets = [[] for _ in range(initial_buckets)]
        
        self.target_load = target_load_factor
        self.min_load = min_load_factor
        self.capacity = bucket_capacity
        
        # Performance tracking
        self.total_probes = 0
        self.total_lookups = 0
    
    def _should_split(self) -> bool:
        """Trigger split based on global load factor."""
        load = self._record_count() / len(self.buckets)
        return load > self.target_load
    
    def _should_contract(self) -> bool:
        """
        Consider contraction if highly underutilized.
        Note: Linear hashing doesn't traditionally contract,
        but some implementations support it.
        """
        if self.L == 0 and self.S == 0:
            return False  # Can't contract below initial size
        
        load = self._record_count() / len(self.buckets)
        return load < self.min_load
    
    def tune_for_read_heavy(self):
        """Optimize for read-heavy workload."""
        self.target_load = 0.60  # Lower load = shorter chains
        # Proactively split to reduce chains
        while self._load_factor() > self.target_load:
            self._perform_split()
    
    def tune_for_write_heavy(self):
        """Optimize for write-heavy workload."""
        self.target_load = 0.85  # Higher load = fewer splits
        # Accept slightly longer chains for fewer split operations
    
    def tune_for_memory(self):
        """Optimize for memory efficiency."""
        self.target_load = 0.95  # Maximize utilization
        self.min_load = 0.40     # Enable contraction
    
    def get_performance_metrics(self) -> dict:
        """Return current performance statistics."""
        bucket_sizes = [len(b) for b in self.buckets]
        
        return {
            "load_factor": self._load_factor(),
            "bucket_count": len(self.buckets),
            "avg_chain_length": sum(bucket_sizes) / len(bucket_sizes),
            "max_chain_length": max(bucket_sizes),
            "empty_buckets": sum(1 for s in bucket_sizes if s == 0),
            "overflow_buckets": sum(1 for s in bucket_sizes if s > self.capacity),
            "avg_probes_per_lookup": self.total_probes / max(1, self.total_lookups)
        }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

import time
import random
import statistics
 
def benchmark_linear_hash(
    hash_table,
    num_records: int = 1_000_000,
    read_ratio: float = 0.8,
    num_operations: int = 100_000
) -> dict:
    """
    Comprehensive benchmark of linear hash table.
    
    Returns performance metrics dictionary.
    """
    # Phase 1: Bulk load
    start_load = time.perf_counter()
    for i in range(num_records):
        hash_table.insert(f"key_{i}", f"value_{i}")
    load_time = time.perf_counter() - start_load
    
    # Phase 2: Mixed workload
    keys = [f"key_{random.randint(0, num_records-1)}" for _ in range(num_operations)]
    operations = [random.random() < read_ratio for _ in range(num_operations)]
    
    latencies = []
    for i, (key, is_read) in enumerate(zip(keys, operations)):
        start = time.perf_counter()
        if is_read:
            hash_table.lookup(key)
        else:
            hash_table.insert(f"new_key_{i}", f"new_value_{i}")
        latencies.append(time.perf_counter() - start)
    
    # Compute statistics
    latencies.sort()
    
    return {
        "bulk_load_time_sec": load_time,
        "load_throughput_ops_sec": num_records / load_time,
        "mixed_throughput_ops_sec": num_operations / sum(latencies),
        "latency_p50_us": latencies[int(0.50 * len(latencies))] * 1e6,
        "latency_p90_us": latencies[int(0.90 * len(latencies))] * 1e6,
        "latency_p99_us": latencies[int(0.99 * len(latencies))] * 1e6,
        "latency_avg_us": statistics.mean(latencies) * 1e6,
        "latency_stdev_us": statistics.stdev(latencies) * 1e6,
        "final_load_factor": hash_table._load_factor(),
        "final_bucket_count": len(hash_table.buckets)
    }