1. 向量化执行引擎基础框架搭建

任务说明

本案例目标为搭建一个最小可运行的向量化执行引擎基础框架，核心任务包括：定义向量化数据块（VectorBlock）、实现基础向量算子（扫描与投影）、构建支持算子链式调用的执行流水线，并对比向量化执行与逐行执行的调用方式差异。最终通过一组样例数据完成一次完整查询，输出可校验的结果集。

核心设计

向量化执行引擎以“批处理”为基本处理单元，每次算子调用处理一批数据而非单行。本框架包含三个核心组件：

• VectorBlock：列式存储的数据块，每列为一个连续数组，附带行数信息；
• VectorOperator：算子抽象基类，定义next()接口，返回下一个数据块或空块表示结束；
• Pipeline：将多个算子串联为执行链，驱动数据从源头流向终点。

代码实现

import numpy as np
from typing import List, Optional, Tuple

class VectorBlock:
    """向量化数据块：列式存储，每列一个numpy数组"""
    def __init__(self, columns: dict):
        self.columns = columns
        self.row_count = len(next(iter(columns.values()))) if columns else 0
    
    def get_column(self, name: str) -> np.ndarray:
        return self.columns[name]

class VectorOperator:
    """算子基类"""
    def __init__(self):
        self.children: List[VectorOperator] = []
    
    def add_child(self, op: 'VectorOperator'):
        self.children.append(op)
        return self
    
    def next(self) -> Optional[VectorBlock]:
        raise NotImplementedError

class ScanOperator(VectorOperator):
    """扫描算子：将内存中的原始列数据包装为VectorBlock"""
    def __init__(self, data: dict):
        super().__init__()
        self.data = data
        self._emitted = False
    
    def next(self) -> Optional[VectorBlock]:
        if not self._emitted:
            self._emitted = True
            return VectorBlock(self.data)
        return None

class ProjectOperator(VectorOperator):
    """投影算子：从输入块中选取指定列，并可进行表达式计算"""
    def __init__(self, projections: List[Tuple[str, str]]):
        """
        projections: [(输出列名, 输入列名或表达式), ...]
        表达式支持 'col_name' 或 'col_name * 2' 形式
        """
        super().__init__()
        self.projections = projections
    
    def next(self) -> Optional[VectorBlock]:
        child = self.children[0]
        block = child.next()
        if block is None:
            return None
        
        result = {}
        for out_name, expr in self.projections:
            if ' ' not in expr:  # 纯列名
                result[out_name] = block.get_column(expr).copy()
            else:  # 简单表达式：仅支持单列与标量运算
                parts = expr.split()
                col_name = parts[0]
                op = parts[1]
                scalar = float(parts[2])
                col_data = block.get_column(col_name)
                if op == '*':
                    result[out_name] = col_data * scalar
                elif op == '+':
                    result[out_name] = col_data + scalar
                else:
                    raise ValueError(f"不支持的运算符: {op}")
        return VectorBlock(result)

class Pipeline:
    """执行流水线：驱动算子链"""
    def __init__(self, root: VectorOperator):
        self.root = root
    
    def execute(self) -> List[VectorBlock]:
        results = []
        while True:
            block = self.root.next()
            if block is None:
                break
            results.append(block)
        return results

def run_vectorized():
    """向量化执行入口"""
    # 样例数据：4行3列
    source_data = {
        'id': np.array([1, 2, 3, 4]),
        'price': np.array([10.5, 20.0, 15.5, 30.0]),
        'qty': np.array([2, 3, 1, 4])
    }
    
    # 构建算子链: scan -> project(计算 amount = price * qty, 并选取 id)
    scan = ScanOperator(source_data)
    project = ProjectOperator([
        ('id', 'id'),
        ('amount', 'price * qty')  # 注意：当前表达式解析仅支持单列与标量运算
    ])
    project.add_child(scan)
    
    pipeline = Pipeline(project)
    blocks = pipeline.execute()
    
    # 汇总输出
    all_ids = []
    all_amounts = []
    for block in blocks:
        all_ids.extend(block.get_column('id').tolist())
        all_amounts.extend(block.get_column('amount').tolist())
    
    return all_ids, all_amounts

def run_row_by_row():
    """逐行执行（对照实现）"""
    source_data = {
        'id': [1, 2, 3, 4],
        'price': [10.5, 20.0, 15.5, 30.0],
        'qty': [2, 3, 1, 4]
    }
    ids = []
    amounts = []
    for i in range(len(source_data['id'])):
        ids.append(source_data['id'][i])
        amounts.append(source_data['price'][i] * source_data['qty'][i])
    return ids, amounts

if __name__ == '__main__':
    vec_ids, vec_amounts = run_vectorized()
    row_ids, row_amounts = run_row_by_row()
    
    print("向量化执行结果:")
    for i in range(len(vec_ids)):
        print(f"  id={vec_ids[i]}, amount={vec_amounts[i]}")
    
    print("\n逐行执行结果:")
    for i in range(len(row_ids)):
        print(f"  id={row_ids[i]}, amount={row_amounts[i]}")
    
    # 验证一致性
    assert vec_ids == row_ids, "id列结果不一致"
    assert vec_amounts == row_amounts, "amount列结果不一致"
    print("\n验证通过：向量化与逐行执行结果一致")

落地与运行

本案例仅依赖Python标准库与NumPy。将上述代码保存为vector_engine.py，在命令行执行：

python vector_engine.py

验证方法

运行程序后观察输出：

• 程序应打印4行向量化执行结果（id=1~4，amount分别为21.0、60.0、15.5、120.0）；
• 随后打印4行逐行执行结果，两段输出应完全一致；
• 程序末尾输出“验证通过：向量化与逐行执行结果一致”，且无异常退出，即判定案例通过。

预期输出

向量化执行结果:
  id=1, amount=21.0
  id=2, amount=60.0
  id=3, amount=15.5
  id=4, amount=120.0

逐行执行结果:
  id=1, amount=21.0
  id=2, amount=60.0
  id=3, amount=15.5
  id=4, amount=120.0

验证通过：向量化与逐行执行结果一致

以上输出由代码逻辑直接推出：amount = price * qty，四行数据分别计算为10.5×2=21.0、20.0×3=60.0、15.5×1=15.5、30.0×4=120.0。运行程序后若看到上述内容且断言通过，即表示向量化执行框架搭建成功。