Sort-Merge Join: Algorithm, Implementation, and Optimization

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

def cost_comparison_pre_sorted():
    """
    Compare join costs with and without pre-sorted input.
    """
    b_R = 10_000   # 10,000 pages
    b_S = 50_000   # 50,000 pages
    M = 1_000      # 1,000 buffer pages
    
    # Case 1: Neither relation sorted
    # Sort R: 2 × 10,000 × 2 = 40,000 (1 merge pass)
    # Sort S: 2 × 50,000 × 2 = 200,000 (1 merge pass)
    # Merge: 10,000 + 50,000 = 60,000
    unsorted_cost = 40_000 + 200_000 + 60_000  # 300,000
    
    # Case 2: Both relations sorted (e.g., clustered index scan)
    # Sort R: 0 (already sorted)
    # Sort S: 0 (already sorted)
    # Merge: 10,000 + 50,000 = 60,000
    sorted_cost = 0 + 0 + 60_000  # 60,000
    
    # Case 3: Hash join (for comparison)
    # Partition: 2 × (10,000 + 50,000) = 120,000
    # Probe: 10,000 + 50,000 = 60,000
    hash_cost = 120_000 + 60_000  # 180,000
    
    print("Join Cost Comparison:")
    print(f"  Sort-Merge (unsorted): {unsorted_cost:,} page I/Os")
    print(f"  Sort-Merge (sorted):   {sorted_cost:,} page I/Os")
    print(f"  Hash Join:             {hash_cost:,} page I/Os")
    print()
    print(f"Pre-sorted advantage: {(unsorted_cost - sorted_cost) / unsorted_cost:.1%} cost reduction")
    print(f"vs Hash (sorted):     {(hash_cost - sorted_cost) / hash_cost:.1%} cheaper than hash")
 
 
# Results:
# Sort-Merge (unsorted): 300,000 page I/Os
# Sort-Merge (sorted):   60,000 page I/Os  <-- 5× cheaper!
# Hash Join:             180,000 page I/Os
# 
# Pre-sorted advantage: 80.0% cost reduction
# vs Hash (sorted):     66.7% cheaper than hash

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

def query_plan_comparison():
    """
    Compare query plans with and without sort-merge join.
    
    Query: SELECT customer_id, SUM(amount)
           FROM orders JOIN customers ON orders.customer_id = customers.id
           GROUP BY customer_id
           ORDER BY customer_id
    """
    b_orders = 50_000
    b_customers = 10_000
    M = 1_000
    
    # Plan A: Hash Join + Sort + Hash Aggregate + Sort
    # (Hash join output is unsorted)
    join_cost = 3 * (b_orders + b_customers)  # 180,000
    sort_for_group = 2 * (b_orders + b_customers) * 2  # ~240,000 (estimate)
    # Could use hash aggregate (no sort needed), but ORDER BY requires sort
    sort_for_order = 2 * (b_orders + b_customers) * 2  # ~240,000
    
    plan_a_cost = join_cost + sort_for_order
    # With streaming aggregate after sort: join + 1 sort
    plan_a_optimized = 180_000 + 240_000  # 420,000
    
    # Plan B: Sort-Merge Join (customers sorted by clustered index)
    # Output is sorted by customer_id!
    sort_orders = 2 * b_orders * 2  # 200,000
    sort_customers = 0  # Clustered index
    merge_cost = b_orders + b_customers  # 60,000
    # Output is sorted - stream aggregate AND no ORDER BY sort needed!
    
    plan_b_cost = sort_orders + merge_cost
    # Total: 260,000
    
    print("Query Plan Cost Comparison:")
    print("-" * 50)
    print("Plan A (Hash Join):")
    print(f"  Hash Join:     180,000")
    print(f"  Sort for ORDER BY: 240,000")
    print(f"  Stream Aggregate: (included)")
    print(f"  TOTAL:         {plan_a_optimized:,}")
    print()
    print("Plan B (Sort-Merge Join):")
    print(f"  Sort Orders:   200,000")
    print(f"  Merge:         60,000")
    print(f"  Aggregate:     (streaming, no cost)")
    print(f"  ORDER BY:      (already sorted, FREE!)")
    print(f"  TOTAL:         {plan_b_cost:,}")
    print()
    print(f"Sort-Merge saves: {plan_a_optimized - plan_b_cost:,} I/Os ({100*(plan_a_optimized - plan_b_cost)/plan_a_optimized:.1f}%)")
 
 
