Hybrid Hash Join

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

// Conceptual Hybrid Hash Join
class HybridHashJoin {
    HashTable* partition0_ht;    // In-memory hash table for partition 0
    vector<DiskPartition> build_partitions;  // Disk partitions 1..n-1
    vector<DiskPartition> probe_partitions;
    size_t memory_for_ht;        // Memory allocated to partition 0
    size_t memory_for_buffers;   // Memory for disk partition buffers
    
    void execute(Relation* R, Relation* S, ResultSet* output) {
        // Phase 1: Build - partition R, keeping partition 0 in memory
        partition0_ht = new HashTable(estimated_p0_size);
        
        for (const Tuple& tuple : *R) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (p == 0) {
                // Keep in memory
                partition0_ht->insert(join_hash(tuple.join_key), tuple);
            } else {
                // Write to disk partition
                write_to_partition(build_partitions[p], tuple);
            }
        }
        
        // Phase 2: Probe - partition S, joining partition 0 immediately
        for (const Tuple& tuple : *S) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (p == 0) {
                // Probe immediately against in-memory hash table
                for (Tuple* match : partition0_ht->find_all(
                        join_hash(tuple.join_key), tuple.join_key)) {
                    output->emit(tuple, *match);
                }
            } else {
                // Write to disk for later processing
                write_to_partition(probe_partitions[p], tuple);
            }
        }
        
        // Phase 3: Process remaining disk-based partition pairs (like Grace)
        for (int i = 1; i < num_partitions; i++) {
            if (!build_partitions[i].is_empty()) {
                join_partition_pair(build_partitions[i], 
                                    probe_partitions[i], output);
            }
        }
        
        delete partition0_ht;
    }
};

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

// Memory allocation for Hybrid Hash Join
class HybridMemoryAllocator {
    size_t total_memory;
    size_t page_size;
    double hash_table_overhead = 1.5;  // Hash table uses 1.5× raw data
    
    struct AllocationPlan {
        size_t ht_memory;        // For partition 0 hash table
        size_t buffer_memory;    // For disk partition buffers
        int num_partitions;
        size_t per_buffer_size;  // Size of each disk buffer
    };
    
    AllocationPlan compute_allocation(size_t estimated_build_size) {
        AllocationPlan plan;
        
        // First, determine number of partitions needed
        // Each partition (except 0) must fit in remaining memory later
        // Available for later = M - any lingering overhead
        size_t join_phase_memory = total_memory * 0.9;  // 90% usable
        
        // Expected size per partition after partitioning
        plan.num_partitions = 1;  // Start with minimum
        while (true) {
            size_t partition_size = estimated_build_size / plan.num_partitions;
            size_t ht_size = partition_size * hash_table_overhead;
            
            if (ht_size <= join_phase_memory) {
                break;  // Partitions will fit
            }
            plan.num_partitions *= 2;  // Double partitions
        }
        
        // Now allocate memory between partition 0 and buffers
        // Partition 0 expected size
        size_t expected_p0_size = estimated_build_size / plan.num_partitions;
        size_t p0_ht_size = expected_p0_size * hash_table_overhead;
        
        // Memory needed for disk buffers (1 page minimum per partition)
        size_t min_buffer_memory = (plan.num_partitions - 1) * page_size;
        
        // Can we fit partition 0 and minimum buffers?
        if (p0_ht_size + min_buffer_memory > total_memory) {
            // Reduce partition 0 allocation
            plan.ht_memory = total_memory - min_buffer_memory;
        } else {
            // Plenty of room: use expected size, give rest to buffers
            plan.ht_memory = p0_ht_size;
        }
        
        plan.buffer_memory = total_memory - plan.ht_memory;
        plan.per_buffer_size = plan.buffer_memory / (plan.num_partitions - 1);
        
        return plan;
    }
    
    // Dynamic adjustment during execution
    void handle_p0_overflow(AllocationPlan& plan, HashTable* ht) {
        // If partition 0 exceeds allocation, need to spill
        // Options:
        // 1. Convert to full Grace (spill p0 to disk)
        // 2. Steal memory from disk buffers (reduce their size)
        // 3. Spill oldest entries from p0 (complex LRU-style)
        
        // Option 1 is simplest and most common
        if (ht->memory_used() > plan.ht_memory * 1.2) {
            // Spill partition 0 to disk, convert to pure Grace
            DiskPartition p0_disk = spill_to_disk(ht);
            // Continue as Grace from here...
        }
    }
};

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

// Detailed Hybrid Hash Join Implementation
class DetailedHybridHashJoin {
    // Configuration
    int num_partitions;
    size_t total_memory;
    size_t p0_memory_limit;
    
