Dynamic Hashing

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

from dataclasses import dataclass
from typing import List
import statistics
 
@dataclass
class HashIndexMetrics:
    """
    Comprehensive metrics for hash index health monitoring.
    
    These metrics should be collected periodically and trended
    to detect degradation before it impacts query performance.
    """
    
    # Configuration
    bucket_capacity: int = 100
    
    # Current state
    total_buckets: int = 0
    total_records: int = 0
    bucket_record_counts: List[int] = None
    overflow_chain_lengths: List[int] = None
    
    # Operation counts (since last reset)
    lookups: int = 0
    lookup_ios: int = 0
    insertions: int = 0
    splits: int = 0
    merges: int = 0
    
    def __post_init__(self):
        if self.bucket_record_counts is None:
            self.bucket_record_counts = []
        if self.overflow_chain_lengths is None:
            self.overflow_chain_lengths = []
    
    @property
    def load_factor(self) -> float:
        """Overall load factor: records / (buckets * capacity)."""
        total_capacity = self.total_buckets * self.bucket_capacity
        return self.total_records / total_capacity if total_capacity else 0
    
    @property
    def average_chain_length(self) -> float:
        """Average overflow chain length across all buckets."""
        if not self.overflow_chain_lengths:
            return 0
        return statistics.mean(self.overflow_chain_lengths)
    
    @property
    def max_chain_length(self) -> int:
        """Longest overflow chain (worst case lookup)."""
        return max(self.overflow_chain_lengths) if self.overflow_chain_lengths else 0
    
    @property
    def bucket_utilization_variance(self) -> float:
        """Standard deviation of bucket utilization."""
        if not self.bucket_record_counts:
            return 0
        utilizations = [c / self.bucket_capacity for c in self.bucket_record_counts]
        return statistics.stdev(utilizations) if len(utilizations) > 1 else 0
    
    @property
    def io_per_lookup(self) -> float:
        """Average I/Os per lookup operation."""
        return self.lookup_ios / self.lookups if self.lookups else 0
    
    def assess_health(self) -> dict:
        """
        Comprehensive health assessment.
        
        Returns dict with status and recommendations.
        """
        issues = []
        warnings = []
        
        # Check load factor
        lf = self.load_factor
        if lf > 0.95:
            issues.append(f"Critical: Load factor {lf:.1%} - immediate expansion needed")
        elif lf > 0.90:
            warnings.append(f"Warning: Load factor {lf:.1%} - approaching capacity")
        elif lf < 0.20:
            issues.append(f"Critical: Load factor {lf:.1%} - severe underutilization")
        elif lf < 0.30:
            warnings.append(f"Warning: Load factor {lf:.1%} - consider shrinkage")
        
        # Check chain lengths
        avg_chain = self.average_chain_length
        max_chain = self.max_chain_length
        
        if max_chain > 10:
            issues.append(f"Critical: Max chain length {max_chain} - hash function issue?")
        elif max_chain > 5:
            warnings.append(f"Warning: Max chain length {max_chain} - investigate distribution")
        
        if avg_chain > 5:
            issues.append(f"Critical: Average chain {avg_chain:.1f} - widespread overflow")
        elif avg_chain > 2:
            warnings.append(f"Warning: Average chain {avg_chain:.1f} - overflow building")
        
        # Check variance
        variance = self.bucket_utilization_variance
        if variance > 0.40:
            issues.append(f"Critical: Utilization variance {variance:.1%} - severe skew")
        elif variance > 0.25:
            warnings.append(f"Warning: Utilization variance {variance:.1%} - uneven distribution")
        
        # Check I/O efficiency
        io_avg = self.io_per_lookup
        if io_avg > 5:
            issues.append(f"Critical: {io_avg:.1f} I/Os per lookup - major degradation")
        elif io_avg > 3:
            warnings.append(f"Warning: {io_avg:.1f} I/Os per lookup - performance degrading")
        
        status = "HEALTHY"
        if issues:
            status = "CRITICAL"
        elif warnings:
            status = "WARNING"
        
        return {
            "status": status,
            "issues": issues,
            "warnings": warnings,
            "metrics": {
                "load_factor": f"{lf:.1%}",
                "avg_chain_length": f"{avg_chain:.2f}",
                "max_chain_length": max_chain,
                "utilization_variance": f"{variance:.1%}",
                "io_per_lookup": f"{io_avg:.2f}",
            },
        }
 
 