# Output:
# Plan B saves: 160,000 I/Os (38.1%)
# The sorted output eliminates both GROUP BY and ORDER BY sorts!

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

def sort_merge_range_join(R, S, r_key, s_start, s_end):
    """
    Range join: R.key BETWEEN S.start AND S.end
    
    Sort R by r_key, S by s_start. For each R tuple, scan S tuples
    where s_start <= r_key. Stop when s_start > r_key.
    Also check s_end >= r_key for completeness.
    
    Much more efficient than nested loop when both are sorted.
    """
    # Sort both relations
    R_sorted = sort(R, key=r_key)
    S_sorted = sort(S, key=s_start)
    
    r_idx = 0
    s_idx = 0
    
    # Queue of "active" S tuples whose range includes current position
    active_S = []
    
    for r in R_sorted:
        r_val = r[r_key]
        
        # Add S tuples whose start <= r_val
        while s_idx < len(S_sorted) and S_sorted[s_idx][s_start] <= r_val:
            active_S.append(S_sorted[s_idx])
            s_idx += 1
        
        # Remove S tuples whose end < r_val
        active_S = [s for s in active_S if s[s_end] >= r_val]
        
        # All remaining active_S tuples match this R tuple
        for s in active_S:
            yield concatenate(r, s)
 
 
def sort_merge_inequality_join(R, S, r_key, s_key, operator='<'):
    """
    Inequality join: R.key < S.key (or >, <=, >=)
    
    For '<': Each R tuple joins with all S tuples having larger key.
    Sort both, then for each R tuple, output with all S tuples beyond it.
    
    Warning: Can produce O(n²) output even with optimal algorithm!
    """
    R_sorted = sort(R, key=r_key)
    S_sorted = sort(S, key=s_key)
    
    s_start = 0  # Track where valid S tuples begin
    
    for r in R_sorted:
        r_val = r[r_key]
        
        # Advance s_start past S tuples that don't satisfy condition
        if operator == '<':
            while s_start < len(S_sorted) and S_sorted[s_start][s_key] <= r_val:
                s_start += 1
        elif operator == '<=':
            while s_start < len(S_sorted) and S_sorted[s_start][s_key] < r_val:
                s_start += 1
        # Similar for >, >=
        
        # Output all matching S tuples
        for s in S_sorted[s_start:]:
            yield concatenate(r, s)
 
 
# Example: Find all orders that exceeded customer's credit limit
# SELECT * FROM orders o JOIN customers c
# ON o.customer_id = c.id AND o.amount > c.credit_limit
 
def efficient_band_join(R, S, r_key, s_key, epsilon):
    """
    Band join: |R.key - S.key| <= epsilon
    
    Sort both relations. For each R tuple, only examine S tuples
    in the band [r_key - epsilon, r_key + epsilon].
    
    Time: O(|R| + |S| + |output|) when band is small.
    """
    R_sorted = sort(R, key=r_key)
    S_sorted = sort(S, key=s_key)
    
    s_start = 0  # Start of valid S range
    
    for r in R_sorted:
        r_val = r[r_key]
        lower = r_val - epsilon
        upper = r_val + epsilon
        
        # Advance s_start to first S tuple in range
        while s_start < len(S_sorted) and S_sorted[s_start][s_key] < lower:
            s_start += 1
        
        # Output all S tuples in range
        s_idx = s_start
        while s_idx < len(S_sorted) and S_sorted[s_idx][s_key] <= upper:
            yield concatenate(r, S_sorted[s_idx])
            s_idx += 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

