Grace 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
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

// Grace Hash Join implementation
class GraceHashJoin {
    size_t available_memory;
    int num_partitions;
    HashFunction partition_hash;
    HashFunction join_hash;
    
public:
    void execute(Relation* R, Relation* S, ResultSet* output) {
        // Determine number of partitions based on build relation size estimate
        num_partitions = compute_partition_count(R->estimated_size());
        
        // Phase 1: Partition both relations to disk
        vector<DiskPartition> R_partitions = partition_to_disk(R);
        vector<DiskPartition> S_partitions = partition_to_disk(S);
        
        // Phase 2: Join corresponding partition pairs
        for (int i = 0; i < num_partitions; i++) {
            if (R_partitions[i].is_empty()) {
                // No build tuples in this partition - no output possible
                continue;
            }
            
            join_partition_pair(R_partitions[i], S_partitions[i], output);
        }
        
        // Cleanup: delete temporary partition files
        for (int i = 0; i < num_partitions; i++) {
            R_partitions[i].delete_file();
            S_partitions[i].delete_file();
        }
    }
    
private:
    vector<DiskPartition> partition_to_disk(Relation* rel) {
        // Allocate output buffers - one page per partition
        vector<Page*> buffers(num_partitions);
        vector<DiskPartition> partitions(num_partitions);
        
        for (int i = 0; i < num_partitions; i++) {
            buffers[i] = allocate_page();
            partitions[i] = DiskPartition::create_temp();
        }
        
        // Scan relation, distribute tuples to partitions
        for (const Tuple& tuple : *rel) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (!buffers[p]->can_fit(tuple)) {
                // Buffer full: flush to disk
                partitions[p].write(buffers[p]);
                buffers[p]->clear();
            }
            
            buffers[p]->add(tuple);
        }
        
        // Final flush of remaining tuples
        for (int i = 0; i < num_partitions; i++) {
            if (!buffers[i]->is_empty()) {
                partitions[i].write(buffers[i]);
            }
            free_page(buffers[i]);
        }
        