# Example health check
metrics = HashIndexMetrics(
    bucket_capacity=100,
    total_buckets=100,
    total_records=9200,  # 92% load factor
    bucket_record_counts=[92] * 80 + [130] * 20,  # Some overflow
    overflow_chain_lengths=[0] * 80 + [3] * 20,
    lookups=10000,
    lookup_ios=12000,  # 1.2 I/Os per lookup average
)
 
health = metrics.assess_health()
print(f"Status: {health['status']}")
print("Issues:", health['issues'])
print("Warnings:", health['warnings'])
print("Metrics:", health['metrics'])

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

from typing import List
import statistics
import random
 
def calculate_balance_metrics(bucket_counts: List[int]) -> dict:
    """
    Calculate load balance metrics for a hash index.
    
    A perfectly balanced index has CV = 0.
    CV < 0.3 is generally acceptable.
    CV > 0.5 indicates significant imbalance.
    """
    if not bucket_counts:
        return {"error": "No buckets"}
    
    mean = statistics.mean(bucket_counts)
    stdev = statistics.stdev(bucket_counts) if len(bucket_counts) > 1 else 0
    cv = stdev / mean if mean > 0 else 0
    
    # Gini coefficient (0 = perfect equality, 1 = complete inequality)
    sorted_counts = sorted(bucket_counts)
    n = len(sorted_counts)
    cumulative = 0
    total = sum(sorted_counts)
    lorenz = []
    for count in sorted_counts:
        cumulative += count
        lorenz.append(cumulative / total if total > 0 else 0)
    
    gini = 1 - 2 * sum(lorenz) / n if n > 0 else 0
    
    return {
        "mean_records": mean,
        "stdev": stdev,
        "coefficient_of_variation": cv,
        "gini_coefficient": gini,
        "min_count": min(bucket_counts),
        "max_count": max(bucket_counts),
        "empty_buckets": sum(1 for c in bucket_counts if c == 0),
        "assessment": "Excellent" if cv < 0.2 else 
                     "Good" if cv < 0.3 else 
                     "Acceptable" if cv < 0.5 else 
                     "Poor",
    }
 
 
def simulate_hash_distributions():
    """Compare different hash quality scenarios."""
    
    scenarios = {
        "Perfect hash": [100] * 100,  # Exactly 100 records per bucket
        "Good hash": [random.gauss(100, 10) for _ in range(100)],
        "Mediocre hash": [random.gauss(100, 30) for _ in range(100)],
        "Skewed data": [200] * 20 + [50] * 80,  # Some hot buckets
        "Poor hash": [random.expovariate(0.01) for _ in range(100)],
    }
    
    print(f"{'Scenario':<20} {'CV':<8} {'Gini':<8} {'Assessment':<12}")
    print("-" * 50)
    
    for name, counts in scenarios.items():
        # Ensure non-negative integer counts
        counts = [max(0, int(c)) for c in counts]
        metrics = calculate_balance_metrics(counts)
        print(f"{name:<20} {metrics['coefficient_of_variation']:.3f}   "
              f"{metrics['gini_coefficient']:.3f}   {metrics['assessment']:<12}")
 
 
simulate_hash_distributions()
 
# Output:
# Scenario             CV       Gini     Assessment  
# --------------------------------------------------
# Perfect hash         0.000    0.000    Excellent   
# Good hash            0.095    0.054    Excellent   
# Mediocre hash        0.289    0.163    Good        
# Skewed data          0.577    0.280    Poor        
# Poor hash            1.023    0.502    Poor

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

from dataclasses import dataclass, field
from typing import List, Optional
import math
 
