第18课：连接算法

查询处理 第18课 / 共25课

📖 课程概述

连接(JOIN)是SQL查询中最常用也最昂贵的操作。不同的连接算法在不同场景下性能差异巨大：嵌套循环连接适合小表，哈希连接适合等值连接大表，排序归并连接适合已排序数据。本课深入实现三种连接算法并分析其适用场景。

🔗 三种连接算法

💻 C语言实现：三种连接算法

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <time.h>

#define MAX_ROWS   10000
#define MAX_KEY    64
#define MAX_VAL    128
#define HASH_SIZE  8192

typedef struct {
    int    key;
    char   value[MAX_VAL];
} Row;

typedef struct {
    Row*    rows;
    int     count;
    int     capacity;
    char    name[32];
} Table;

typedef struct {
    int    left_key;
    char   left_val[MAX_VAL];
    int    right_key;
    char   right_val[MAX_VAL];
} JoinResult;

// 创建表
Table* table_create(const char* name, int capacity) {
    Table* t = calloc(1, sizeof(Table));
    t->rows = calloc(capacity, sizeof(Row));
    t->capacity = capacity;
    t->count = 0;
    strcpy(t->name, name);
    return t;
}

void table_add(Table* t, int key, const char* val) {
    if (t->count >= t->capacity) return;
    t->rows[t->count].key = key;
    strcpy(t->rows[t->count].value, val);
    t->count++;
}

// ===== 1. 嵌套循环连接 =====
int nl_join(Table* outer, Table* inner, JoinResult* results) {
    int count = 0;
    clock_t start = clock();
    for (int i = 0; i < outer->count; i++) {
        for (int j = 0; j < inner->count; j++) {
            if (outer->rows[i].key == inner->rows[j].key) {
                JoinResult* r = &results[count++];
                r->left_key = outer->rows[i].key;
                strcpy(r->left_val, outer->rows[i].value);
                r->right_key = inner->rows[j].key;
                strcpy(r->right_val, inner->rows[j].value);
            }
        }
    }
    double ms = (double)(clock() - start) / CLOCKS_PER_SEC * 1000;
    printf("  [NL Join] %s×%s: %d条结果, %.2fms\n",
           outer->name, inner->name, count, ms);
    return count;
}

// ===== 2. 哈希连接 =====
typedef struct HashEntry {
    int    key;
    int    row_idx;
    struct HashEntry* next;
} HashEntry;

int hash_join(Table* build, Table* probe, JoinResult* results) {
    int count = 0;
    clock_t start = clock();

    // Build phase: 建立哈希表
    HashEntry* hash_table[HASH_SIZE] = {0};
    for (int i = 0; i < build->count; i++) {
        uint32_t h = build->rows[i].key % HASH_SIZE;
        HashEntry* e = malloc(sizeof(HashEntry));
        e->key = build->rows[i].key;
        e->row_idx = i;
        e->next = hash_table[h];
        hash_table[h] = e;
    }

    // Probe phase: 探测
    for (int i = 0; i < probe->count; i++) {
        uint32_t h = probe->rows[i].key % HASH_SIZE;
        HashEntry* e = hash_table[h];
        while (e) {
            if (e->key == probe->rows[i].key) {
                JoinResult* r = &results[count++];
                r->left_key = build->rows[e->row_idx].key;
                strcpy(r->left_val, build->rows[e->row_idx].value);
                r->right_key = probe->rows[i].key;
                strcpy(r->right_val, probe->rows[i].value);
            }
            e = e->next;
        }
    }

    // 清理
    for (int i = 0; i < HASH_SIZE; i++) {
        HashEntry* e = hash_table[i];
        while (e) { HashEntry* next = e->next; free(e); e = next; }
    }

    double ms = (double)(clock() - start) / CLOCKS_PER_SEC * 1000;
    printf("  [Hash Join] %s×%s: %d条结果, %.2fms\n",
           build->name, probe->name, count, ms);
    return count;
}

// ===== 3. 排序归并连接 =====
int row_cmp(const void* a, const void* b) {
    return ((Row*)a)->key - ((Row*)b)->key;
}

