第19课：排序与聚合

查询处理 第19课 / 共25课

📖 课程概述

排序(Sort)和聚合(Aggregate)是查询处理中常见的操作。ORDER BY需要排序，GROUP BY需要分组聚合。本课实现外部排序(External Sort)算法处理超内存数据，以及流式聚合和哈希聚合两种聚合策略。

📊 排序与聚合算法

💻 C语言实现：外部排序与聚合

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

#define PAGE_SIZE     4096
#define RECORDS_PER_PAGE (PAGE_SIZE / sizeof(SortRecord))
#define BUFFER_PAGES  8
#define MAX_RUNS      64

typedef struct {
    int    key;
    double value;
    int    group_id;
} SortRecord;

typedef struct {
    SortRecord* records;
    int         count;
    int         capacity;
} Run;

// 比较函数
int cmp_key(const void* a, const void* b) {
    return ((SortRecord*)a)->key - ((SortRecord*)b)->key;
}

int cmp_group(const void* a, const void* b) {
    return ((SortRecord*)a)->group_id - ((SortRecord*)b)->group_id;
}

// 内排序：对一批记录排序
Run* sort_run(SortRecord* data, int count) {
    Run* run = malloc(sizeof(Run));
    run->records = malloc(count * sizeof(SortRecord));
    memcpy(run->records, data, count * sizeof(SortRecord));
    run->count = count;
    run->capacity = count;
    qsort(run->records, count, sizeof(SortRecord), cmp_key);
    return run;
}

// 外部归并排序
typedef struct {
    Run**  runs;
    int    num_runs;
    int*   positions;  // 每个run的当前位置
} MergeState;

Run* external_merge_sort(SortRecord* data, int total_records) {
    printf("[ExtSort] 总记录: %d, 缓冲区: %d页\n", total_records, BUFFER_PAGES);

    // Phase 1: 生成有序runs
    int run_size = RECORDS_PER_PAGE * BUFFER_PAGES;
    Run* runs[MAX_RUNS];
    int num_runs = 0;

    for (int i = 0; i < total_records; i += run_size) {
        int chunk = (i + run_size < total_records) ? run_size : total_records - i;
        runs[num_runs] = sort_run(&data[i], chunk);
        printf("  [ExtSort] Run %d: %d条记录\n", num_runs, chunk);
        num_runs++;
    }

    // Phase 2: 多路归并
    Run* result = malloc(sizeof(Run));
    result->count = total_records;
    result->records = malloc(total_records * sizeof(SortRecord));
    result->capacity = total_records;

    int* pos = calloc(num_runs, sizeof(int));
    int output_idx = 0;

    // 简化: 逐一取最小
    while (output_idx < total_records) {
        int min_run = -1;
        SortRecord* min_rec = NULL;
        for (int i = 0; i < num_runs; i++) {
            if (pos[i] < runs[i]->count) {
                if (!min_rec || runs[i]->records[pos[i]].key < min_rec->key) {
                    min_rec = &runs[i]->records[pos[i]];
                    min_run = i;
                }
            }
        }
        if (min_run < 0) break;
        result->records[output_idx++] = *min_rec;
        pos[min_run]++;
    }

    // 清理
    for (int i = 0; i < num_runs; i++) {
        free(runs[i]->records);
        free(runs[i]);
    }
    free(pos);

    printf("  [ExtSort] 归并完成: %d条记录\n", output_idx);
    return result;
}

// ===== 聚合操作 =====

// 聚合结果
typedef struct {
    int    group_id;
    int    count;
    double sum;
    double min_val;
    double max_val;
    double avg;
} AggResult;

// 排序聚合(流式聚合)
int sort_aggregate(SortRecord* data, int n, AggResult* results) {
    // 先按group_id排序
    qsort(data, n, sizeof(SortRecord), cmp_group);

    int num_groups = 0;
    int i = 0;
    while (i < n) {
        AggResult* agg = &results[num_groups];
        agg->group_id = data[i].group_id;
        agg->count = 0;
        agg->sum = 0;
        agg->min_val = data[i].value;
        agg->max_val = data[i].value;

        while (i < n && data[i].group_id == agg->group_id) {
            agg->count++;
            agg->sum += data[i].value;
            if (data[i].value < agg->min_val) agg->min_val = data[i].value;
            if (data[i].value > agg->max_val) agg->max_val = data[i].value;
            i++;
        }
        agg->avg = agg->sum / agg->count;
        num_groups++;
    }
    return num_groups;
}

// 哈希聚合
#define AGG_BUCKETS 1024

