第20课：执行引擎

查询处理 第20课 / 共25课

📖 课程概述

执行引擎是查询处理的最后一环，负责将优化器生成的物理计划转化为实际的数据操作。执行模型决定了算子如何协作：火山模型(Volcano)是经典的拉取模型，向量化执行(Vectorized)和编译执行(Compiled)是现代优化方向。本课实现三种执行模型并对比性能。

⚙️ 三种执行模型

💻 C语言实现：火山模型执行引擎

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

#define MAX_ROWS  50000
#define BATCH_SIZE 1024

// 数据行
typedef struct {
    int    id;
    char   name[32];
    int    age;
    double salary;
} Row;

// 算子基类
typedef struct Operator Operator;

typedef void (*OpenFunc)(Operator* self);
typedef int  (*NextFunc)(Operator* self, Row* out);
typedef void (*CloseFunc)(Operator* self);

struct Operator {
    OpenFunc  open;
    NextFunc  next;
    CloseFunc close;
    Operator* child;
    Operator* child2;  // 双子节点(Join等)
    int       started;
    long      rows_returned;
};

// ===== Scan算子 =====
typedef struct {
    Operator  base;
    Row*      data;
    int       count;
    int       pos;
    const char* table_name;
} ScanOp;

void scan_open(Operator* self) { ((ScanOp*)self)->pos = 0; self->started = 1; }
int  scan_next(Operator* self, Row* out) {
    ScanOp* s = (ScanOp*)self;
    if (s->pos >= s->count) return 0;
    *out = s->data[s->pos++];
    self->rows_returned++;
    return 1;
}
void scan_close(Operator* self) {}

ScanOp* scan_create(Row* data, int count, const char* name) {
    ScanOp* s = calloc(1, sizeof(ScanOp));
    s->base.open = scan_open; s->base.next = scan_next; s->base.close = scan_close;
    s->data = data; s->count = count; s->table_name = name;
    return s;
}

// ===== Filter算子 =====
typedef struct {
    Operator  base;
    int       (*predicate)(Row*);
    const char* pred_desc;
} FilterOp;

void filter_open(Operator* self) { if (self->child) self->child->open(self->child); self->started = 1; }
int  filter_next(Operator* self, Row* out) {
    FilterOp* f = (FilterOp*)self;
    while (self->child->next(self->child, out)) {
        if (f->predicate(out)) {
            self->rows_returned++;
            return 1;
        }
    }
    return 0;
}
void filter_close(Operator* self) { if (self->child) self->child->close(self->child); }

FilterOp* filter_create(Operator* child, int (*pred)(Row*), const char* desc) {
    FilterOp* f = calloc(1, sizeof(FilterOp));
    f->base.open = filter_open; f->base.next = filter_next; f->base.close = filter_close;
    f->base.child = child; f->predicate = pred; f->pred_desc = desc;
    return f;
}

// ===== Project算子 =====
typedef struct {
    Operator  base;
    void      (*project)(Row* out, Row* in);
} ProjectOp;

void project_open(Operator* self) { if (self->child) self->child->open(self->child); self->started = 1; }
int  project_next(Operator* self, Row* out) {
    ProjectOp* p = (ProjectOp*)self;
    Row tmp;
    if (self->child->next(self->child, &tmp)) {
        p->project(out, &tmp);
        self->rows_returned++;
        return 1;
    }
    return 0;
}
void project_close(Operator* self) { if (self->child) self->child->close(self->child); }

// ===== Sort算子 =====
typedef struct {
    Operator  base;
    Row*      sorted;
    int       count;
    int       pos;
    int       (*cmp)(const Row*, const Row*);
} SortOp;

void sort_open(Operator* self) {
    SortOp* s = (SortOp*)self;
    // 收集所有输入
    s->sorted = malloc(MAX_ROWS * sizeof(Row));
    s->count = 0;
    Row tmp;
    while (self->child->next(self->child, &tmp)) {
        s->sorted[s->count++] = tmp;
    }
    // 排序
    qsort(s->sorted, s->count, sizeof(Row),
          (int(*)(const void*,const void*))s->cmp);
    s->pos = 0;
    self->started = 1;
}
int  sort_next(Operator* self, Row* out) {
    SortOp* s = (SortOp*)self;
    if (s->pos >= s->count) return 0;
    *out = s->sorted[s->pos++];
    self->rows_returned++;
    return 1;
}
void sort_close(Operator* self) {
    SortOp* s = (SortOp*)self;
    free(s->sorted);
    if (self->child) self->child->close(self->child);
}

