Shrinkage Handling

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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

from dataclasses import dataclass
from typing import List, Optional, Tuple
from enum import Enum
 
class ShrinkagePolicy(Enum):
    """Different policies for triggering shrinkage."""
    THRESHOLD = "threshold"      # Merge when below fill percentage
    COMBINED = "combined"        # Both buckets together below threshold
    LOAD_FACTOR = "load_factor"  # Global load factor too low
    HYSTERESIS = "hysteresis"    # Different thresholds for grow/shrink
 
 
@dataclass
class UnderutilizationDetector:
    """
    Detect when buckets should be considered for merging.
    
    Key insight: Shrinkage thresholds should be lower than split
    thresholds to prevent oscillation (split-merge-split cycles).
    """
    
    bucket_capacity: int = 100
    
    # Threshold for single bucket underutilization
    single_bucket_threshold: float = 0.25  # 25% or less triggers check
    
    # Threshold for combined (merged) bucket viability
    combined_threshold: float = 0.70  # Merged bucket must not exceed 70%
    
    # Global load factor floor
    global_load_floor: float = 0.30  # Shrink when overall < 30%
    
    def check_single_bucket(self, record_count: int) -> bool:
        """
        Check if a single bucket is underutilized.
        
        Returns True if the bucket is sparse enough to consider merging.
        """
        fill_ratio = record_count / self.bucket_capacity
        return fill_ratio <= self.single_bucket_threshold
    
    def check_merge_viability(self, 
                               bucket_a_count: int, 
                               bucket_b_count: int) -> Tuple[bool, str]:
        """
        Check if two buckets can be safely merged.
        
        Returns (can_merge, reason)
        """
        combined_count = bucket_a_count + bucket_b_count
        
        if combined_count > self.bucket_capacity:
            ratio = combined_count / self.bucket_capacity
            return False, f"Would overflow ({ratio:.1%} of capacity)"
        
        combined_ratio = combined_count / self.bucket_capacity
        
        if combined_ratio > self.combined_threshold:
            return False, f"Would be too full ({combined_ratio:.1%} > {self.combined_threshold:.1%})"
        
        return True, f"Safe to merge ({combined_ratio:.1%} of capacity)"
    
    def check_global_underutilization(self, 
                                        total_records: int, 
                                        total_buckets: int) -> bool:
        """
        Check if the entire index is underutilized.
        
        Used by Linear Hashing-style global shrinkage.
        """
        if total_buckets == 0:
            return False
        
        load_factor = total_records / (total_buckets * self.bucket_capacity)
        return load_factor < self.global_load_floor
    
    def analyze_shrinkage_potential(self,
                                     buckets: List[int],  # record counts
                                     bucket_depths: List[int]) -> dict:
        """
        Analyze an index to find shrinkage opportunities.
        
        Returns statistics about underutilization and merge candidates.
        """
        underutilized = []
        merge_candidates = []
        
        total_records = sum(buckets)
        total_capacity = len(buckets) * self.bucket_capacity
        
        # Find underutilized buckets
        for i, count in enumerate(buckets):
            if self.check_single_bucket(count):
                underutilized.append(i)
        
        # Find merge candidates (buckets that could merge with sibling)
        # In practice, sibling identification depends on the scheme
        # Here we assume adjacent buckets with same depth are siblings
        for i in range(len(buckets) - 1):
            if bucket_depths[i] == bucket_depths[i + 1]:
                can_merge, _ = self.check_merge_viability(buckets[i], buckets[i + 1])
                if can_merge:
                    merge_candidates.append((i, i + 1))
        
        return {
            "total_buckets": len(buckets),
            "total_records": total_records,
            "overall_utilization": total_records / total_capacity if total_capacity else 0,
            "underutilized_buckets": len(underutilized),
            "underutilized_indices": underutilized,
            "merge_candidates": merge_candidates,
            "potential_space_savings": len(merge_candidates) * self.bucket_capacity,
            "global_shrinkage_recommended": self.check_global_underutilization(
                total_records, len(buckets)
            ),
        }
 
 
# Example usage
detector = UnderutilizationDetector(bucket_capacity=100)
 
# Analyze an underutilized index
bucket_counts = [15, 10, 5, 20, 25, 8, 12, 18]  # Very sparse
bucket_depths = [3, 3, 3, 3, 3, 3, 3, 3]  # All same depth
 
analysis = detector.analyze_shrinkage_potential(bucket_counts, bucket_depths)
 