int merge_join(Table* left, Table* right, JoinResult* results) {
    int count = 0;
    clock_t start = clock();

    // Sort phase
    qsort(left->rows, left->count, sizeof(Row), row_cmp);
    qsort(right->rows, right->count, sizeof(Row), row_cmp);

    // Merge phase
    int i = 0, j = 0;
    while (i < left->count && j < right->count) {
        if (left->rows[i].key < right->rows[j].key) {
            i++;
        } else if (left->rows[i].key > right->rows[j].key) {
            j++;
        } else {
            // 匹配! 处理重复键
            int li = i, rj = j;
            while (li < left->count && left->rows[li].key == left->rows[i].key) li++;
            while (rj < right->count && right->rows[rj].key == right->rows[j].key) rj++;
            // 笛卡尔积匹配的行
            for (int a = i; a < li; a++) {
                for (int b = j; b < rj; b++) {
                    JoinResult* r = &results[count++];
                    r->left_key = left->rows[a].key;
                    strcpy(r->left_val, left->rows[a].value);
                    r->right_key = right->rows[b].key;
                    strcpy(r->right_val, right->rows[b].value);
                }
            }
            i = li;
            j = rj;
        }
    }

    double ms = (double)(clock() - start) / CLOCKS_PER_SEC * 1000;
    printf("  [Merge Join] %s×%s: %d条结果, %.2fms\n",
           left->name, right->name, count, ms);
    return count;
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   连接算法对比                       ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    srand(42);

    // 创建测试表
    Table* users = table_create("users", 5000);
    Table* orders = table_create("orders", 20000);

    for (int i = 0; i < 5000; i++) {
        char val[32];
        snprintf(val, sizeof(val), "user_%d", i);
        table_add(users, i, val);
    }
    for (int i = 0; i < 20000; i++) {
        char val[32];
        snprintf(val, sizeof(val), "order_%d", i);
        table_add(orders, rand() % 5000, val);  // 随机用户ID
    }

    JoinResult* results = malloc(sizeof(JoinResult) * MAX_ROWS * 10);

    // 对比三种算法
    printf("--- 三种连接算法对比 ---\n");
    nl_join(users, orders, results);
    hash_join(users, orders, results);
    merge_join(users, orders, results);

    // 不同规模对比
    printf("\n--- 不同规模对比 ---\n");
    int sizes[] = {100, 500, 1000, 5000};
    for (int s = 0; s < 4; s++) {
        printf("\n规模: %d × %d\n", sizes[s], sizes[s]*4);
        Table* t1 = table_create("t1", sizes[s]);
        Table* t2 = table_create("t2", sizes[s]*4);
        for (int i = 0; i < sizes[s]; i++) table_add(t1, i, "v");
        for (int i = 0; i < sizes[s]*4; i++) table_add(t2, rand() % sizes[s], "v");
        nl_join(t1, t2, results);
        hash_join(t1, t2, results);
        merge_join(t1, t2, results);
        free(t1->rows); free(t1);
        free(t2->rows); free(t2);
    }

    free(results);
    printf("\n✅ 连接算法对比运行完成\n");
    return 0;
}

🐍 Python实现：连接性能可视化

"""
连接算法性能可视化
"""
import time, random
from collections import defaultdict

def nl_join(left, right, on):
    results = []
    for lr in left:
        for rr in right:
            if lr[on] == rr[on]:
                results.append({**lr, **rr})
    return results

def hash_join(build, probe, on):
    # Build
    ht = defaultdict(list)
    for r in build: ht[r[on]].append(r)
    # Probe
    results = []
    for r in probe:
        for match in ht.get(r[on], []):
            results.append({**match, **r})
    return results

def merge_join(left, right, on):
    left_s = sorted(left, key=lambda r: r[on])
    right_s = sorted(right, key=lambda r: r[on])
    results = []
    i = j = 0
    while i < len(left_s) and j < len(right_s):
        if left_s[i][on] < right_s[j][on]: i += 1
        elif left_s[i][on] > right_s[j][on]: j += 1
        else:
            li = i
            while li < len(left_s) and left_s[li][on] == left_s[i][on]: li += 1
            rj = j
            while rj < len(right_s) and right_s[rj][on] == right_s[j][on]: rj += 1
            for a in range(i, li):
                for b in range(j, rj):
                    results.append({**left_s[a], **right_s[b]})
            i, j = li, rj
    return results

def bench(algo, left, right, on, name):
    t0 = time.perf_counter()
    result = algo(left, right, on)
    ms = (time.perf_counter() - t0) * 1000
    print(f"  {name:>15}: {len(result):>6} rows, {ms:>8.2f}ms")
    return ms

# 测试
print("=== 连接算法性能对比 ===\n")
for n in [100, 500, 1000, 3000]:
    left = [{"id": i, "name": f"u{i}"} for i in range(n)]
    right = [{"uid": random.randint(0, n-1), "order": f"o{i}"} for i in range(n*4)]
    print(f"规模: {n} × {n*4}")
    bench(nl_join, left, right, "id", "NL Join")
    bench(hash_join, left, right, "id", "Hash Join")
    bench(merge_join, left, right, "id", "Merge Join")
    print()

print("✅ 连接算法性能对比完成")

🔑 关键概念总结