def compare_memory_behavior():
    """
    Compare memory usage patterns of hash join vs sort-merge.
    """
    # Scenario: Join with skewed data
    # R: 1 million rows, 100MB
    # S: 10 million rows, 1GB (heavily skewed - 30% have same key value)
    
    # Hash Join Memory Analysis:
    # Build phase creates hash table for R (100MB base)
    # With 30% of S sharing one key value:
    #   - That hash bucket has 300,000 entries sharing collision lists
    #   - Causes severe hash table degradation
    #   - May trigger recursive partitioning
    # 
    # Memory: Unpredictable, could spike 2-3x expected
    
    # Sort-Merge Memory Analysis:
    # Sort R: Uses allocated buffer (say 100MB), produces runs
    # Sort S: Uses same buffer, produces runs (more runs, but same buffer)
    # Merge: Uses buffer for K-way merge
    # 
    # Skewed partition handling:
    #   - 30% matching keys create large partition
    #   - Handled by spilling partition to temp file
    #   - Uses same buffer size, just more I/O
    # 
    # Memory: Exactly as predicted by buffer allocation
    
    print("Memory Behavior Comparison:")
    print("-" * 50)
    print("Hash Join:")
    print("  Base estimate: 100 MB for hash table")
    print("  Actual (skewed): 200-300 MB (collision lists)")
    print("  Worst case: Recursive partitioning, unpredictable")
    print()
    print("Sort-Merge Join:")
    print("  Buffer allocation: 100 MB")
    print("  Actual usage: 100 MB (exactly)")
    print("  Skew impact: More runs, more I/O, same memory")
 
 
def memory_planning_example():
    """
    How predictable memory helps query planning.
    """
    # System has 2GB for query processing
    # Running 10 concurrent queries
    # Each gets 200MB budget
    
    # With hash join:
    # - Hard to guarantee 200MB is enough
    # - One query's hash table might balloon
    # - Risk of OOM or excessive spilling
    
    # With sort-merge:
    # - Each query uses exactly its 200MB buffer
    # - Sort produces runs that fit in budget
    # - Merge uses bounded number of buffers
    # - No risk of exceeding allocation
    
    print("\nQuery Memory Planning:")
    print("-" * 50)
    print("Available: 2 GB for 10 concurrent queries")
    print("Per-query budget: 200 MB")
    print()
    print("Hash Join risks:")
    print("  - Cannot guarantee budget compliance")
    print("  - Skew or estimation errors cause overruns")
    print()
    print("Sort-Merge guarantees:")
    print("  - Exact 200 MB usage per query")
    print("  - External sort handles any input size")
    print("  - Bounded, predictable behavior")

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

import concurrent.futures
from typing import List, Tuple
 
def parallel_distributed_sort_merge_join(
    R_partitions: List[str],  # R data files across nodes
    S_partitions: List[str],  # S data files across nodes  
    join_key: str,
    num_workers: int
) -> Iterator:
    """
    Distributed parallel sort-merge join.
    
    Strategy:
    1. Range-partition both R and S on join key
    2. Each worker sorts and merges its partition range
    3. Concatenate results (already in global sorted order)
    """
    # Step 1: Sample to determine key ranges
    key_ranges = determine_ranges(R_partitions + S_partitions, 
                                   join_key, num_workers)
    
    # Step 2: Redistribute data to workers by key range
    R_by_range = repartition_by_ranges(R_partitions, join_key, key_ranges)
    S_by_range = repartition_by_ranges(S_partitions, join_key, key_ranges)
    
    # Step 3: Each worker sorts and joins its range independently
    with concurrent.futures.ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i in range(num_workers):
            future = executor.submit(
                sort_merge_join_partition,
                R_by_range[i],
                S_by_range[i],
                join_key
            )
            futures.append((i, future))
        
        # Collect results in order (ranges are ordered)
        for i, future in sorted(futures, key=lambda x: x[0]):
            yield from future.result()
 
 