int hash_aggregate(SortRecord* data, int n, AggResult* results) {
    AggResult buckets[AGG_BUCKETS];
    int bucket_used[AGG_BUCKETS] = {0};
    int num_groups = 0;

    // 建哈希表聚合
    for (int i = 0; i < n; i++) {
        int h = data[i].group_id % AGG_BUCKETS;
        // 简化: 线性探测
        while (bucket_used[h] && buckets[h].group_id != data[i].group_id)
            h = (h + 1) % AGG_BUCKETS;

        if (!bucket_used[h]) {
            buckets[h].group_id = data[i].group_id;
            buckets[h].count = 0;
            buckets[h].sum = 0;
            buckets[h].min_val = data[i].value;
            buckets[h].max_val = data[i].value;
            bucket_used[h] = 1;
        }
        buckets[h].count++;
        buckets[h].sum += data[i].value;
        if (data[i].value < buckets[h].min_val) buckets[h].min_val = data[i].value;
        if (data[i].value > buckets[h].max_val) buckets[h].max_val = data[i].value;
    }

    // 收集结果
    for (int i = 0; i < AGG_BUCKETS; i++) {
        if (bucket_used[i]) {
            buckets[i].avg = buckets[i].sum / buckets[i].count;
            results[num_groups++] = buckets[i];
        }
    }
    return num_groups;
}

void print_agg(AggResult* results, int n) {
    printf("  %-8s %6s %10s %10s %10s %10s\n",
           "Group", "Count", "Sum", "Min", "Max", "Avg");
    for (int i = 0; i < n && i < 10; i++) {
        printf("  %-8d %6d %10.1f %10.1f %10.1f %10.1f\n",
               results[i].group_id, results[i].count,
               results[i].sum, results[i].min_val,
               results[i].max_val, results[i].avg);
    }
    if (n > 10) printf("  ... 共 %d 组\n", n);
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   排序与聚合                         ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    srand(42);
    int N = 50000;
    SortRecord* data = malloc(N * sizeof(SortRecord));
    for (int i = 0; i < N; i++) {
        data[i].key = rand();
        data[i].value = (double)(rand() % 10000) / 10;
        data[i].group_id = rand() % 100;
    }

    // 外部排序
    printf("--- 外部归并排序 ---\n");
    clock_t t0 = clock();
    Run* sorted = external_merge_sort(data, N);
    printf("  耗时: %.2fms\n", (double)(clock()-t0)/CLOCKS_PER_SEC*1000);
    printf("  前5个: ");
    for (int i = 0; i < 5; i++) printf("%d ", sorted->records[i].key);
    printf("\n  验证有序: ");
    int is_sorted = 1;
    for (int i = 1; i < N; i++) {
        if (sorted->records[i].key < sorted->records[i-1].key) {
            is_sorted = 0; break;
        }
    }
    printf("%s\n", is_sorted ? "✅" : "❌");

    // 排序聚合
    printf("\n--- 排序聚合 ---\n");
    AggResult* agg1 = malloc(200 * sizeof(AggResult));
    t0 = clock();
    int ng1 = sort_aggregate(data, N, agg1);
    printf("  耗时: %.2fms, %d组\n", (double)(clock()-t0)/CLOCKS_PER_SEC*1000, ng1);
    print_agg(agg1, ng1);

    // 哈希聚合
    printf("\n--- 哈希聚合 ---\n");
    AggResult* agg2 = malloc(200 * sizeof(AggResult));
    t0 = clock();
    int ng2 = hash_aggregate(data, N, agg2);
    printf("  耗时: %.2fms, %d组\n", (double)(clock()-t0)/CLOCKS_PER_SEC*1000, ng2);
    print_agg(agg2, ng2);

    free(data); free(sorted->records); free(sorted); free(agg1); free(agg2);
    printf("\n✅ 排序与聚合运行完成\n");
    return 0;
}

🐍 Python实现：聚合性能对比

"""
排序聚合 vs 哈希聚合 性能对比
"""
import time, random
from collections import defaultdict

N = 500000
NUM_GROUPS = 1000
data = [(random.randint(0, NUM_GROUPS-1), random.random()*1000) for _ in range(N)]

# 排序聚合
t0 = time.perf_counter()
data_sorted = sorted(data, key=lambda x: x[0])
sort_agg = []
i = 0
while i < len(data_sorted):
    gid = data_sorted[i][0]
    cnt = 0; s = 0; mn = float('inf'); mx = float('-inf')
    while i < len(data_sorted) and data_sorted[i][0] == gid:
        cnt += 1; s += data_sorted[i][1]
        mn = min(mn, data_sorted[i][1]); mx = max(mx, data_sorted[i][1])
        i += 1
    sort_agg.append((gid, cnt, s, mn, mx, s/cnt))
sort_time = (time.perf_counter() - t0) * 1000

# 哈希聚合
t0 = time.perf_counter()
ht = defaultdict(lambda: [0, 0, float('inf'), float('-inf')])
for gid, val in data:
    ht[gid][0] += 1
    ht[gid][1] += val
    ht[gid][2] = min(ht[gid][2], val)
    ht[gid][3] = max(ht[gid][3], val)
hash_agg = [(gid, s[0], s[1], s[2], s[3], s[1]/s[0]) for gid, s in ht.items()]
hash_time = (time.perf_counter() - t0) * 1000

print(f"记录: {N}, 分组: {NUM_GROUPS}")
print(f"排序聚合: {sort_time:.1f}ms, {len(sort_agg)}组")
print(f"哈希聚合: {hash_time:.1f}ms, {len(hash_agg)}组")
print(f"哈希聚合快: {sort_time/hash_time:.1f}x")
print("✅ 聚合对比完成")

🔑 关键概念总结