Partition Handling

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

// Partitioned hash join algorithm
class PartitionedHashJoin {
    int num_partitions;
    vector<PartitionFile> build_partitions;
    vector<PartitionFile> probe_partitions;
    HashFunction partition_hash;  // hash₁
    HashFunction join_hash;       // hash₂
    
    void execute(Relation* R, Relation* S, ResultSet* output) {
        // Phase 1: Partition both relations
        partition_relation(R, build_partitions);
        partition_relation(S, probe_partitions);
        
        // Phase 2: Join corresponding partitions
        for (int i = 0; i < num_partitions; i++) {
            if (build_partitions[i].empty()) {
                // No build tuples in this partition = no output
                continue;
            }
            
            // Build hash table from partition i of R
            HashTable* ht = build_hash_table(build_partitions[i]);
            
            // Probe with partition i of S
            probe_and_output(probe_partitions[i], ht, output);
            
            delete ht;
        }
    }
    
private:
    void partition_relation(Relation* rel, vector<PartitionFile>& partitions) {
        // Allocate one buffer page per partition
        vector<Page*> buffers(num_partitions);
        for (int i = 0; i < num_partitions; i++) {
            buffers[i] = allocate_page();
        }
        
        // Scan relation and partition
        for (Tuple& tuple : *rel) {
            int p = partition_hash(tuple.join_key) % num_partitions;
            
            if (buffers[p]->is_full()) {
                // Flush buffer to disk
                partitions[p].write(buffers[p]);
                buffers[p]->clear();
            }
            
            buffers[p]->add(tuple);
        }
        
        // Flush remaining tuples
        for (int i = 0; i < num_partitions; i++) {
            if (!buffers[i]->empty()) {
                partitions[i].write(buffers[i]);
            }
            free_page(buffers[i]);
        }
    }
    