@dataclass
class OverflowAnalyzer:
    """
    Analyze and manage overflow chains in hash indexes.
    
    Overflow analysis helps determine when intervention is needed
    and what type of intervention is most appropriate.
    """
    
    bucket_capacity: int = 100
    max_acceptable_chain: int = 3
    
    def analyze_bucket(self, primary_count: int, 
                       overflow_pages: List[int]) -> dict:
        """
        Analyze overflow situation for a single bucket.
        
        Args:
            primary_count: Records in primary bucket page
            overflow_pages: Record counts in each overflow page
        """
        total_records = primary_count + sum(overflow_pages)
        chain_length = len(overflow_pages)
        
        # Calculate expected I/Os for a random lookup
        if total_records == 0:
            expected_ios = 1  # Just check primary
        else:
            # Probability of finding record in each page
            p_primary = primary_count / total_records
            expected_ios = 1  # Always read primary
            
            cumulative_prob = p_primary
            for i, overflow_count in enumerate(overflow_pages):
                if overflow_count > 0:
                    expected_ios += (1 - cumulative_prob)  # May need this page
                    cumulative_prob += overflow_count / total_records
        
        return {
            "total_records": total_records,
            "chain_length": chain_length,
            "expected_ios": expected_ios,
            "primary_utilization": primary_count / self.bucket_capacity,
            "needs_attention": chain_length > self.max_acceptable_chain,
            "recommendation": self._recommend_action(chain_length, primary_count),
        }
    
    def _recommend_action(self, chain_length: int, primary_count: int) -> str:
        """Recommend action based on overflow state."""
        if chain_length == 0:
            return "No action needed"
        elif chain_length <= 2:
            return "Monitor - acceptable overflow"
        elif chain_length <= 5:
            return "Consider splitting or lowering load factor"
        else:
            return "Critical - immediate intervention required"
    
    def calculate_expected_chain_length(self, 
                                          records: int, 
                                          buckets: int) -> float:
        """
        Calculate expected chain length given records and buckets.
        
        Uses Poisson approximation for overflow probability.
        """
        if buckets == 0:
            return float('inf')
        
        avg_per_bucket = records / buckets
        
        # Probability that a bucket overflows (more than capacity)
        # Using Poisson approximation: P(X > k) where X ~ Poisson(λ)
        lambda_param = avg_per_bucket
        
        # Sum of P(X = j) for j = 0 to capacity
        p_no_overflow = sum(
            (lambda_param ** j) * math.exp(-lambda_param) / math.factorial(j)
            for j in range(self.bucket_capacity + 1)
        )
        p_overflow = 1 - p_no_overflow
        
        # Expected extra pages if overflow occurs
        expected_excess = max(0, avg_per_bucket - self.bucket_capacity)
        expected_chain = expected_excess / self.bucket_capacity if expected_excess > 0 else 0
        
        return expected_chain
    
    def recommend_capacity(self, target_records: int, 
                           max_chain: float = 0.5) -> dict:
        """
        Recommend bucket count to achieve maximum chain length target.
        """
        # Binary search for optimal bucket count
        low, high = target_records // self.bucket_capacity, target_records
        
        while low < high:
            mid = (low + high) // 2
            expected = self.calculate_expected_chain_length(target_records, mid)
            
            if expected > max_chain:
                low = mid + 1
            else:
                high = mid
        
        return {
            "recommended_buckets": low,
            "expected_chain_length": self.calculate_expected_chain_length(target_records, low),
            "load_factor": target_records / (low * self.bucket_capacity),
            "target_records": target_records,
        }
 
 
# Example usage
analyzer = OverflowAnalyzer(bucket_capacity=100, max_acceptable_chain=3)
 
# Analyze a bucket with overflow
result = analyzer.analyze_bucket(
    primary_count=100,
    overflow_pages=[80, 45, 12]  # 3 overflow pages
)
print("Bucket Analysis:")
print(f"  Total records: {result['total_records']}")
print(f"  Chain length: {result['chain_length']}")
print(f"  Expected I/Os: {result['expected_ios']:.2f}")
print(f"  Recommendation: {result['recommendation']}")
 