print("Shrinkage Analysis:")
print(f"  Overall utilization: {analysis['overall_utilization']:.1%}")
print(f"  Underutilized buckets: {analysis['underutilized_buckets']} of {analysis['total_buckets']}")
print(f"  Merge candidates: {len(analysis['merge_candidates'])} pairs")
print(f"  Potential space savings: {analysis['potential_space_savings']} records worth")
print(f"  Global shrinkage recommended: {analysis['global_shrinkage_recommended']}")

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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253

from typing import List, Tuple, Optional
from dataclasses import dataclass, field
 
@dataclass
class MergeableBucket:
    """Bucket supporting merge operations."""
    id: int
    local_depth: int
    records: List[tuple] = field(default_factory=list)
    capacity: int = 100
    
    def can_absorb(self, other: 'MergeableBucket') -> bool:
        """Check if this bucket can absorb another's records."""
        return len(self.records) + len(other.records) <= self.capacity
    
    def get_sibling_prefix(self) -> int:
        """
        Calculate the prefix of our sibling bucket.
        
        If our prefix is P with local_depth bits, sibling has
        prefix P XOR (1 << (local_depth - 1))
        (flip the last bit of our prefix)
        """
        # This is conceptual - actual prefix comes from directory position
        return -1  # Placeholder
 
 
class BucketMerger:
    """
    Implements bucket merge operations for dynamic hashing.
    
    Merging is more complex than splitting because:
    1. Must find sibling bucket
    2. Must verify both buckets exist and are compatible
    3. Must atomically update directory and deallocate
    """
    
    def find_sibling(self, 
                     directory: List[MergeableBucket],
                     bucket_idx: int,
                     global_depth: int) -> Optional[Tuple[int, MergeableBucket]]:
        """
        Find the sibling bucket for a potential merge.
        
        The sibling shares all prefix bits except the last one
        used by the bucket's local depth.
        """
        bucket = directory[bucket_idx]
        
        if bucket.local_depth == 0:
            return None  # Can't merge at depth 0
        
        # Calculate sibling's directory index
        # Flip the bit at position (local_depth - 1) from the left
        flip_bit = 1 << (global_depth - bucket.local_depth)
        sibling_idx = bucket_idx ^ flip_bit
        
        if sibling_idx >= len(directory):
            return None
        
        sibling = directory[sibling_idx]
        
        # Sibling must have same local depth for merge
        if sibling.local_depth != bucket.local_depth:
            return None
        
        # Sibling must be a different bucket (not same bucket via shared pointers)
        if sibling.id == bucket.id:
            return None
            
        return sibling_idx, sibling
    
    def merge_buckets(self,
                      bucket: MergeableBucket,
                      sibling: MergeableBucket) -> MergeableBucket:
        """
        Merge two sibling buckets into one.
        
        All records from sibling move to bucket.
        Sibling becomes empty and can be deallocated.
        """
        if not bucket.can_absorb(sibling):
            raise ValueError("Cannot merge: combined records exceed capacity")
        
        # Move all records from sibling to bucket
        bucket.records.extend(sibling.records)
        sibling.records.clear()
        
        # Decrease local depth (we're using one fewer bit)
        bucket.local_depth -= 1
        
        print(f"Merged bucket {sibling.id} into bucket {bucket.id}")
        print(f"  New local depth: {bucket.local_depth}")
        print(f"  Total records: {len(bucket.records)}")
        
        return bucket
    
    def update_directory_for_merge(self,
                                    directory: List[MergeableBucket],
                                    surviving_bucket: MergeableBucket,
                                    empty_bucket: MergeableBucket):
        """
        Update directory entries after a merge.
        
        All entries pointing to the empty bucket should now
        point to the surviving bucket.
        """
        updates = 0
        for i in range(len(directory)):
            if directory[i].id == empty_bucket.id:
                directory[i] = surviving_bucket
                updates += 1
        
        print(f"  Updated {updates} directory entries")
        return updates
 
 