    HashTable* build_hash_table(PartitionFile& partition) {
        HashTable* ht = new HashTable();
        for (Page* page : partition.pages()) {
            for (Tuple& tuple : *page) {
                // Use second hash function for hash table
                uint64_t h = join_hash(tuple.join_key);
                ht->insert(h, tuple);
            }
        }
        return 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

// Using two different hash functions
class HashFunctions {
    // Hash function 1: for partitioning
    // Uses one set of mixing constants
    static uint64_t partition_hash(const JoinKey& key) {
        uint64_t h = key.as_uint64();
        // MurmurHash-style mixing
        h ^= h >> 33;
        h *= 0xff51afd7ed558ccdULL;
        h ^= h >> 33;
        h *= 0xc4ceb9fe1a85ec53ULL;
        h ^= h >> 33;
        return h;
    }
    
    // Hash function 2: for in-partition hash table
    // Uses different constants or algorithm
    static uint64_t table_hash(const JoinKey& key) {
        uint64_t h = key.as_uint64();
        // Different mixing pattern
        h *= 0x9e3779b97f4a7c15ULL;  // Based on golden ratio
        h ^= h >> 30;
        h *= 0xbf58476d1ce4e5b9ULL;
        h ^= h >> 27;
        h *= 0x94d049bb133111ebULL;
        h ^= h >> 31;
        return h;
    }
};
 
// Example showing why different functions matter
void demonstrate_hash_difference() {
    // All keys that map to partition 5 (out of 16) using partition_hash
    // have partition_hash(key) mod 16 == 5
    
    // If we used partition_hash for the hash table with 32 buckets:
    // All these keys would map to buckets 5 or 21 only!
    // (because hash mod 16 == 5 implies hash mod 32 is 5 or 21)
    
    // Using table_hash distributes these keys across all 32 buckets
}

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

// Partition count determination
class PartitionPlanner {
    size_t available_memory;     // In bytes
    size_t page_size;
    double fudge_factor;
    
    int compute_partition_count(
        size_t estimated_build_size,  // bytes
        size_t estimated_tuple_width
    ) {
        size_t memory_pages = available_memory / page_size;
        size_t build_pages = (estimated_build_size + page_size - 1) / page_size;
        
        // Account for hash table overhead
        // Hash table typically needs 1.5-2x the raw data size
        size_t ht_pages = build_pages * 2;
        
        // Minimum partitions to ensure each fits in memory
        int min_partitions = (int)ceil(fudge_factor * ht_pages / memory_pages);
        
        // Maximum partitions limited by buffer pool allocation
        // Need one buffer page per partition during partitioning phase
        int max_partitions = memory_pages - 2;  // Reserve pages for I/O
        
        // Choose power of 2 for efficient modulo
        int n = 1;
        while (n < min_partitions) n *= 2;
        
        // Clamp to valid range
        return min(n, max_partitions);
    }
    
    // Validate that partitions will fit
    bool validate_partition_plan(int n, size_t build_size) {
        size_t expected_partition_size = build_size / n;
        size_t ht_size = expected_partition_size * 2;  // 2x for hash table
        
        // Allow for skew: largest partition might be 2-3× average
        size_t worst_case = ht_size * 3;
        
        return worst_case < available_memory * 0.9;  // 90% threshold
    }
};

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

// I/O cost estimation for partitioned hash join
struct IOCostEstimate {
    size_t build_pages;
    size_t probe_pages;
    size_t memory_pages;
    double io_cost_per_page;  // milliseconds
    double random_io_penalty;  // multiplier for random vs sequential
    
    // Simple hash join (if build fits)
    double simple_hash_join_cost() {
        if (build_pages * 2 > memory_pages) {
            return INFINITY;  // Won't fit
        }
        // Sequential read of both relations
        return (build_pages + probe_pages) * io_cost_per_page;
    }
    
    // Two-pass partitioned hash join
    double partitioned_hash_join_cost() {
        // Phase 1: read + write both relations
        double phase1 = 2 * (build_pages + probe_pages) * io_cost_per_page;
        
        // Phase 2: read both relations again
        double phase2 = (build_pages + probe_pages) * io_cost_per_page;
        
        return phase1 + phase2;
    }
    
    // Block nested loop for comparison
    double block_nested_loop_cost() {
        size_t blocks_of_R = (build_pages + memory_pages - 1) / (memory_pages - 1);
        
        // Read R once, read S once per block of R
        double cost = build_pages * io_cost_per_page;
        cost += probe_pages * blocks_of_R * io_cost_per_page;
        
        // Add penalty for random access pattern
        cost *= random_io_penalty * 0.3;  // Partial random factor
        
        return cost;
    }
};
 
// Example: join 100,000-page R with 1,000,000-page S, 10,000-page memory
// Simple hash join: not possible (R doesn't fit)
// Partitioned: 3 × (100K + 1M) × 1ms = 3,300 seconds
// BNL: 100K × 1ms + 1M × 10 × 1ms = 10,100 seconds
// Partitioned hash join wins by 3×

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

// Recursive partitioning for overflow
void process_partition_recursive(
    PartitionFile& build_partition,
    PartitionFile& probe_partition,
    int depth,
    ResultSet* output
) {
    // Read build partition and check size
    size_t partition_size = build_partition.size_bytes();
    
    if (partition_size <= available_memory * 0.8) {
        // Base case: partition fits - do simple hash join
        HashTable* ht = build_hash_table(build_partition);
        probe_and_output(probe_partition, ht, output);
        delete ht;
        return;
    }
    
    // Overflow case: need to re-partition
    if (depth > MAX_RECURSION_DEPTH) {
        // Extreme skew detected - fall back to nested loop
        log_warning("Partition overflow at max depth, using nested loop");
        nested_loop_join(build_partition, probe_partition, output);
        return;
    }
    
    // Recursive case: re-partition this partition into sub-partitions
    int sub_partitions = compute_sub_partition_count(partition_size);
    
    vector<PartitionFile> build_subparts(sub_partitions);
    vector<PartitionFile> probe_subparts(sub_partitions);
    
    // Use different hash function for sub-partitioning
    HashFunction sub_hash = get_hash_function(depth + 1);
    
    // Re-partition build side
    for (Tuple& t : build_partition) {
        int sp = sub_hash(t.join_key) % sub_partitions;
        build_subparts[sp].append(t);
    }
    
    // Re-partition probe side
    for (Tuple& t : probe_partition) {
        int sp = sub_hash(t.join_key) % sub_partitions;
        probe_subparts[sp].append(t);
    }
    
    // Recursively process sub-partitions
    for (int i = 0; i < sub_partitions; i++) {
        if (!build_subparts[i].empty()) {
            process_partition_recursive(
                build_subparts[i], probe_subparts[i],
                depth + 1, 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

// Buffer allocation strategies for partitioning
class PartitionBufferManager {
    size_t total_memory;
    int num_partitions;
    
    // Strategy 1: Fixed equal allocation
    vector<Buffer> fixed_equal_allocation() {
        size_t input_reserved = 4 * PAGE_SIZE;
        size_t available = total_memory - input_reserved;
        size_t per_partition = available / num_partitions;
        
        vector<Buffer> buffers;
        for (int i = 0; i < num_partitions; i++) {
            buffers.push_back(Buffer(per_partition));
        }
        return buffers;
    }
    
    // Strategy 2: Dynamic with stealing
    class DynamicBufferPool {
        size_t total;
        size_t allocated;
        vector<Buffer*> partition_buffers;
        
        void write_and_potentially_expand(int partition, const Tuple& t) {
            Buffer* buf = partition_buffers[partition];
            
            if (buf->is_full()) {
                // Flush to disk
                flush_buffer(partition);
                
                // Try to expand if under-average allocation
                size_t avg = total / num_partitions;
                if (buf->size < avg && allocated < total) {
                    expand_buffer(buf, min(avg, total - allocated));
                }
            }
            
            buf->add(t);
        }
    };
    
    // Strategy 3: Double buffering for async I/O
    class DoubleBufferPartition {
        Buffer* active;   // Currently filling
        Buffer* flushing; // Being written to disk
        AsyncWriter* writer;
        
        void add(const Tuple& t) {
            if (active->is_full()) {
                // Wait for previous flush to complete
                writer->wait();
                
                // Swap buffers
                swap(active, flushing);
                
                // Start async write of full buffer
                writer->write_async(flushing);
            }
            active->add(t);
        }
    };
};