SortOp* sort_create(Operator* child, int (*cmp)(const Row*, const Row*)) {
    SortOp* s = calloc(1, sizeof(SortOp));
    s->base.open = sort_open; s->base.next = sort_next; s->base.close = sort_close;
    s->base.child = child; s->cmp = cmp;
    return s;
}

// ===== Limit算子 =====
typedef struct {
    Operator base;
    int      limit;
    int      returned;
} LimitOp;

void limit_open(Operator* self) { ((LimitOp*)self)->returned = 0; if (self->child) self->child->open(self->child); self->started = 1; }
int  limit_next(Operator* self, Row* out) {
    LimitOp* l = (LimitOp*)self;
    if (l->returned >= l->limit) return 0;
    if (self->child->next(self->child, out)) {
        l->returned++;
        self->rows_returned++;
        return 1;
    }
    return 0;
}
void limit_close(Operator* self) { if (self->child) self->child->close(self->child); }

// ===== 谓词函数 =====
int pred_age_gt_25(Row* r) { return r->age > 25; }
int pred_salary_gt_5000(Row* r) { return r->salary > 5000; }

int cmp_salary_desc(const Row* a, const Row* b) {
    if (b->salary > a->salary) return 1;
    if (b->salary < a->salary) return -1;
    return 0;
}

void project_name_age(Row* out, Row* in) {
    *out = *in;  // 简化：直接传递
}

// ===== 向量化执行 =====
int vectorized_filter(Row* data, int n, int (*pred)(Row*), Row* out) {
    int count = 0;
    for (int i = 0; i < n; i++) {
        if (pred(&data[i])) {
            out[count++] = data[i];
        }
    }
    return count;
}

void vectorized_sort(Row* data, int n, int (*cmp)(const Row*, const Row*)) {
    qsort(data, n, sizeof(Row), (int(*)(const void*,const void*))cmp);
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   执行引擎                           ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    srand(42);
    int N = 50000;
    Row* data = malloc(N * sizeof(Row));
    for (int i = 0; i < N; i++) {
        data[i].id = i;
        snprintf(data[i].name, 32, "user_%d", i);
        data[i].age = 20 + rand() % 40;
        data[i].salary = 3000 + rand() % 10000;
    }

    // ===== 火山模型 =====
    printf("--- 火山模型执行 ---\n");
    printf("SQL: SELECT * FROM users WHERE age > 25 ORDER BY salary DESC LIMIT 10\n\n");

    clock_t t0 = clock();

    ScanOp*   scan   = scan_create(data, N, "users");
    FilterOp* filter = filter_create((Operator*)scan, pred_age_gt_25, "age>25");
    SortOp*   sort   = sort_create((Operator*)filter, cmp_salary_desc);
    LimitOp*  limit  = calloc(1, sizeof(LimitOp));
    limit->base.open = limit_open; limit->base.next = limit_next; limit->base.close = limit_close;
    limit->base.child = (Operator*)sort;
    limit->limit = 10;

    Operator* plan = (Operator*)limit;
    plan->open(plan);

    Row result;
    int count = 0;
    while (plan->next(plan, &result)) {
        printf("  id=%d name=%s age=%d salary=%.0f\n",
               result.id, result.name, result.age, result.salary);
        count++;
    }
    plan->close(plan);

    double volcano_ms = (double)(clock()-t0)/CLOCKS_PER_SEC*1000;
    printf("  火山模型: %d条结果, %.2fms\n", count, volcano_ms);

    // ===== 向量化执行 =====
    printf("\n--- 向量化执行 ---\n");
    t0 = clock();

    Row* filtered = malloc(N * sizeof(Row));
    int f_count = vectorized_filter(data, N, pred_age_gt_25, filtered);
    vectorized_sort(filtered, f_count, cmp_salary_desc);

    int v_count = f_count < 10 ? f_count : 10;
    for (int i = 0; i < v_count; i++) {
        printf("  id=%d name=%s age=%d salary=%.0f\n",
               filtered[i].id, filtered[i].name, filtered[i].age, filtered[i].salary);
    }

    double vector_ms = (double)(clock()-t0)/CLOCKS_PER_SEC*1000;
    printf("  向量化: %d条结果, %.2fms\n", v_count, vector_ms);

    printf("\n--- 性能对比 ---\n");
    printf("火山模型: %.2fms\n", volcano_ms);
    printf("向量化:   %.2fms (%.1fx)\n", vector_ms, volcano_ms/vector_ms);

    free(data); free(filtered);
    printf("\n✅ 执行引擎运行完成\n");
    return 0;
}

