Blocking Nature

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

# Illustration: Why sorting must wait for all input
 
def non_blocking_filter(input_stream, predicate):
    """
    Selection is non-blocking: output immediately when condition met.
    """
    for tuple in input_stream:
        if predicate(tuple):
            yield tuple  # Immediate output!
 
def blocking_sort(input_stream, key):
    """
    Sorting is blocking: must collect everything before outputting.
    """
    # Phase 1: Consume ALL input (blocking)
    all_tuples = []
    for tuple in input_stream:
        all_tuples.append(tuple)
    # At this point, input is exhausted but no output produced yet
    
    # Phase 2: Sort
    all_tuples.sort(key=key)
    
    # Phase 3: Finally produce output
    for tuple in all_tuples:
        yield tuple
 
# The key difference:
# - non_blocking_filter yields during input consumption
# - blocking_sort only yields AFTER all input consumed
 
# Timeline for 1 million tuples:
# 
# Non-blocking filter:
# Time 0:      Input tuple 1 → Output tuple 1 (if matches)
# Time 1:      Input tuple 2 → Output tuple 2 (if matches)
# ...
# Time 999999: Input tuple 1M → Output (if matches)
# 
# Blocking sort:
# Time 0:      Input tuple 1 → (buffered, no output)
# Time 1:      Input tuple 2 → (buffered, no output)
# ...
# Time 999999: Input tuple 1M → (buffered, no output)
# Time 1000000: Sort begins → (no output)
# Time 1000000+sort_time: Output tuple 1
# Time 1000000+sort_time+1: Output tuple 2
# ...

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

def execute_pipeline_with_blocking(plan):
    """
    Execution with a blocking operator (sort) in the middle.
    """
    # Phase 1: Everything below the sort runs as a pipeline
    print("Phase 1: Executing sub-plan below sort...")
    filter_to_sort_buffer = []
    
    for page in scan("table"):
        for tuple in page:
            if filter_predicate(tuple):
                filter_to_sort_buffer.append(tuple)
                # Note: We must buffer here - can't pass to sort yet
    
    print(f"Buffered {len(filter_to_sort_buffer)} tuples for sort")
    # Memory usage peaks here!
    
    # Phase 2: Sort (blocking)
    print("Phase 2: Sorting...")
    sorted_data = sort(filter_to_sort_buffer, key=sort_key)
    # Original buffer can now be freed
    
    # Phase 3: Everything above sort runs as a pipeline
    print("Phase 3: Executing sub-plan above sort...")
    for tuple in sorted_data:
        projected = project(tuple)
        yield projected  # Streaming output
 
 