        return partitions;
    }
    
    void join_partition_pair(DiskPartition& R_part, DiskPartition& S_part,
                             ResultSet* output) {
        // Check if build partition fits in memory
        if (R_part.size() > available_memory * 0.8) {
            // Need recursive partitioning
            recursive_join(R_part, S_part, output, 1);
            return;
        }
        
        // Build hash table from R partition
        HashTable* ht = new HashTable();
        for (const Page& page : R_part.pages()) {
            for (const Tuple& tuple : page) {
                uint64_t h = join_hash(tuple.join_key);
                ht->insert(h, tuple);
            }
        }
        
        // Probe with S partition
        for (const Page& page : S_part.pages()) {
            for (const Tuple& tuple : page) {
                uint64_t h = join_hash(tuple.join_key);
                for (Tuple* match : ht->find_all(h, tuple.join_key)) {
                    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

// Recursive partitioning in Grace Hash Join
class GraceHashJoinRecursive {
    static const int MAX_RECURSION_DEPTH = 10;
    vector<HashFunction> hash_functions;  // Different function per level
    
    void recursive_join(DiskPartition& R_part, DiskPartition& S_part,
                        ResultSet* output, int depth) {
        // Safety check: prevent infinite recursion
        if (depth > MAX_RECURSION_DEPTH) {
            handle_degenerate_case(R_part, S_part, output);
            return;
        }
        
        // Determine sub-partition count for this level
        size_t partition_size = R_part.size();
        int sub_partitions = compute_sub_partition_count(partition_size, depth);
        
        // Get hash function for this recursion level
        HashFunction h = hash_functions[depth];
        
        // Re-partition both relations
        vector<DiskPartition> R_subparts = repartition(R_part, h, sub_partitions);
        vector<DiskPartition> S_subparts = repartition(S_part, h, sub_partitions);
        
        // Process each sub-partition pair
        for (int i = 0; i < sub_partitions; i++) {
            if (R_subparts[i].is_empty()) continue;
            
            if (R_subparts[i].size() <= available_memory * 0.8) {
                // Sub-partition fits: do simple hash join
                simple_hash_join(R_subparts[i], S_subparts[i], output);
            } else {
                // Still doesn't fit: recurse deeper
                recursive_join(R_subparts[i], S_subparts[i], output, depth + 1);
            }
            
            // Clean up temporary files as we go
            R_subparts[i].delete_file();
            S_subparts[i].delete_file();
        }
    }
    
    void handle_degenerate_case(DiskPartition& R_part, DiskPartition& S_part,
                                ResultSet* output) {
        // If we hit max recursion, likely many tuples share same key
        // Options:
        // 1. Fall back to nested loop join
        // 2. Use block nested loop with available memory
        // 3. Sort-merge join the partition pair
        
        log_warning("Grace: degenerate case at partition level, "
                    "falling back to nested loop");
        
        nested_loop_join_partitions(R_part, S_part, output);
    }
    
    // Generate different hash functions per level
    static HashFunction get_level_hash(int level) {
        // Different seeds produce different hash distributions
        // Common approach: use level number as seed component
        return [level](const JoinKey& key) {
            uint64_t h = key.as_uint64();
            h ^= (level * 0x9e3779b97f4a7c15ULL);  // Level-specific mixing
            h *= 0xbf58476d1ce4e5b9ULL;
            h ^= h >> 27;
            h *= 0x94d049bb133111ebULL;
            h ^= h >> 31;
            return h;
        };
    }
};

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

// Memory requirement calculations for Grace Hash Join
class GraceMemoryPlanner {
    size_t available_memory;
    size_t page_size;
    double hash_table_overhead = 1.5;  // HT needs 1.5× raw tuple data
    
    // Minimum partitions needed for guaranteed single-level processing
    int min_partitions_needed(size_t build_size_bytes) {
        // Each partition must fit in available memory with HT overhead
        size_t max_partition_size = available_memory / hash_table_overhead;
        return (int)ceil((double)build_size_bytes / max_partition_size);
    }
    
    // Maximum partitions possible (limited by buffer memory)
    int max_partitions_possible() {
        // Need 1 input buffer + n output buffers = n+1 pages total
        // Reserve some pages for OS and miscellaneous
        size_t buffer_memory = available_memory / 2;  // Half for buffers
        return buffer_memory / page_size - 1;
    }
    
    // Optimal partition count balancing I/O and memory
    int optimal_partition_count(size_t build_size_bytes) {
        int min_n = min_partitions_needed(build_size_bytes);
        int max_n = max_partitions_possible();
        
        if (min_n > max_n) {
            // Cannot create enough partitions with current memory
            // Will need recursive partitioning
            return max_n;  // Create maximum partitions, recurse as needed
        }
        
        // Choose power of 2 >= min_n for efficient modulo
        int n = 1;
        while (n < min_n) n *= 2;
        
        // But don't exceed max
        return min(n, max_n);
    }
    
    // Memory layout during join phase
    struct JoinPhaseMemory {
        size_t hash_table;     // For build partition + overhead
        size_t probe_buffer;   // For reading probe partition
        size_t output_buffer;  // For join output
        size_t working;        // Miscellaneous
    };
    
    JoinPhaseMemory allocate_join_memory(size_t partition_size) {
        JoinPhaseMemory layout;
        layout.hash_table = partition_size * hash_table_overhead;
        layout.probe_buffer = page_size * 4;  // 4 pages for probe reading
        layout.output_buffer = page_size * 4;
        layout.working = available_memory - layout.hash_table 
                        - layout.probe_buffer - layout.output_buffer;
        return layout;
    }
};

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

// Optimized Grace Hash Join with double buffering and Bloom filters
class OptimizedGraceHashJoin {
    // Bloom filter per partition for fast emptiness checks
    vector<BloomFilter*> partition_blooms;
    
    // Async I/O handles for double buffering
    AsyncIOManager io_manager;
    
    void partition_with_optimizations(Relation* rel, 
                                       vector<DiskPartition>& partitions,
                                       bool is_build) {
        // Double buffers for async writes
        vector<Page*> active_bufs(num_partitions);
        vector<Page*> flush_bufs(num_partitions);
        vector<AsyncHandle> pending_writes(num_partitions);
        
        for (int i = 0; i < num_partitions; i++) {
            active_bufs[i] = allocate_page();
            flush_bufs[i] = allocate_page();
            if (is_build) {
                partition_blooms[i] = new BloomFilter(estimated_per_partition);
            }
        }
        
        // Compress before writing if enabled
        Compressor* compressor = create_lz4_compressor();
        
        for (const Tuple& tuple : *rel) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (is_build) {
                // Add to Bloom filter during build
                partition_blooms[p]->insert(join_hash(tuple.join_key));
            }
            
            if (!active_bufs[p]->can_fit(tuple)) {
                // Wait for any pending write to this partition
                if (pending_writes[p].valid()) {
                    io_manager.wait(pending_writes[p]);
                }
                
                // Swap buffers
                swap(active_bufs[p], flush_bufs[p]);
                
                // Compress and start async write
                auto compressed = compressor->compress(flush_bufs[p]);
                pending_writes[p] = io_manager.write_async(
                    partitions[p].file(), compressed
                );
                
                active_bufs[p]->clear();
            }
            
            active_bufs[p]->add(tuple);
        }
        
        // Final flush and cleanup
        for (int i = 0; i < num_partitions; i++) {
            if (pending_writes[i].valid()) {
                io_manager.wait(pending_writes[i]);
            }
            // Flush remaining
            if (!active_bufs[i]->is_empty()) {
                auto compressed = compressor->compress(active_bufs[i]);
                partitions[i].write(compressed);
            }
            free_page(active_bufs[i]);
            free_page(flush_bufs[i]);
        }
    }
    
    void join_with_bloom_filtering(vector<DiskPartition>& R_parts,
                                   vector<DiskPartition>& S_parts,
                                   ResultSet* output) {
        // Sort partitions by size for better memory management
        vector<int> order(num_partitions);
        iota(order.begin(), order.end(), 0);
        sort(order.begin(), order.end(), [&](int a, int b) {
            return R_parts[a].size() < R_parts[b].size();
        });
        
        for (int i : order) {
            if (R_parts[i].is_empty()) continue;
            
            // Quick check: any probe tuples in this partition?
            // Use build Bloom filter against sampled probe keys
            if (!probe_partition_might_have_matches(S_parts[i], 
                                                     partition_blooms[i])) {
                // Skip this partition pair entirely
                continue;
            }
            
            join_single_partition(R_parts[i], S_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

// Parallel Grace Hash Join
class ParallelGraceHashJoin {
    ThreadPool& workers;
    int num_workers;
    
    void execute_parallel(Relation* R, Relation* S, ResultSet* output) {
        // Phase 1: Parallel partitioning
        // Each worker has local partition buffers, merge at end
        vector<future<vector<DiskPartition>>> partition_futures;
        
        size_t r_chunk_size = R->size() / num_workers;
        for (int w = 0; w < num_workers; w++) {
            auto start = R->begin() + w * r_chunk_size;
            auto end = (w == num_workers - 1) ? R->end() 
                                               : start + r_chunk_size;
            partition_futures.push_back(
                workers.submit([=]() {
                    return partition_chunk(start, end);
                })
            );
        }
        
        // Merge partition files from all workers
        vector<vector<DiskPartition>> worker_partitions;
        for (auto& f : partition_futures) {
            worker_partitions.push_back(f.get());
        }
        vector<DiskPartition> R_partitions = merge_partitions(worker_partitions);
        
        // Repeat for S
        vector<DiskPartition> S_partitions = partition_parallel(S);
        
        // Phase 2: Parallel joining
        // Assign partition pairs to workers
        atomic<int> next_partition(0);
        vector<future<void>> join_futures;
        
        for (int w = 0; w < num_workers; w++) {
            join_futures.push_back(
                workers.submit([&, w]() {
                    while (true) {
                        int p = next_partition.fetch_add(1);
                        if (p >= num_partitions) break;
                        
                        if (!R_partitions[p].is_empty()) {
                            join_partition_pair(R_partitions[p], 
                                                S_partitions[p], 
                                                output->worker_output(w));
                        }
                    }
                })
            );
        }
        
        // Wait for all joins to complete
        for (auto& f : join_futures) {
            f.wait();
        }
    }
    
    // Work stealing for load balancing
    void join_with_work_stealing() {
        // Instead of static assignment, use work-stealing queue
        WorkStealingQueue<int> partition_queue;
        for (int i = 0; i < num_partitions; i++) {
            partition_queue.push(i);
        }
        
        // Workers steal work when idle
        // Helps balance when partition sizes vary
    }
};