🐍 Python实现：三种模型对比

"""
执行模型对比: 火山 vs 向量化 vs 管道编译
"""
import time
from dataclasses import dataclass
from typing import List, Callable, Optional

@dataclass
class Row:
    id: int
    name: str
    age: int
    salary: float

# 生成数据
import random
random.seed(42)
N = 200000
data = [Row(i, f"u{i}", 20+random.randint(0,40), 3000+random.random()*10000) for i in range(N)]

# 1. 火山模型
class VolcanoOp:
    def open(self): pass
    def next(self) -> Optional[Row]: return None
    def close(self): pass

class Scan(VolcanoOp):
    def __init__(self, data): self.data = data; self.pos = 0
    def open(self): self.pos = 0
    def next(self):
        if self.pos >= len(self.data): return None
        r = self.data[self.pos]; self.pos += 1; return r

class Filter(VolcanoOp):
    def __init__(self, child, pred): self.child = child; self.pred = pred
    def open(self): self.child.open()
    def next(self):
        while True:
            r = self.child.next()
            if r is None or self.pred(r): return r

class Sort(VolcanoOp):
    def __init__(self, child, key): self.child = child; self.key = key
    def open(self):
        self.child.open()
        self.sorted = []
        while True:
            r = self.child.next()
            if r is None: break
            self.sorted.append(r)
        self.sorted.sort(key=self.key, reverse=True)
        self.pos = 0
    def next(self):
        if self.pos >= len(self.sorted): return None
        r = self.sorted[self.pos]; self.pos += 1; return r

class Limit(VolcanoOp):
    def __init__(self, child, n): self.child = child; self.n = n; self.count = 0
    def open(self): self.child.open(); self.count = 0
    def next(self):
        if self.count >= self.n: return None
        r = self.child.next()
        if r: self.count += 1
        return r

# 火山模型执行
t0 = time.perf_counter()
plan = Limit(Sort(Filter(Scan(data), lambda r: r.age > 25), lambda r: -r.salary), 10)
plan.open()
results = []
while True:
    r = plan.next()
    if r is None: break
    results.append(r)
volcano_time = (time.perf_counter() - t0) * 1000

# 2. 向量化执行
t0 = time.perf_counter()
filtered = [r for r in data if r.age > 25]
filtered.sort(key=lambda r: -r.salary)
results2 = filtered[:10]
vector_time = (time.perf_counter() - t0) * 1000

# 3. 管道式(python模拟编译执行)
t0 = time.perf_counter()
# 一次性流式处理
results3 = sorted((r for r in data if r.age > 25), key=lambda r: -r.salary)[:10]
pipe_time = (time.perf_counter() - t0) * 1000

print(f"数据量: {N}, 结果: {len(results)}条")
print(f"{'模型':>10} | {'耗时ms':>8} | {'相对':>6}")
print("-" * 35)
print(f"{'火山模型':>10} | {volcano_time:>7.2f} | {'1.0x':>6}")
print(f"{'向量化':>10} | {vector_time:>7.2f} | {volcano_time/vector_time:.1f}x")
print(f"{'管道式':>10} | {pipe_time:>7.2f} | {volcano_time/pipe_time:.1f}x")
print("\n✅ 执行模型对比完成")

🔑 关键概念总结