class ExtendibleHashWithShrinkage:
    """
    Extendible Hash Index with shrinkage support.
    
    Extends basic Extendible Hashing to support bucket merging
    and directory halving.
    """
    
    def __init__(self, bucket_capacity: int = 4):
        self.global_depth = 1
        self.bucket_capacity = bucket_capacity
        self.next_bucket_id = 0
        self.directory: List[MergeableBucket] = []
        self.merger = BucketMerger()
        self.shrink_threshold = 0.25  # Consider merge at 25% full
        self._init_directory()
    
    def _init_directory(self):
        for i in range(2 ** self.global_depth):
            bucket = MergeableBucket(
                id=self.next_bucket_id,
                local_depth=self.global_depth,
                capacity=self.bucket_capacity
            )
            self.next_bucket_id += 1
            self.directory.append(bucket)
    
    def delete(self, key: str, hash_value: int):
        """Delete a record and potentially trigger shrinkage."""
        idx = self._get_directory_index(hash_value)
        bucket = self.directory[idx]
        
        # Find and remove the record
        original_count = len(bucket.records)
        bucket.records = [r for r in bucket.records if r[0] != key]
        
        if len(bucket.records) == original_count:
            return  # Key not found
        
        # Check for shrinkage opportunity
        fill_ratio = len(bucket.records) / self.bucket_capacity
        if fill_ratio <= self.shrink_threshold:
            self._attempt_merge(idx, bucket)
    
    def _get_directory_index(self, hash_value: int) -> int:
        shift = 32 - self.global_depth
        return hash_value >> shift
    
    def _attempt_merge(self, bucket_idx: int, bucket: MergeableBucket):
        """Attempt to merge an underutilized bucket with its sibling."""
        result = self.merger.find_sibling(
            self.directory, bucket_idx, self.global_depth
        )
        
        if result is None:
            print(f"No sibling found for bucket {bucket.id}")
            return
        
        sibling_idx, sibling = result
        
        # Check if merge is viable
        if not bucket.can_absorb(sibling):
            combined = len(bucket.records) + len(sibling.records)
            print(f"Cannot merge: combined {combined} exceeds capacity {self.bucket_capacity}")
            return
        
        # Perform the merge
        print(f"
>>> Attempting merge of buckets {bucket.id} and {sibling.id}")
        self.merger.merge_buckets(bucket, sibling)
        self.merger.update_directory_for_merge(self.directory, bucket, sibling)
        
        # Check if directory can be halved
        self._attempt_directory_halve()
    
    def _attempt_directory_halve(self):
        """
        Attempt to halve the directory size.
        
        Can only halve if every pair of entries points to the same bucket
        (i.e., all buckets have local_depth < global_depth).
        """
        if self.global_depth <= 1:
            return  # Can't go below 2 entries
        
        # Check if all buckets have depth less than global depth
        for bucket in self.directory:
            if bucket.local_depth >= self.global_depth:
                return  # Can't halve yet
        
        # Verify that pairs point to same bucket
        new_size = len(self.directory) // 2
        for i in range(new_size):
            if self.directory[i].id != self.directory[i + new_size].id:
                return  # Pairs not identical
        
        # Safe to halve
        print(f"Halving directory: {len(self.directory)} -> {new_size}")
        self.directory = self.directory[:new_size]
        self.global_depth -= 1
 
 
# Demonstration
def demonstrate_shrinkage():
    """Show bucket merging in action."""
    
    index = ExtendibleHashWithShrinkage(bucket_capacity=4)
    
    # Insert some records
    test_data = [
        ("k1", 0b00110000_00000000_00000000_00000000),
        ("k2", 0b00110001_00000000_00000000_00000000),
        ("k3", 0b01110000_00000000_00000000_00000000),
        ("k4", 0b01110001_00000000_00000000_00000000),
    ]
    
    for key, hash_val in test_data:
        index.directory[hash_val >> 31].records.append((key, hash_val, 0))
    
    print("After insertions:")
    for i, b in enumerate(index.directory):
        print(f"  Dir[{i}] -> Bucket {b.id} ({len(b.records)} records)")
    
    # Delete records to trigger shrinkage
    print("
Deleting k1, k2, k3...")
    for key, hash_val in test_data[:3]:
        index.delete(key, hash_val)
    
    print("
After deletions:")
    for i, b in enumerate(index.directory):
        print(f"  Dir[{i}] -> Bucket {b.id} ({len(b.records)} records)")
 
 
demonstrate_shrinkage()

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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

from typing import List
from dataclasses import dataclass, field
 
@dataclass
class ContractibleBucket:
    """Bucket for Linear Hashing with contraction support."""
    records: List[tuple] = field(default_factory=list)
    capacity: int = 100
 
 
class LinearHashWithContraction:
    """
    Linear Hashing with both expansion and contraction.
    
    Key insight: Contraction is splitting in reverse.
    The most recently added bucket merges first.
    """
    
    def __init__(self, initial_buckets: int = 4,
                 bucket_capacity: int = 4,
                 expand_threshold: float = 0.80,
                 contract_threshold: float = 0.30):
        
        self.level = 0
        while (1 << self.level) < initial_buckets:
            self.level += 1
        
        initial = 1 << self.level
        self.buckets: List[ContractibleBucket] = [
            ContractibleBucket(capacity=bucket_capacity)
            for _ in range(initial)
        ]
        
        self.split_pointer = 0
        self.bucket_capacity = bucket_capacity
        self.expand_threshold = expand_threshold
        self.contract_threshold = contract_threshold
        self.total_records = 0
        self.min_buckets = 2  # Never shrink below this
    
    def _load_factor(self) -> float:
        """Current load factor."""
        capacity = len(self.buckets) * self.bucket_capacity
        return self.total_records / capacity if capacity else 0
    
    def delete(self, key: str, hash_value: int):
        """Delete a record and potentially trigger contraction."""
        idx = self._get_bucket_index(hash_value)
        bucket = self.buckets[idx]
        
        # Find and remove
        original_count = len(bucket.records)
        bucket.records = [r for r in bucket.records if r[0] != key]
        
        if len(bucket.records) < original_count:
            self.total_records -= 1
            
            # Check for contraction
            if self._load_factor() < self.contract_threshold:
                self._contract()
    
    def _get_bucket_index(self, hash_value: int) -> int:
        """Determine bucket for a key."""
        idx = hash_value & ((1 << self.level) - 1)
        if idx < self.split_pointer:
            idx = hash_value & ((1 << (self.level + 1)) - 1)
        return idx
    
    def _contract(self):
        """
        Remove the last bucket, merging its contents.
        
        The last bucket was created when split_pointer reached
        its current position. Contraction reverses this.
        """
        if len(self.buckets) <= self.min_buckets:
            print("Cannot contract: at minimum bucket count")
            return
        
        # Identify the bucket to remove and its merge partner
        last_bucket_idx = len(self.buckets) - 1
        last_bucket = self.buckets[last_bucket_idx]
        
        # Partner is at index: last_bucket_idx - 2^level
        # (where it was before the split that created this bucket)
        partner_idx = last_bucket_idx - (1 << self.level)
        
        if partner_idx < 0:
            # We're at the start of a new level - go back
            self.level -= 1
            self.split_pointer = (1 << self.level) - 1
            partner_idx = last_bucket_idx - (1 << self.level)
        
        partner = self.buckets[partner_idx]
        
        print(f">>> Contracting: merging bucket {last_bucket_idx} into {partner_idx}")
        print(f"    Level: {self.level}, Split pointer before: {self.split_pointer}")
        
        # Move all records from last bucket to partner
        partner.records.extend(last_bucket.records)
        
        # Remove the last bucket
        self.buckets.pop()
        
        # Update split pointer
        if self.split_pointer > 0:
            self.split_pointer -= 1
        elif self.level > 0:
            self.level -= 1
            self.split_pointer = (1 << self.level) - 1
        
        print(f"    Split pointer after: {self.split_pointer}")
        print(f"    Buckets remaining: {len(self.buckets)}")
    
    def display_state(self):
        """Display current state."""
        print(f"
=== Linear Hash State ===")
        print(f"Level: {self.level}, Split pointer: {self.split_pointer}")
        print(f"Buckets: {len(self.buckets)}")
        print(f"Records: {self.total_records}")
        print(f"Load factor: {self._load_factor():.2%}")
        
        for i, b in enumerate(self.buckets):
            status = "(already split)" if i < self.split_pointer else ""
            print(f"  Bucket {i} {status}: {len(b.records)} records")
 
 
# Demonstration
def demonstrate_linear_contraction():
    """Show Linear Hashing expansion followed by contraction."""
    
    index = LinearHashWithContraction(
        initial_buckets=2,
        bucket_capacity=2,
        expand_threshold=0.75,
        contract_threshold=0.25
    )
    
    # Insert records to cause expansion
    print("=== Expansion Phase ===")
    import random
    random.seed(42)
    
    keys = []
    for i in range(8):
        key = f"key_{i}"
        hash_val = random.getrandbits(32)
        keys.append((key, hash_val))
        
        idx = index._get_bucket_index(hash_val)
        index.buckets[idx].records.append((key, hash_val, i))
        index.total_records += 1
        
        if index._load_factor() > index.expand_threshold:
            # Simulate expansion (simplified)
            index.buckets.append(ContractibleBucket(capacity=2))
            index.split_pointer += 1
            if index.split_pointer >= (1 << index.level):
                index.level += 1
                index.split_pointer = 0
    
    index.display_state()
    
    # Delete records to cause contraction
    print("
=== Contraction Phase ===")
    for key, hash_val in keys[:6]:  # Delete most records
        index.delete(key, hash_val)
    
    index.display_state()
 
 
demonstrate_linear_contraction()