    // State
    HashTable* p0_hash_table;
    vector<PartitionBuffer> build_buffers;  // For partitions 1..n-1
    vector<PartitionBuffer> probe_buffers;
    vector<DiskPartition> build_partitions;
    vector<DiskPartition> probe_partitions;
    bool p0_spilled = false;  // Track if partition 0 has been spilled
    
public:
    void execute(Relation* R, Relation* S, ResultSet* output) {
        // ========================================
        // INITIALIZATION
        // ========================================
        initialize_partitions();
        p0_hash_table = new HashTable();
        
        // ========================================
        // PHASE 1: BUILD - Partition R, keep p0 in memory
        // ========================================
        for (const Tuple& tuple : *R) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (p == 0 && !p0_spilled) {
                // Try to insert into in-memory hash table
                if (!try_insert_p0(tuple)) {
                    // Memory limit reached - spill partition 0
                    spill_partition_0();
                    // Now write to disk partition instead
                    write_to_disk_partition(0, tuple);
                } else {
                    p0_hash_table->insert(join_hash(tuple.join_key), tuple);
                }
            } else {
                // Write to disk partition
                write_to_disk_partition(p, tuple);
            }
        }
        
        // Flush all disk partition buffers
        flush_all_build_buffers();
        
        // ========================================
        // PHASE 2: PROBE - Partition S, probe p0 immediately
        // ========================================
        for (const Tuple& tuple : *S) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (p == 0 && !p0_spilled) {
                // Probe against in-memory hash table immediately
                auto matches = p0_hash_table->find_all(
                    join_hash(tuple.join_key), tuple.join_key);
                    
                for (Tuple* match : matches) {
                    output->emit(tuple, *match);
                }
                // Note: tuple NOT written to disk
            } else {
                // Write to disk partition for later processing
                write_to_probe_partition(p, tuple);
            }
        }
        
        // Flush all probe partition buffers
        flush_all_probe_buffers();
        
        // ========================================
        // PHASE 3: JOIN remaining partitions (Grace-style)
        // ========================================
        // Free partition 0's hash table to make room for others
        delete p0_hash_table;
        p0_hash_table = nullptr;
        
        // Start from partition 0 if it was spilled, else from 1
        int start_partition = p0_spilled ? 0 : 1;
        
        for (int i = start_partition; i < num_partitions; i++) {
            if (build_partitions[i].is_empty()) {
                continue;  // Skip empty partitions
            }
            
            join_disk_partition_pair(i, output);
        }
    }
    
private:
    bool try_insert_p0(const Tuple& tuple) {
        size_t tuple_overhead = sizeof(HashEntry) + tuple.size();
        return (p0_hash_table->memory_used() + tuple_overhead) <= p0_memory_limit;
    }
    
    void spill_partition_0() {
        log_warning("Partition 0 exceeded memory limit, spilling to disk");
        
        // Write all entries from hash table to disk partition
        for (HashEntry& entry : *p0_hash_table) {
            write_to_disk_partition(0, entry.tuple);
        }
        
        // Clear hash table but keep structure for potential reuse
        p0_hash_table->clear();
        p0_spilled = true;
    }
    
    void join_disk_partition_pair(int partition_idx, ResultSet* output) {
        // Load build partition into memory
        HashTable* ht = new HashTable();
        
        if (build_partitions[partition_idx].size() > (total_memory * 0.8)) {
            // Won't fit - need recursive partitioning
            recursive_partition_and_join(partition_idx, output);
            return;
        }
        
        for (const Page& page : build_partitions[partition_idx].pages()) {
            for (const Tuple& tuple : page) {
                ht->insert(join_hash(tuple.join_key), tuple);
            }
        }
        
        // Probe with corresponding probe partition
        for (const Page& page : probe_partitions[partition_idx].pages()) {
            for (const Tuple& tuple : page) {
                auto matches = ht->find_all(
                    join_hash(tuple.join_key), tuple.join_key);
                for (Tuple* match : matches) {
                    output->emit(tuple, *match);
                }
            }
        }
        
        delete ht;
    }
};

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

// Multi-partition Hybrid Hash Join
class MultiPartitionHybrid {
    int num_partitions;
    int num_in_memory;  // Number of partitions kept in memory
    vector<HashTable*> in_memory_tables;  // Partitions 0..k-1
    vector<DiskPartition> disk_build_parts;  // Partitions k..n-1
    vector<DiskPartition> disk_probe_parts;
    
    void initialize(size_t build_size_estimate, size_t memory) {
        // Determine total partitions needed
        num_partitions = compute_partition_count(build_size_estimate, memory);
        
        // How many can we fit in memory?
        // Each partition is approximately build_size / num_partitions
        size_t per_partition = build_size_estimate / num_partitions;
        size_t per_partition_ht = per_partition * 1.5;  // 1.5× overhead
        
        // Leave some memory for disk buffers
        size_t memory_for_hts = memory * 0.8;
        num_in_memory = min(
            (int)(memory_for_hts / per_partition_ht),
            num_partitions
        );
        
        // Allocate hash tables for in-memory partitions
        in_memory_tables.resize(num_in_memory);
        for (int i = 0; i < num_in_memory; i++) {
            in_memory_tables[i] = new HashTable();
        }
        
        // Allocate disk partitions for the rest
        disk_build_parts.resize(num_partitions - num_in_memory);
        disk_probe_parts.resize(num_partitions - num_in_memory);
    }
    