# For sort-merge join specifically:
def execute_sort_merge_join(R, S, join_key, buffer_pages):
    """
    Sort-merge join execution showing blocking points.
    """
    # BLOCKING: Sort R
    print("Sorting R (blocking)...")
    sorted_R = external_sort(R, join_key, buffer_pages // 2)
    # All of R must be processed before continuing
    
    # BLOCKING: Sort S  
    print("Sorting S (blocking)...")
    sorted_S = external_sort(S, join_key, buffer_pages // 2)
    # All of S must be processed before merge
    
    # NON-BLOCKING: Merge phase
    print("Merging (streaming)...")
    for match in merge_join(sorted_R, sorted_S, join_key):
        yield match  # Results stream out immediately

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

from dataclasses import dataclass
from typing import List, Set
from enum import Enum
 
class OperatorType(Enum):
    SCAN = "scan"
    FILTER = "filter"
    PROJECT = "project"
    SORT = "sort"
    MERGE_JOIN = "merge_join"
    HASH_JOIN_BUILD = "hash_build"
    HASH_JOIN_PROBE = "hash_probe"
    AGGREGATE = "aggregate"
 
BLOCKING_OPERATORS = {
    OperatorType.SORT,
    OperatorType.HASH_JOIN_BUILD,
    OperatorType.AGGREGATE,
}
 
@dataclass
class Operator:
    type: OperatorType
    children: List['Operator']
    estimated_rows: int
    
    @property
    def is_blocking(self) -> bool:
        return self.type in BLOCKING_OPERATORS
 
def identify_pipeline_stages(root: Operator) -> List[Set[Operator]]:
    """
    Identify pipeline stages in a query plan.
    
    Returns list of stages, each stage is a set of operators
    that can execute together without blocking.
    """
    stages = []
    current_stage = set()
    
    def traverse(op: Operator):
        if op.is_blocking:
            # Blocking operator ends current stage
            if current_stage:
                stages.append(current_stage.copy())
                current_stage.clear()
            
            # Process children first (they run before this operator)
            for child in op.children:
                traverse(child)
            
            # This operator is its own "stage" (the blocking point)
            stages.append({op})
        else:
            # Non-blocking operator joins current stage
            current_stage.add(op)
            for child in op.children:
                traverse(child)
    
    traverse(root)
    if current_stage:
        stages.append(current_stage)
    
    return stages
 
 
def analyze_sort_merge_plan():
    """
    Analyze a sort-merge join plan for pipeline stages.
    """
    # Build plan: Scan R → Sort R → Merge ← Sort S ← Scan S
    scan_r = Operator(OperatorType.SCAN, [], 1000000)
    scan_s = Operator(OperatorType.SCAN, [], 5000000)
    sort_r = Operator(OperatorType.SORT, [scan_r], 1000000)
    sort_s = Operator(OperatorType.SORT, [scan_s], 5000000)
    merge = Operator(OperatorType.MERGE_JOIN, [sort_r, sort_s], 500000)
    
    stages = identify_pipeline_stages(merge)
    
    print("Pipeline stages for Sort-Merge Join:")
    for i, stage in enumerate(stages):
        ops = [op.type.value for op in stage]
        print(f"  Stage {i+1}: {ops}")
    
    # Output:
    # Stage 1: ['scan'] (Scan R, feeds into Sort R)
    # Stage 2: ['sort'] (Sort R - blocking)
    # Stage 3: ['scan'] (Scan S, feeds into Sort S)
    # Stage 4: ['sort'] (Sort S - blocking)
    # Stage 5: ['merge_join'] (Merge - consumes both sorted inputs)
 
 
analyze_sort_merge_plan()

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

import concurrent.futures
from typing import List, Iterator
 
def parallel_sort_merge_join(R_source, S_source, join_key, 
                              num_workers: int = 4):
    """
    Parallel execution of sort-merge join.
    
    Both sorts run concurrently, reducing wall-clock time.
    """
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
        # Submit both sorts concurrently
        future_R = executor.submit(external_sort, R_source, join_key)
        future_S = executor.submit(external_sort, S_source, join_key)
        
        # Wait for both to complete (still blocking overall)
        sorted_R = future_R.result()
        sorted_S = future_S.result()
    
    # Merge phase (still sequential, but could parallelize for ranges)
    yield from merge_join(sorted_R, sorted_S, join_key)
 
 
def parallel_external_sort(source, key, num_workers: int = 4):
    """
    Parallel external merge sort.
    
    - Phase 1: Partition data, sort each partition in parallel
    - Phase 2: Merge sorted partitions
    """
    # Partition input for parallel sorting
    partitions = partition_data(source, num_workers)
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
        # Sort all partitions concurrently
        futures = [
            executor.submit(in_memory_sort, partition, key)
            for partition in partitions
        ]
        sorted_partitions = [f.result() for f in futures]
    
    # K-way merge of sorted partitions
    return k_way_merge(sorted_partitions)
 
 
def range_partitioned_merge(sorted_R, sorted_S, join_key, ranges):
    """
    Parallel merge using range partitioning.
    
    If data is range-partitioned on join key, each range can be
    merged independently in parallel.
    """
    results = []
    
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for (low, high) in ranges:
            # Extract tuples in this key range from both relations
            R_range = filter_range(sorted_R, join_key, low, high)
            S_range = filter_range(sorted_S, join_key, low, high)
            
            # Merge this range on a separate thread
            future = executor.submit(
                list, merge_join(R_range, S_range, join_key)
            )
            futures.append(future)
        
        # Collect results (already in correct order due to range design)
        for future in futures:
            results.extend(future.result())
    
    return results

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

def analyze_blocking_timeline():
    """
    Compare blocking timeline for sort-merge vs hash join.
    
    Scenario: Join R (1000 pages) with S (5000 pages)
    """
    # Time units (arbitrary, proportional to work)
    R_scan_time = 1000
    R_sort_time = 2000  # External sort with merge passes
    S_scan_time = 5000
    S_sort_time = 10000
    merge_time = 6000   # Merge both sorted relations
    
    build_time = 1000   # Build hash table from R
    probe_time = 5000   # Probe hash table with S
    
    # Sort-Merge Timeline:
    # 
    # Time 0-3000: Sort R (scan + sort) - No output
    # Time 0-15000: Sort S (scan + sort) - No output (parallel possible)
    # Time 15000: Both sorts complete
    # Time 15000-21000: Merge - Output streaming
    # 
    # First row at: 15000
    # Last row at: 21000
    
    sort_merge_first_row = max(R_scan_time + R_sort_time, 
                                S_scan_time + S_sort_time)
    sort_merge_complete = sort_merge_first_row + merge_time
    
    print("Sort-Merge Join:")
    print(f"  First row at: {sort_merge_first_row}")
    print(f"  Complete at: {sort_merge_complete}")
    
    # Hash Join Timeline:
    # 
    # Time 0-1000: Build from R (scan + hash) - No output
    # Time 1000-6000: Probe with S - Output streaming
    # 
    # First row at: 1000 (as soon as build done)
    # Last row at: 6000
    
    hash_join_first_row = R_scan_time + build_time
    hash_join_complete = hash_join_first_row + probe_time
    
    print("
Hash Join:")
    print(f"  First row at: {hash_join_first_row}")  
    print(f"  Complete at: {hash_join_complete}")
    
    # Analysis
    print(f"
First Row: Hash join is {sort_merge_first_row - hash_join_first_row}x faster")
    print(f"Complete: Hash join is {(sort_merge_complete - hash_join_complete) / hash_join_complete:.1%} faster")
 
 
analyze_blocking_timeline()