Hash Join Algorithms

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

-- Conceptual representation of the build phase
-- For a join: SELECT * FROM Orders O JOIN Customers C ON O.customer_id = C.customer_id
 
-- STEP 1: The optimizer identifies Customers as the smaller relation
-- Customers: 100,000 rows × 200 bytes = ~20 MB
-- Orders: 10,000,000 rows × 150 bytes = ~1.5 GB
 
-- STEP 2: Build phase processes Customers
-- Pseudocode:
--   hash_table = new HashMap()
--   
--   FOR each tuple t in Customers:
--       key = t.customer_id
--       bucket = hash(key) mod num_buckets
--       hash_table[bucket].append(t)
--   
--   RETURN hash_table
 
-- STEP 3: The resulting hash table structure
-- Bucket 0: [Customer(id=1000, name='Acme Corp', ...), Customer(id=5000, ...)]
-- Bucket 1: [Customer(id=1001, name='Beta Inc', ...)]
-- Bucket 2: []  -- empty bucket (sparse is normal)
-- Bucket 3: [Customer(id=1003, ...), Customer(id=7003, ...), ...]
-- ...
-- Bucket N-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

// Common hash functions used in database systems
 
// 1. Multiplicative Hashing (simple, fast)
// Based on Knuth's suggestion using the golden ratio
uint64_t multiplicative_hash(uint64_t key, int num_buckets) {
    const uint64_t A = 11400714819323198485ULL; // 2^64 / φ
    return ((key * A) >> (64 - log2(num_buckets)));
}
 
// 2. MurmurHash3 (excellent distribution, widely used)
// PostgreSQL, MySQL, and many systems use variants
uint32_t murmur3_hash(const void* key, int len, uint32_t seed) {
    // Processes input in 4-byte chunks with mixing
    // Provides excellent avalanche properties
    // ~10-15 cycles for typical keys
    // ... implementation details omitted for brevity
}
 
// 3. CRC32 Hardware-Accelerated (Intel CRC32C instruction)
// Modern CPUs have dedicated instructions making this very fast
uint32_t crc32c_hash(uint64_t key) {
    // Uses _mm_crc32_u64 intrinsic on x86
    // Single cycle on modern CPUs
    return _mm_crc32_u64(0, key);
}
 