def sort_merge_join_partition(R_data, S_data, join_key):
    """
    Standard sort-merge join on a single partition.
    Runs on a single worker/node.
    """
    # Local sort
    R_sorted = sort(R_data, key=join_key)
    S_sorted = sort(S_data, key=join_key)
    
    # Standard merge
    return list(merge_join(R_sorted, S_sorted, join_key))
 
 
def parallel_sort_single_relation(data, key, num_workers):
    """
    Parallel sort of a single relation on a multi-core machine.
    
    1. Partition data into N chunks
    2. Sort each chunk in parallel
    3. Merge sorted chunks
    """
    chunk_size = len(data) // num_workers
    chunks = [data[i*chunk_size:(i+1)*chunk_size] for i in range(num_workers)]
    
    # Parallel sort
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
        sorted_chunks = list(executor.map(
            lambda chunk: sorted(chunk, key=lambda x: x[key]),
            chunks
        ))
    
    # K-way merge of sorted chunks
    return k_way_merge(sorted_chunks, key)
 
 
# Parallel merge phase is also possible with range-based parallelism
def parallel_merge_range_partitioned(
    sorted_R: str,
    sorted_S: str,
    join_key: str,
    key_ranges: List[Tuple],
    num_workers: int
) -> Iterator:
    """
    Parallel merge using non-overlapping key ranges.
    
    Each worker merges tuples within its assigned key range.
    Results are concatenated (no final merge needed - ranges are disjoint).
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i, (low, high) in enumerate(key_ranges):
            future = executor.submit(
                merge_range,
                sorted_R, sorted_S, join_key, low, high
            )
            futures.append((i, future))
        
        # Yield in order
        for i, future in sorted(futures, key=lambda x: x[0]):
            yield from future.result()

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

def recommend_join_algorithm(
    R_stats: RelationStats,
    S_stats: RelationStats,
    join_condition: JoinCondition,
    query_context: QueryContext
) -> str:
    """
    Recommend join algorithm based on characteristics.
    
    This is a simplified heuristic; real optimizers use cost estimation.
    """
    # Rule 1: Non-equi-join requires sort-merge or nested loop
    if not join_condition.is_equality:
        if R_stats.is_sorted and S_stats.is_sorted:
            return "SORT_MERGE"  # Best case for range join
        elif R_stats.num_tuples * S_stats.num_tuples < NESTED_LOOP_THRESHOLD:
            return "NESTED_LOOP"
        else:
            return "SORT_MERGE"  # Must sort, but still efficient
    
    # Rule 2: Pre-sorted inputs strongly favor sort-merge
    if R_stats.is_sorted and S_stats.is_sorted:
        return "SORT_MERGE"  # Cannot beat merge cost of b_R + b_S
    
    # Rule 3: ORDER BY on join key favors sort-merge
    if query_context.has_order_by_on_join_key:
        # Sort-merge's sorted output eliminates final sort
        return "SORT_MERGE"
    
    # Rule 4: GROUP BY on join key favors sort-merge
    if query_context.has_group_by_on_join_key:
        return "SORT_MERGE"  # Enables stream aggregate
    
    # Rule 5: Small LIMIT favors hash (or index nested loop)
    if query_context.limit and query_context.limit < 100:
        if has_index(S_stats, join_condition.S_key):
            return "INDEX_NESTED_LOOP"
        return "HASH_JOIN"
    
    # Rule 6: One sorted + need for ordered output
    if query_context.has_order_by_on_join_key:
        if R_stats.is_sorted or S_stats.is_sorted:
            return "SORT_MERGE"  # Partial sort savings
    
    # Default: For unsorted equi-join, hash join usually wins
    return "HASH_JOIN"
 
 
# Decision summary:
# 
# SORT_MERGE_INDICATORS:
#   - Pre-sorted inputs (clustered index, previous sort)
#   - ORDER BY / GROUP BY on join key
#   - Non-equality join conditions
#   - Subsequent joins on same key  
#   - Memory-constrained execution
#
# HASH_JOIN_INDICATORS:
#   - Unsorted inputs, no order needed
#   - Small LIMIT clause
#   - Highly skewed data
#   - Abundant memory, slow I/O