# Capacity planning
plan = analyzer.recommend_capacity(target_records=100000, max_chain=0.5)
print(f"
Capacity Planning for 100,000 records:")
print(f"  Recommended buckets: {plan['recommended_buckets']}")
print(f"  Expected load factor: {plan['load_factor']:.1%}")

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

from typing import List, Dict, Callable
import hashlib
 
class SkewMitigation:
    """
    Strategies for handling skewed data in hash indexes.
    
    Skew mitigation is crucial for maintaining O(1) performance
    when data doesn't follow uniform distribution assumptions.
    """
    
    @staticmethod
    def salted_hash(key: str, salt: int = 0) -> int:
        """
        Add salt to key before hashing to redistribute.
        
        Useful when a few keys dominate - create virtual buckets
        by salting with partition number.
        """
        salted_key = f"{key}:{salt}"
        return int(hashlib.md5(salted_key.encode()).hexdigest(), 16)
    
    @staticmethod
    def composite_key(primary: str, secondary: str) -> str:
        """
        Combine skewed key with unique identifier.
        
        For example, if querying by "country" (skewed):
        composite_key("USA", "user_12345") distributes better
        than just hashing "USA".
        """
        return f"{primary}:{secondary}"
    
    @staticmethod
    def adaptive_bucket_sizing(access_counts: Dict[int, int],
                                base_capacity: int = 100) -> Dict[int, int]:
        """
        Assign different capacities to hot vs cold buckets.
        
        Hot buckets get larger capacity to reduce overflow.
        This is an advanced technique requiring dynamic page sizing.
        """
        if not access_counts:
            return {}
        
        avg_access = sum(access_counts.values()) / len(access_counts)
        
        capacities = {}
        for bucket_id, access_count in access_counts.items():
            # Scale capacity based on relative access frequency
            relative_heat = access_count / avg_access if avg_access > 0 else 1
            
            # Hot buckets get up to 3x capacity
            multiplier = min(3.0, max(0.5, relative_heat))
            capacities[bucket_id] = int(base_capacity * multiplier)
        
        return capacities
    
    @staticmethod
    def split_hot_buckets(bucket_counts: List[int],
                          overflow_lengths: List[int],
                          threshold: float = 2.0) -> List[int]:
        """
        Identify buckets that should be split regardless of global policy.
        
        Returns list of bucket indices that are significantly hotter
        than average and should be split proactively.
        """
        if not bucket_counts:
            return []
        
        avg_count = sum(bucket_counts) / len(bucket_counts)
        avg_overflow = sum(overflow_lengths) / len(overflow_lengths) if overflow_lengths else 0
        
        hot_buckets = []
        for i, (count, overflow) in enumerate(zip(bucket_counts, overflow_lengths)):
            # Consider hot if significantly above average in either metric
            count_ratio = count / avg_count if avg_count > 0 else 0
            overflow_ratio = overflow / max(1, avg_overflow)
            
            if count_ratio > threshold or overflow_ratio > threshold:
                hot_buckets.append(i)
        
        return hot_buckets
 
 
class ConsistentHashing:
    """
    Consistent hashing for extreme skew tolerance.
    
    Instead of mod-based bucket assignment, use a ring where
    buckets own ranges. Allows fine-grained rebalancing.
    """
    
    def __init__(self, num_buckets: int, virtual_nodes: int = 100):
        """
        Create consistent hash ring.
        
        Virtual nodes improve balance - each physical bucket
        owns multiple points on the ring.
        """
        self.ring: Dict[int, int] = {}  # hash_point -> bucket_id
        
        for bucket_id in range(num_buckets):
            for v in range(virtual_nodes):
                # Create virtual node hash
                virtual_key = f"bucket_{bucket_id}_vnode_{v}"
                hash_point = int(hashlib.md5(virtual_key.encode()).hexdigest(), 16)
                self.ring[hash_point] = bucket_id
        
        self.sorted_hashes = sorted(self.ring.keys())
    
    def get_bucket(self, key: str) -> int:
        """Find bucket for a key using consistent hashing."""
        key_hash = int(hashlib.md5(key.encode()).hexdigest(), 16)
        
        # Binary search for next hash point
        import bisect
        idx = bisect.bisect_left(self.sorted_hashes, key_hash)
        
        if idx >= len(self.sorted_hashes):
            idx = 0  # Wrap around
        
        hash_point = self.sorted_hashes[idx]
        return self.ring[hash_point]
    
    def add_bucket(self, bucket_id: int, virtual_nodes: int = 100):
        """Add a new bucket to the ring (minimal data movement)."""
        for v in range(virtual_nodes):
            virtual_key = f"bucket_{bucket_id}_vnode_{v}"
            hash_point = int(hashlib.md5(virtual_key.encode()).hexdigest(), 16)
            self.ring[hash_point] = bucket_id
        
        self.sorted_hashes = sorted(self.ring.keys())
 
 
# Demonstration
def demonstrate_skew_handling():
    """Show skew detection and mitigation."""
    
    # Simulate skewed access pattern
    bucket_counts = [100, 100, 500, 100, 100, 800, 100, 100]  # Buckets 2, 5 are hot
    overflow_lengths = [0, 0, 5, 0, 0, 8, 0, 0]
    
    hot_buckets = SkewMitigation.split_hot_buckets(
        bucket_counts, overflow_lengths, threshold=2.0
    )
    
    print(f"Hot buckets identified: {hot_buckets}")
    print("Recommendation: Split these buckets proactively")
    
    # Adaptive sizing
    access_counts = {i: count for i, count in enumerate(bucket_counts)}
    capacities = SkewMitigation.adaptive_bucket_sizing(access_counts)
    
    print(f"
Adaptive capacities:")
    for bucket, cap in capacities.items():
        print(f"  Bucket {bucket}: {cap} (vs base 100)")
 
 
demonstrate_skew_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
134
135
136
137
138
139
140
141
142
143
144
145
146

from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
import json
 
@dataclass
class MaintenanceEvent:
    """Record of a maintenance action."""
    timestamp: datetime
    action: str
    reason: str
    before_metrics: dict
    after_metrics: Optional[dict] = None
    duration_seconds: float = 0
 
 
class ProactiveMaintenanceScheduler:
    """
    Schedule and execute proactive maintenance for hash indexes.
    
    Key insight: It's cheaper to maintain regularly than to
    recover from severe degradation.
    """
    
    def __init__(self):
        self.maintenance_history: List[MaintenanceEvent] = []
        self.check_interval_hours = 24
        self.last_check: Optional[datetime] = None
        
    def should_reorganize(self, metrics: dict) -> tuple[bool, str]:
        """
        Determine if reorganization is needed based on metrics.
        
        Returns (should_reorg, reason).
        """
        reasons = []
        
        # Check individual metrics
        if metrics.get('load_factor', 0) > 0.88:
            reasons.append(f"Load factor {metrics['load_factor']:.1%} approaching limit")
        
        if metrics.get('max_chain_length', 0) > 4:
            reasons.append(f"Max chain length {metrics['max_chain_length']} exceeds threshold")
        
        if metrics.get('avg_chain_length', 0) > 1.5:
            reasons.append(f"Average chain {metrics['avg_chain_length']:.2f} indicates widespread overflow")
        
        if metrics.get('utilization_variance', 0) > 0.35:
            reasons.append(f"Utilization variance {metrics['utilization_variance']:.1%} indicates imbalance")
        
        # Time-based maintenance
        days_since_reorg = metrics.get('days_since_reorganization', float('inf'))
        if days_since_reorg > 30:
            reasons.append(f"{days_since_reorg} days since last reorganization")
        
        if reasons:
            return True, "; ".join(reasons)
        return False, "Index healthy"
    
    def recommend_maintenance(self, metrics: dict) -> dict:
        """
        Generate maintenance recommendations.
        
        Returns actionable recommendations with priority.
        """
        recommendations = []
        
        should_reorg, reason = self.should_reorganize(metrics)
        
        if should_reorg:
            # Determine urgency
            urgent = (
                metrics.get('load_factor', 0) > 0.93 or
                metrics.get('max_chain_length', 0) > 8 or
                metrics.get('avg_chain_length', 0) > 3
            )
            
            recommendations.append({
                "action": "REORGANIZE",
                "priority": "HIGH" if urgent else "MEDIUM",
                "reason": reason,
                "estimated_duration": self._estimate_reorg_time(metrics),
                "recommended_window": "off-peak hours" if not urgent else "ASAP",
            })
        
        # Check for tuning opportunities
        if metrics.get('load_factor', 0) < 0.50:
            recommendations.append({
                "action": "SHRINK",
                "priority": "LOW",
                "reason": f"Low utilization ({metrics.get('load_factor', 0):.1%})",
                "estimated_savings": f"{(1 - metrics.get('load_factor', 0) / 0.70) * 100:.0f}% space",
            })
        
        # Check hash function quality
        if metrics.get('utilization_variance', 0) > 0.40:
            recommendations.append({
                "action": "EVALUATE_HASH_FUNCTION",
                "priority": "MEDIUM",
                "reason": "High utilization variance suggests poor hash distribution",
            })
        
        return {
            "status": "MAINTENANCE_NEEDED" if recommendations else "HEALTHY",
            "recommendations": recommendations,
            "metrics_summary": metrics,
            "next_check": self._calculate_next_check(recommendations),
        }
    
    def _estimate_reorg_time(self, metrics: dict) -> str:
        """Estimate reorganization duration."""
        records = metrics.get('total_records', 0)
        
        if records < 100000:
            return "< 1 minute"
        elif records < 1000000:
            return "1-5 minutes"
        elif records < 10000000:
            return "5-30 minutes"
        else:
            return "> 30 minutes"
    
    def _calculate_next_check(self, recommendations: List[dict]) -> str:
        """Determine when to check again."""
        if any(r['priority'] == 'HIGH' for r in recommendations):
            return "After maintenance completion"
        elif any(r['priority'] == 'MEDIUM' for r in recommendations):
            return "Within 24 hours"
        else:
            return "Standard interval (weekly)"
 
 
# Example usage
scheduler = ProactiveMaintenanceScheduler()
 
sample_metrics = {
    'load_factor': 0.87,
    'max_chain_length': 6,
    'avg_chain_length': 1.8,
    'utilization_variance': 0.28,
    'total_records': 500000,
    'days_since_reorganization': 45,
}
 
recommendation = scheduler.recommend_maintenance(sample_metrics)
print(json.dumps(recommendation, indent=2))