    void process_build_tuple(const Tuple& tuple) {
        int p = partition_hash(tuple.join_key) % num_partitions;
        
        if (p < num_in_memory) {
            // In-memory partition
            in_memory_tables[p]->insert(join_hash(tuple.join_key), tuple);
        } else {
            // Disk partition
            disk_build_parts[p - num_in_memory].append(tuple);
        }
    }
    
    void process_probe_tuple(const Tuple& tuple, ResultSet* output) {
        int p = partition_hash(tuple.join_key) % num_partitions;
        
        if (p < num_in_memory) {
            // Probe against in-memory hash table immediately
            auto matches = in_memory_tables[p]->find_all(
                join_hash(tuple.join_key), tuple.join_key);
            for (Tuple* match : matches) {
                output->emit(tuple, *match);
            }
        } else {
            // Write to disk for later processing
            disk_probe_parts[p - num_in_memory].append(tuple);
        }
    }
    
    void finish(ResultSet* output) {
        // Free in-memory hash tables
        for (auto* ht : in_memory_tables) {
            delete ht;
        }
        in_memory_tables.clear();
        
        // Process disk partitions (Grace-style)
        for (int i = 0; i < disk_build_parts.size(); i++) {
            if (!disk_build_parts[i].is_empty()) {
                join_partition_pair(disk_build_parts[i], 
                                    disk_probe_parts[i], output);
            }
        }
    }
};

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

// Adaptive overflow handling in Hybrid Hash Join
class AdaptiveHybridHashJoin {
    enum class OverflowStrategy {
        FULL_SPILL,      // Convert to Grace
        PARTIAL_SPILL,   // Spill portion of p0
        STEAL_BUFFERS,   // Reduce disk buffers
        ADAPTIVE_SELECT  // Switch which partition is in-memory
    };
    
    OverflowStrategy strategy = OverflowStrategy::ADAPTIVE_SELECT;
    vector<size_t> partition_sizes;  // Track sizes during build
    int in_memory_partition = 0;     // Which partition is in memory
    
    void handle_memory_overflow() {
        switch (strategy) {
            case OverflowStrategy::FULL_SPILL:
                // Dump partition 0 to disk, continue as Grace
                spill_in_memory_partition();
                break;
                
            case OverflowStrategy::PARTIAL_SPILL:
                // Spill half of entries, keep half
                partial_spill_oldest_entries(0.5);
                break;
                
            case OverflowStrategy::STEAL_BUFFERS:
                // Reduce disk buffer sizes
                if (can_reduce_buffers()) {
                    reallocate_buffer_memory();
                } else {
                    spill_in_memory_partition();
                }
                break;
                
            case OverflowStrategy::ADAPTIVE_SELECT:
                // Find smallest disk partition, swap with in-memory
                adaptive_partition_swap();
                break;
        }
    }
    
    void adaptive_partition_swap() {
        // During build, track sizes of disk partitions
        size_t in_memory_size = p0_hash_table->memory_used();
        
        // Find smallest disk partition
        int smallest_disk = -1;
        size_t smallest_size = SIZE_MAX;
        for (int i = 1; i < num_partitions; i++) {
            if (partition_sizes[i] < smallest_size) {
                smallest_size = partition_sizes[i];
                smallest_disk = i;
            }
        }
        
        // If a disk partition is smaller than current in-memory, swap
        if (smallest_size < in_memory_size * 0.5) {
            // Spill current in-memory partition
            spill_partition(in_memory_partition);
            
            // Load smallest disk partition into memory
            // (will be done at end of build phase)
            in_memory_partition = smallest_disk;
            
            log_info("Swapped in-memory partition {} -> {}", 
                     0, smallest_disk);
        } else {
            // No beneficial swap possible, use partial spill
            partial_spill_oldest_entries(0.5);
        }
    }
    
    void partial_spill_oldest_entries(double fraction) {
        // Use LRU or FIFO to select entries to spill
        size_t to_spill = p0_hash_table->num_entries() * fraction;
        
        auto entries_to_spill = p0_hash_table->get_oldest_entries(to_spill);
        for (auto& entry : entries_to_spill) {
            // Write to disk partition 0
            disk_partition_0.append(entry.tuple);
            p0_hash_table->remove(entry);
        }
        
        // Mark that partition 0 is split between memory and disk
        p0_is_split = true;
    }
};