// 4. Composite Key Hashing
// For joins on multiple columns (composite keys)
uint64_t composite_hash(const Tuple& t, const vector<int>& key_columns) {
    uint64_t hash = 0;
    for (int col : key_columns) {
        // Combine hashes using XOR-shift pattern
        hash ^= hash_column(t, col);
        hash = hash * 31 + (hash >> 17);  // mixing step
    }
    return 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
56
57
58
59
60
61
62
63
64
65

// Chained Hash Table Implementation
struct ChainedHashTable {
    struct Entry {
        Tuple tuple;
        Entry* next;
    };
    
    Entry** buckets;      // Array of bucket heads
    size_t num_buckets;
    size_t num_entries;
    
    void insert(const Tuple& t, uint64_t key) {
        size_t bucket = hash(key) % num_buckets;
        Entry* entry = new Entry{t, buckets[bucket]};
        buckets[bucket] = entry;  // Prepend to chain
        num_entries++;
    }
    
    vector<Tuple*> find(uint64_t key) {
        vector<Tuple*> results;
        size_t bucket = hash(key) % num_buckets;
        Entry* curr = buckets[bucket];
        while (curr) {
            if (get_key(curr->tuple) == key) {
                results.push_back(&curr->tuple);
            }
            curr = curr->next;
        }
        return results;
    }
};
 
// Linear Probing Hash Table Implementation
struct LinearProbingHashTable {
    struct Slot {
        bool occupied;
        uint64_t key;
        Tuple tuple;
    };
    
    Slot* slots;
    size_t capacity;
    size_t size;
    
    void insert(const Tuple& t, uint64_t key) {
        size_t idx = hash(key) % capacity;
        while (slots[idx].occupied) {
            idx = (idx + 1) % capacity;  // Linear probe
        }
        slots[idx] = {true, key, t};
        size++;
    }
    
    // Note: More complex for duplicate keys
    Tuple* find(uint64_t key) {
        size_t idx = hash(key) % capacity;
        while (slots[idx].occupied) {
            if (slots[idx].key == key) {
                return &slots[idx].tuple;
            }
            idx = (idx + 1) % capacity;
        }
        return nullptr;  // Not found
    }
};

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

// Memory estimation before build phase
struct BuildMemoryEstimate {
    size_t estimated_tuples;
    size_t avg_tuple_bytes;
    size_t join_key_bytes;
    
    size_t estimate_hash_table_size() {
        // Bucket array: use ~2x tuples for low load factor
        size_t num_buckets = next_prime(estimated_tuples * 2);
        size_t bucket_array = num_buckets * sizeof(void*);
        
        // Tuple storage with chain pointers
        size_t tuple_storage = estimated_tuples * 
            (avg_tuple_bytes + sizeof(void*) + sizeof(uint64_t));
            //  tuple data   + next pointer  + cached hash
        
        // Add 10% for fragmentation and alignment
        size_t total = (bucket_array + tuple_storage) * 1.1;
        
        return total;
    }
    
    bool fits_in_memory(size_t available_memory) {
        size_t required = estimate_hash_table_size();
        // Reserve 20% for probe-phase working memory
        return required < available_memory * 0.8;
    }
    
    int estimate_partitions_needed(size_t available_memory) {
        size_t required = estimate_hash_table_size();
        size_t usable = available_memory * 0.8;
        // Each partition should fit comfortably
        // Add one for safety margin
        return (required / usable) + 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

// Build phase initialization
class HashJoinBuildOperator {
    HashTable* hash_table;
    vector<int> join_key_columns;
    size_t tuples_processed;
    
    void initialize(const PlanNode& build_input, size_t available_memory) {
        // 1. Get cardinality estimate from optimizer
        size_t estimated_rows = build_input.estimated_cardinality();
        size_t tuple_width = build_input.output_schema().row_width();
        
        // 2. Determine hash table sizing
        size_t memory_needed = estimate_memory(estimated_rows, tuple_width);
        
        if (memory_needed > available_memory) {
            // Will need partitioning - handled by Hybrid/Grace strategies
            throw NeedPartitioningException(memory_needed, available_memory);
        }
        
        // 3. Allocate hash table with appropriate size
        size_t num_buckets = compute_bucket_count(estimated_rows);
        hash_table = new HashTable(num_buckets);
        
        // 4. Identify join key columns from query plan
        join_key_columns = build_input.join_keys();
        
        tuples_processed = 0;
    }
};

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

void process_build_relation(Operator* build_child) {
    Tuple tuple;
    
    // Request tuples from child operator (may be scan, filter, prior join, etc.)
    while (build_child->next(&tuple)) {
        // 1. Extract join key value(s)
        JoinKey key = extract_key(tuple, join_key_columns);
        
        // 2. Compute hash value
        uint64_t hash_value = compute_hash(key);
        
        // 3. Insert into hash table
        //    Many implementations store the hash value alongside the tuple
        //    to avoid recomputing during probe phase comparisons
        hash_table->insert(hash_value, key, tuple);
        
        tuples_processed++;
        
        // 4. Optional: Check memory pressure periodically
        if (tuples_processed % 10000 == 0) {
            if (memory_pressure_detected()) {
                // Some systems can switch to hybrid approach mid-build
                trigger_partial_partition_flush();
            }
        }
    }
    
    // Build phase complete
    log_build_complete(tuples_processed, hash_table->memory_used());
}

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

// NULL handling during build phase
void insert_tuple(const Tuple& tuple, JoinType join_type) {
    JoinKey key = extract_key(tuple, join_key_columns);
    
    // Check for NULL in any key column
    if (key.has_null()) {
        if (join_type == JoinType::INNER) {
            // Skip: NULL can never match
            null_tuples_skipped++;
            return;
        } else if (join_type == JoinType::LEFT_OUTER ||
                   join_type == JoinType::FULL_OUTER) {
            // Must preserve for potential non-matching output
            null_key_tuples.push_back(tuple);
            return;
        }
        // RIGHT_OUTER: build side doesn't need NULL tracking
    }
    
    // Normal insertion for non-NULL keys
    uint64_t hash_value = compute_hash(key);
    hash_table->insert(hash_value, key, tuple);
}
 
// For FULL OUTER JOIN, we also need to track which build tuples matched
struct BuildTupleEntry {
    Tuple tuple;
    bool matched;  // Set during probe phase
};