上一讲我们实现了查询执行引擎MiniDB 能执行 SQL 查询了。但执行方式还很原始——不管什么查询都是全表扫描。这一讲我们要实现查询优化器让 MiniDB 能做出聪明的执行决策什么时候用索引、如何组织连接顺序、如何下推过滤条件。一、优化器的工作流程SQL → 解析器 → AST → 逻辑计划 → 优化规则 → 物理计划 → 执行 ↓ ┌────────────────┐ │ 优化器核心 │ │ │ │ 1. 规则优化 │ │ · 谓词下推 │ │ · 投影下推 │ │ · 常量折叠 │ │ │ │ 2. 成本估算 │ │ · 行数估算 │ │ · I/O成本 │ │ · CPU成本 │ │ │ │ 3. 计划枚举 │ │ · 连接顺序 │ │ · 索引选择 │ └────────────────┘二、逻辑计划与物理计划2.1 逻辑算子# sql/optimizer/logical_plan.py from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import List, Optional, Any from sql.ast import Expression class LogicalNode(ABC): 逻辑计划节点 abstractmethod def get_output_columns(self) - List[str]: 获取输出的列名 pass abstractmethod def get_children(self) - List[LogicalNode]: pass dataclass class LogicalScan(LogicalNode): 逻辑扫描 table_name: str output_columns: List[str] field(default_factorylist) alias: str def get_output_columns(self): return self.output_columns def get_children(self): return [] dataclass class LogicalFilter(LogicalNode): 逻辑过滤 child: LogicalNode predicate: Expression def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child] dataclass class LogicalProjection(LogicalNode): 逻辑投影 child: LogicalNode columns: List[Expression] def get_output_columns(self): return [c.value for c in self.columns if hasattr(c, value)] def get_children(self): return [self.child] dataclass class LogicalJoin(LogicalNode): 逻辑连接 left: LogicalNode right: LogicalNode join_type: str INNER # INNER, LEFT, RIGHT condition: Optional[Expression] None def get_output_columns(self): return self.left.get_output_columns() self.right.get_output_columns() def get_children(self): return [self.left, self.right] dataclass class LogicalAggregate(LogicalNode): 逻辑聚合 child: LogicalNode group_by: List[str] aggregates: List[tuple] # [(func, column), ...] def get_output_columns(self): cols list(self.group_by) for func, col in self.aggregates: cols.append(f{func}({col})) return cols def get_children(self): return [self.child] dataclass class LogicalSort(LogicalNode): 逻辑排序 child: LogicalNode order_by: List[tuple] # [(column, asc), ...] def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child] dataclass class LogicalLimit(LogicalNode): 逻辑限制 child: LogicalNode limit: Optional[int] offset: int 0 def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child]三、优化规则3.1 规则引擎# sql/optimizer/rules.py from abc import ABC, abstractmethod from typing import List, Optional from .logical_plan import * class OptimizationRule(ABC): 优化规则基类 abstractmethod def apply(self, node: LogicalNode) - Optional[LogicalNode]: 应用规则到节点 返回优化后的节点如果规则不适用返回None pass class RuleBasedOptimizer: 基于规则的优化器 def __init__(self): self.rules: List[OptimizationRule] [ PredicatePushdown(), ProjectionPushdown(), ConstantFolding(), EliminateRedundantProjection(), MergeFilters(), ] def optimize(self, plan: LogicalNode) - LogicalNode: 反复应用规则直到收敛 changed True while changed: changed False for rule in self.rules: new_plan self._apply_rule(plan, rule) if new_plan is not plan: plan new_plan changed True return plan def _apply_rule(self, node: LogicalNode, rule: OptimizationRule) - LogicalNode: 递归应用规则 # 先应用到子节点 children node.get_children() new_children [self._apply_rule(c, rule) for c in children] # 如果子节点有变化重建节点 if new_children ! children: node self._rebuild_node(node, new_children) # 应用到当前节点 result rule.apply(node) return result if result is not None else node def _rebuild_node(self, node: LogicalNode, new_children: List[LogicalNode]) - LogicalNode: 用新的子节点重建节点 if isinstance(node, LogicalFilter): node.child new_children[0] elif isinstance(node, LogicalProjection): node.child new_children[0] elif isinstance(node, LogicalJoin): node.left, node.right new_children elif isinstance(node, LogicalAggregate): node.child new_children[0] elif isinstance(node, LogicalSort): node.child new_children[0] elif isinstance(node, LogicalLimit): node.child new_children[0] return node3.2 谓词下推class PredicatePushdown(OptimizationRule): 谓词下推 将过滤条件下推到靠近数据源的位置 尽早减少数据量 例 SELECT * FROM (SELECT * FROM users WHERE age 20) t WHERE t.age 30 → SELECT * FROM users WHERE age 20 AND age 30 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalFilter): return None child node.child # 如果子节点也是Filter合并谓词 if isinstance(child, LogicalFilter): merged self._merge_predicates(node.predicate, child.predicate) return LogicalFilter(childchild.child, predicatemerged) # 如果子节点是Projection尝试下推 if isinstance(child, LogicalProjection): # 检查谓词是否只引用了投影中保留的列 if self._predicate_uses_only_columns(node.predicate, child.columns): # 将过滤下推到投影之下 new_filter LogicalFilter(childchild.child, predicatenode.predicate) return LogicalProjection(childnew_filter, columnschild.columns) # 如果子节点是Join尝试下推到连接的一侧 if isinstance(child, LogicalJoin): left_columns child.left.get_output_columns() right_columns child.right.get_output_columns() predicate node.predicate # 分解谓词 left_pred, right_pred, join_pred self._split_predicate( predicate, left_columns, right_columns ) new_left child.left new_right child.right if left_pred is not None: new_left LogicalFilter(childchild.left, predicateleft_pred) if right_pred is not None: new_right LogicalFilter(childchild.right, predicateright_pred) new_join LogicalJoin( leftnew_left, rightnew_right, join_typechild.join_type, conditionjoin_pred or child.condition ) if join_pred is None and left_pred is None and right_pred is None: return None # 无法下推 return new_join return None def _merge_predicates(self, p1: Expression, p2: Expression) - Expression: 合并两个谓词为AND from sql.ast import Expression, ExpressionType return Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftp1, rightp2 ) def _predicate_uses_only_columns(self, predicate: Expression, columns: List[Expression]) - bool: 检查谓词是否只使用了指定列 used_columns self._extract_columns(predicate) allowed {c.value for c in columns if hasattr(c, value)} return all(c in allowed for c in used_columns) def _extract_columns(self, expr: Expression) - set: 从表达式中提取所有引用的列名 from sql.ast import ExpressionType columns set() if expr.expr_type ExpressionType.COLUMN_REF: columns.add(expr.value) elif expr.expr_type ExpressionType.BINARY_OP: columns.update(self._extract_columns(expr.left)) columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.UNARY_OP: columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.FUNCTION_CALL: for arg in expr.args: columns.update(self._extract_columns(arg)) return columns def _split_predicate(self, predicate: Expression, left_cols: List[str], right_cols: List[str]) - tuple: 将连接条件分解为 - 只涉及左表的条件 - 只涉及右表的条件 - 涉及两表的连接条件 from sql.ast import ExpressionType if predicate.expr_type ExpressionType.BINARY_OP: if predicate.operator AND: left_part, right_part, join_part self._split_predicate( predicate.left, left_cols, right_cols ) l2, r2, j2 self._split_predicate( predicate.right, left_cols, right_cols ) # 合并结果 left_result self._combine_and(left_part, l2) right_result self._combine_and(right_part, r2) join_result self._combine_and(join_part, j2) return left_result, right_result, join_result else: # 二元操作, , 等 left_used self._extract_columns(predicate.left) right_used self._extract_columns(predicate.right) all_used left_used | right_used left_set set(left_cols) right_set set(right_cols) if all_used.issubset(left_set): return predicate, None, None elif all_used.issubset(right_set): return None, predicate, None else: return None, None, predicate return None, None, predicate def _combine_and(self, p1: Optional[Expression], p2: Optional[Expression]) - Optional[Expression]: 用AND组合两个表达式 if p1 is None: return p2 if p2 is None: return p1 from sql.ast import Expression, ExpressionType return Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftp1, rightp2 )3.3 投影下推class ProjectionPushdown(OptimizationRule): 投影下推 尽早减少列的数量减少数据传输和处理的开销 例 SELECT name FROM (SELECT id, name, age FROM users) t → SELECT name FROM users def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalProjection): return None child node.child # 需要的列 needed_columns set() for col_expr in node.columns: needed_columns.update(self._extract_columns(col_expr)) # 如果子节点是Scan缩小扫描的列 if isinstance(child, LogicalScan): if set(child.output_columns) ! needed_columns: return LogicalScan( table_namechild.table_name, output_columnslist(needed_columns), aliaschild.alias ) # 如果子节点是Projection合并 if isinstance(child, LogicalProjection): # 只需要父投影需要的列 filtered_child_cols [ c for c in child.columns if hasattr(c, value) and c.value in needed_columns ] if filtered_child_cols: return LogicalProjection( childchild.child, columnsfiltered_child_cols ) return None def _extract_columns(self, expr) - set: 提取表达式引用的列 from sql.ast import ExpressionType if expr.expr_type ExpressionType.COLUMN_REF: return {expr.value} elif expr.expr_type ExpressionType.BINARY_OP: return self._extract_columns(expr.left) | self._extract_columns(expr.right) elif expr.expr_type ExpressionType.UNARY_OP: return self._extract_columns(expr.right) elif expr.expr_type ExpressionType.FUNCTION_CALL: cols set() for arg in expr.args: cols.update(self._extract_columns(arg)) return cols return set()3.4 常量折叠class ConstantFolding(OptimizationRule): 常量折叠 在编译时计算常量表达式减少运行时开销 例 WHERE age 10 5 → WHERE age 15 SELECT 2 * 3 1 → SELECT 7 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if isinstance(node, LogicalFilter): new_predicate self._fold_constants(node.predicate) if new_predicate is not node.predicate: return LogicalFilter(childnode.child, predicatenew_predicate) if isinstance(node, LogicalProjection): new_columns [self._fold_constants(c) for c in node.columns] if new_columns ! node.columns: return LogicalProjection(childnode.child, columnsnew_columns) return None def _fold_constants(self, expr) - Expression: 递归折叠常量表达式 from sql.ast import Expression, ExpressionType if expr.expr_type ExpressionType.BINARY_OP: left self._fold_constants(expr.left) right self._fold_constants(expr.right) # 如果两边都是常量直接计算 if (left.expr_type ExpressionType.LITERAL and right.expr_type ExpressionType.LITERAL): try: result self._eval_op(left.value, expr.operator, right.value) return Expression(expr_typeExpressionType.LITERAL, valueresult) except: pass if left is not expr.left or right is not expr.right: return Expression( expr_typeExpressionType.BINARY_OP, operatorexpr.operator, leftleft, rightright ) elif expr.expr_type ExpressionType.UNARY_OP: operand self._fold_constants(expr.right) if (operand.expr_type ExpressionType.LITERAL and expr.operator -): return Expression(expr_typeExpressionType.LITERAL, value-operand.value) return expr def _eval_op(self, left_val, op, right_val): 计算二元操作 if op : return left_val right_val if op -: return left_val - right_val if op *: return left_val * right_val if op /: return left_val / right_val if op : return left_val right_val if op !: return left_val ! right_val if op : return left_val right_val if op : return left_val right_val if op : return left_val right_val if op : return left_val right_val raise ValueError(fUnknown operator: {op})3.5 其他优化规则class EliminateRedundantProjection(OptimizationRule): 消除冗余投影 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalProjection): return None child node.child # 如果子节点也是Projection且列相同消除一层 if isinstance(child, LogicalProjection): child_cols {c.value for c in child.columns if hasattr(c, value)} parent_cols {c.value for c in node.columns if hasattr(c, value)} if child_cols parent_cols: return child # 如果投影保留了所有列消除投影 if isinstance(child, LogicalScan): child_cols set(child.output_columns) parent_cols {c.value for c in node.columns if hasattr(c, value)} if child_cols parent_cols: return child return None class MergeFilters(OptimizationRule): 合并连续的过滤条件 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalFilter): return None child node.child if isinstance(child, LogicalFilter): from sql.ast import Expression, ExpressionType merged Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftchild.predicate, rightnode.predicate ) return LogicalFilter(childchild.child, predicatemerged) return None四、成本估算4.1 统计信息收集# sql/optimizer/statistics.py from dataclasses import dataclass, field from typing import Dict, Optional dataclass class ColumnStats: 列统计信息 distinct_values: int 0 null_count: int 0 min_value: Optional[object] None max_value: Optional[object] None avg_width: float 0 dataclass class TableStats: 表统计信息 row_count: int 0 page_count: int 0 columns: Dict[str, ColumnStats] field(default_factorydict) class StatisticsCollector: 统计信息收集器 def __init__(self, catalog, storage): self.catalog catalog self.storage storage self.stats: Dict[str, TableStats] {} def collect(self, table_name: str): 收集表的统计信息 schema self.catalog.get_table_schema(table_name) table_stats TableStats( row_count0, page_count0, columns{} ) for col in schema.columns: table_stats.columns[col.name] ColumnStats() # 扫描数据收集统计 distinct_values {col.name: set() for col in schema.columns} row_count 0 pages self.catalog.get_table_pages(table_name) for page_id in pages: page self.storage.read_page(page_id) if page is None: continue table_stats.page_count 1 for slot in range(page.num_records): record page.read_record(slot) if record is None: continue row_count 1 # 解析记录并更新统计 # ... (简化处理) table_stats.row_count row_count for col_name, values in distinct_values.items(): if col_name in table_stats.columns: table_stats.columns[col_name].distinct_values len(values) self.stats[table_name] table_stats return table_stats def get_table_stats(self, table_name: str) - Optional[TableStats]: 获取表的统计信息 return self.stats.get(table_name) def estimate_selectivity(self, predicate) - float: 估计谓词的选择率 返回 0~1 之间的值表示满足条件的行比例 from sql.ast import ExpressionType if predicate.expr_type ExpressionType.BINARY_OP: if predicate.operator AND: return (self.estimate_selectivity(predicate.left) * self.estimate_selectivity(predicate.right)) elif predicate.operator OR: sel1 self.estimate_selectivity(predicate.left) sel2 self.estimate_selectivity(predicate.right) return sel1 sel2 - sel1 * sel2 elif predicate.operator : return 0.01 # 等值条件通常选择性较低 elif predicate.operator in (, , , ): return 0.333 # 范围条件通常选择1/3 elif predicate.operator !: return 0.667 return 0.5 # 默认选择率4.2 成本模型class CostModel: 成本模型 # 成本常数 SEQ_PAGE_COST 1.0 # 顺序读一页的成本 RANDOM_PAGE_COST 10.0 # 随机读一页的成本 CPU_TUPLE_COST 0.01 # 处理一行的CPU成本 INDEX_LOOKUP_COST 2.0 # 索引查找成本 def __init__(self, statistics: StatisticsCollector): self.statistics statistics def estimate_scan_cost(self, table_name: str) - float: 估计全表扫描成本 stats self.statistics.get_table_stats(table_name) if stats is None: return 1000 # 默认成本 # I/O成本 io_cost stats.page_count * self.SEQ_PAGE_COST # CPU成本 cpu_cost stats.row_count * self.CPU_TUPLE_COST return io_cost cpu_cost def estimate_index_cost(self, table_name: str, index_name: str, predicate) - float: 估计索引扫描成本 stats self.statistics.get_table_stats(table_name) if stats is None: return 500 selectivity self.statistics.estimate_selectivity(predicate) estimated_rows stats.row_count * selectivity # B树查找成本 tree_height 3 # 假设树高3层 lookup_cost tree_height * self.RANDOM_PAGE_COST # 读取数据页成本 data_cost estimated_rows * self.RANDOM_PAGE_COST / 100 # 假设每页100行 return lookup_cost data_cost def estimate_join_cost(self, left_rows: int, right_rows: int) - float: 估计连接成本 # Nested Loop Join nested_loop_cost left_rows * right_rows * self.CPU_TUPLE_COST # Hash Join hash_join_cost (left_rows right_rows) * self.CPU_TUPLE_COST * 2 return min(nested_loop_cost, hash_join_cost)五、索引选择5.1 索引匹配class IndexSelector: 索引选择器 def __init__(self, catalog, cost_model: CostModel): self.catalog catalog self.cost_model cost_model def choose_best_access_method(self, table_name: str, predicate) - tuple: 选择最佳访问方法 返回(access_method, estimated_cost) access_method: seq_scan 或 (index_scan, index_name) # 全表扫描的成本 seq_cost self.cost_model.estimate_scan_cost(table_name) best_method (seq_scan, None) best_cost seq_cost # 检查是否有可用索引 indexes self.catalog.get_table_indexes(table_name) for index in indexes: # 检查谓词是否能匹配索引 if self._can_use_index(predicate, index.columns): index_cost self.cost_model.estimate_index_cost( table_name, index.name, predicate ) if index_cost best_cost: best_cost index_cost best_method (index_scan, index.name) return best_method, best_cost def _can_use_index(self, predicate, index_columns: List[str]) - bool: 检查谓词是否能使用索引 from sql.ast import ExpressionType if predicate is None: return False # 提取谓词中涉及的列 used_columns self._extract_columns(predicate) # 检查是否匹配索引的前缀列 for i, col in enumerate(index_columns): if col in used_columns: return True return False def _extract_columns(self, expr) - set: 提取表达式中的列 from sql.ast import ExpressionType columns set() if expr.expr_type ExpressionType.COLUMN_REF: columns.add(expr.value) elif expr.expr_type ExpressionType.BINARY_OP: columns.update(self._extract_columns(expr.left)) columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.UNARY_OP: columns.update(self._extract_columns(expr.right)) return columns六、完整优化器# sql/optimizer/optimizer.py class Optimizer: 查询优化器 def __init__(self, catalog, storage): self.catalog catalog self.storage storage self.statistics StatisticsCollector(catalog, storage) self.cost_model CostModel(self.statistics) self.rule_optimizer RuleBasedOptimizer() self.index_selector IndexSelector(catalog, self.cost_model) def optimize(self, logical_plan: LogicalNode) - LogicalNode: 优化逻辑计划 步骤 1. 应用规则优化 2. 选择物理实现 3. 返回优化后的计划 # 1. 规则优化 optimized self.rule_optimizer.optimize(logical_plan) # 2. 索引选择 optimized self._choose_indexes(optimized) # 3. 连接顺序优化 optimized self._optimize_join_order(optimized) return optimized def _choose_indexes(self, plan: LogicalNode) - LogicalNode: 为扫描节点选择索引 if isinstance(plan, LogicalFilter): # 检查是否可以改用索引扫描 if isinstance(plan.child, LogicalScan): table_name plan.child.table_name method, cost self.index_selector.choose_best_access_method( table_name, plan.predicate ) if method[0] index_scan: # 替换为索引扫描 return IndexScanNode( table_nametable_name, index_namemethod[1], predicateplan.predicate ) # 递归处理子节点 new_child self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalFilter(childnew_child, predicateplan.predicate) elif isinstance(plan, LogicalJoin): new_left self._choose_indexes(plan.left) new_right self._choose_indexes(plan.right) if new_left is not plan.left or new_right is not plan.right: return LogicalJoin( leftnew_left, rightnew_right, join_typeplan.join_type, conditionplan.condition ) elif isinstance(plan, LogicalProjection): new_child self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalProjection(childnew_child, columnsplan.columns) return plan def _optimize_join_order(self, plan: LogicalNode) - LogicalNode: 优化连接顺序 # 收集所有参与连接的表 tables self._collect_tables(plan) if len(tables) 2: return plan # 使用贪心算法选择连接顺序 # ... (简化实现) return plan def _collect_tables(self, plan: LogicalNode) - List[str]: 收集计划中涉及的所有表 if isinstance(plan, LogicalScan): return [plan.table_name] elif isinstance(plan, LogicalJoin): return (self._collect_tables(plan.left) self._collect_tables(plan.right)) tables [] for child in plan.get_children(): tables.extend(self._collect_tables(child)) return tables class IndexScanNode(LogicalNode): 索引扫描节点物理计划 def __init__(self, table_name: str, index_name: str, predicateNone): self.table_name table_name self.index_name index_name self.predicate predicate def get_output_columns(self): return [] def get_children(self): return []七、完整演示def test_optimizer(): print( * 60) print( 查询优化器测试) print( * 60) from sql.ast import Expression, ExpressionType # 构建逻辑计划 # SELECT name, age FROM users WHERE age 25 AND age 50 # 1. 原始计划 scan LogicalScan(table_nameusers, output_columns[id, name, age]) age_col Expression(expr_typeExpressionType.COLUMN_REF, valueage) lit_25 Expression(expr_typeExpressionType.LITERAL, value25) lit_50 Expression(expr_typeExpressionType.LITERAL, value50) cond1 Expression(expr_typeExpressionType.BINARY_OP, operator, leftage_col, rightlit_25) cond2 Expression(expr_typeExpressionType.BINARY_OP, operator, leftage_col, rightlit_50) where Expression(expr_typeExpressionType.BINARY_OP, operatorAND, leftcond1, rightcond2) filter_node LogicalFilter(childscan, predicatewhere) name_col Expression(expr_typeExpressionType.COLUMN_REF, valuename) age_col2 Expression(expr_typeExpressionType.COLUMN_REF, valueage) projection LogicalProjection( childfilter_node, columns[name_col, age_col2] ) print(\n 原始逻辑计划:) print_plan(projection, 0) # 2. 应用优化 optimizer RuleBasedOptimizer() optimized optimizer.optimize(projection) print(\n 优化后逻辑计划:) print_plan(optimized, 0) # 3. 测试常量折叠 print(\n 常量折叠测试:) fold_expr Expression( expr_typeExpressionType.BINARY_OP, operator, leftExpression(expr_typeExpressionType.LITERAL, value10), rightExpression(expr_typeExpressionType.LITERAL, value5) ) folding ConstantFolding() result folding._fold_constants(fold_expr) print(f 10 5 {result.value}) # 4. 测试谓词下推 print(\n⬇️ 谓词下推测试:) inner_scan LogicalScan(table_nameusers, output_columns[id, name, age]) inner_age Expression(expr_typeExpressionType.COLUMN_REF, valueage) inner_cond Expression(expr_typeExpressionType.BINARY_OP, operator, leftinner_age, rightlit_25) inner_filter LogicalFilter(childinner_scan, predicateinner_cond) outer_proj LogicalProjection( childinner_filter, columns[name_col, age_col2] ) outer_age Expression(expr_typeExpressionType.COLUMN_REF, valueage) outer_cond Expression(expr_typeExpressionType.BINARY_OP, operator, leftouter_age, rightlit_50) outer_filter LogicalFilter(childouter_proj, predicateouter_cond) print( 优化前: Filter(age50) → Projection → Filter(age25) → Scan) pushed PredicatePushdown().apply(outer_filter) if pushed: print( 优化后:, end ) print_plan(pushed, 0) def print_plan(node: LogicalNode, indent: int): 打印逻辑计划 prefix * indent if isinstance(node, LogicalScan): print(f{prefix}Scan({node.table_name}, cols{node.output_columns})) elif isinstance(node, LogicalFilter): print(f{prefix}Filter({simplify_expr(node.predicate)})) print_plan(node.child, indent 1) elif isinstance(node, LogicalProjection): cols [c.value for c in node.columns if hasattr(c, value)] print(f{prefix}Projection({cols})) print_plan(node.child, indent 1) elif isinstance(node, LogicalJoin): print(f{prefix}Join({node.join_type})) print_plan(node.left, indent 1) print_plan(node.right, indent 1) elif isinstance(node, LogicalAggregate): print(f{prefix}Aggregate(group{node.group_by}, agg{node.aggregates})) print_plan(node.child, indent 1) elif isinstance(node, LogicalSort): print(f{prefix}Sort({node.order_by})) print_plan(node.child, indent 1) elif isinstance(node, LogicalLimit): print(f{prefix}Limit({node.limit}, offset{node.offset})) print_plan(node.child, indent 1) else: print(f{prefix}{type(node).__name__}) def simplify_expr(expr) - str: 简化表达式显示 from sql.ast import ExpressionType if expr.expr_type ExpressionType.BINARY_OP: return f({simplify_expr(expr.left)} {expr.operator} {simplify_expr(expr.right)}) elif expr.expr_type ExpressionType.COLUMN_REF: return expr.value elif expr.expr_type ExpressionType.LITERAL: return str(expr.value) return str(expr) if __name__ __main__: test_optimizer()八、总结这一讲实现了MiniDB的查询优化器逻辑计划与物理执行分离便于应用优化规则优化规则谓词下推、投影下推、常量折叠、消除冗余成本估算统计信息收集、I/O成本和CPU成本模型索引选择根据谓词匹配合适的索引规则引擎反复应用规则直到收敛现在MiniDB能做出聪明的执行决策了。下一讲将实现网络层与客户端让MiniDB成为一个可以通过网络访问的数据库服务。 第8讲：查询优化器
上一讲我们实现了查询执行引擎MiniDB 能执行 SQL 查询了。但执行方式还很原始——不管什么查询都是全表扫描。这一讲我们要实现查询优化器让 MiniDB 能做出聪明的执行决策什么时候用索引、如何组织连接顺序、如何下推过滤条件。一、优化器的工作流程SQL → 解析器 → AST → 逻辑计划 → 优化规则 → 物理计划 → 执行 ↓ ┌────────────────┐ │ 优化器核心 │ │ │ │ 1. 规则优化 │ │ · 谓词下推 │ │ · 投影下推 │ │ · 常量折叠 │ │ │ │ 2. 成本估算 │ │ · 行数估算 │ │ · I/O成本 │ │ · CPU成本 │ │ │ │ 3. 计划枚举 │ │ · 连接顺序 │ │ · 索引选择 │ └────────────────┘二、逻辑计划与物理计划2.1 逻辑算子# sql/optimizer/logical_plan.py from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import List, Optional, Any from sql.ast import Expression class LogicalNode(ABC): 逻辑计划节点 abstractmethod def get_output_columns(self) - List[str]: 获取输出的列名 pass abstractmethod def get_children(self) - List[LogicalNode]: pass dataclass class LogicalScan(LogicalNode): 逻辑扫描 table_name: str output_columns: List[str] field(default_factorylist) alias: str def get_output_columns(self): return self.output_columns def get_children(self): return [] dataclass class LogicalFilter(LogicalNode): 逻辑过滤 child: LogicalNode predicate: Expression def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child] dataclass class LogicalProjection(LogicalNode): 逻辑投影 child: LogicalNode columns: List[Expression] def get_output_columns(self): return [c.value for c in self.columns if hasattr(c, value)] def get_children(self): return [self.child] dataclass class LogicalJoin(LogicalNode): 逻辑连接 left: LogicalNode right: LogicalNode join_type: str INNER # INNER, LEFT, RIGHT condition: Optional[Expression] None def get_output_columns(self): return self.left.get_output_columns() self.right.get_output_columns() def get_children(self): return [self.left, self.right] dataclass class LogicalAggregate(LogicalNode): 逻辑聚合 child: LogicalNode group_by: List[str] aggregates: List[tuple] # [(func, column), ...] def get_output_columns(self): cols list(self.group_by) for func, col in self.aggregates: cols.append(f{func}({col})) return cols def get_children(self): return [self.child] dataclass class LogicalSort(LogicalNode): 逻辑排序 child: LogicalNode order_by: List[tuple] # [(column, asc), ...] def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child] dataclass class LogicalLimit(LogicalNode): 逻辑限制 child: LogicalNode limit: Optional[int] offset: int 0 def get_output_columns(self): return self.child.get_output_columns() def get_children(self): return [self.child]三、优化规则3.1 规则引擎# sql/optimizer/rules.py from abc import ABC, abstractmethod from typing import List, Optional from .logical_plan import * class OptimizationRule(ABC): 优化规则基类 abstractmethod def apply(self, node: LogicalNode) - Optional[LogicalNode]: 应用规则到节点 返回优化后的节点如果规则不适用返回None pass class RuleBasedOptimizer: 基于规则的优化器 def __init__(self): self.rules: List[OptimizationRule] [ PredicatePushdown(), ProjectionPushdown(), ConstantFolding(), EliminateRedundantProjection(), MergeFilters(), ] def optimize(self, plan: LogicalNode) - LogicalNode: 反复应用规则直到收敛 changed True while changed: changed False for rule in self.rules: new_plan self._apply_rule(plan, rule) if new_plan is not plan: plan new_plan changed True return plan def _apply_rule(self, node: LogicalNode, rule: OptimizationRule) - LogicalNode: 递归应用规则 # 先应用到子节点 children node.get_children() new_children [self._apply_rule(c, rule) for c in children] # 如果子节点有变化重建节点 if new_children ! children: node self._rebuild_node(node, new_children) # 应用到当前节点 result rule.apply(node) return result if result is not None else node def _rebuild_node(self, node: LogicalNode, new_children: List[LogicalNode]) - LogicalNode: 用新的子节点重建节点 if isinstance(node, LogicalFilter): node.child new_children[0] elif isinstance(node, LogicalProjection): node.child new_children[0] elif isinstance(node, LogicalJoin): node.left, node.right new_children elif isinstance(node, LogicalAggregate): node.child new_children[0] elif isinstance(node, LogicalSort): node.child new_children[0] elif isinstance(node, LogicalLimit): node.child new_children[0] return node3.2 谓词下推class PredicatePushdown(OptimizationRule): 谓词下推 将过滤条件下推到靠近数据源的位置 尽早减少数据量 例 SELECT * FROM (SELECT * FROM users WHERE age 20) t WHERE t.age 30 → SELECT * FROM users WHERE age 20 AND age 30 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalFilter): return None child node.child # 如果子节点也是Filter合并谓词 if isinstance(child, LogicalFilter): merged self._merge_predicates(node.predicate, child.predicate) return LogicalFilter(childchild.child, predicatemerged) # 如果子节点是Projection尝试下推 if isinstance(child, LogicalProjection): # 检查谓词是否只引用了投影中保留的列 if self._predicate_uses_only_columns(node.predicate, child.columns): # 将过滤下推到投影之下 new_filter LogicalFilter(childchild.child, predicatenode.predicate) return LogicalProjection(childnew_filter, columnschild.columns) # 如果子节点是Join尝试下推到连接的一侧 if isinstance(child, LogicalJoin): left_columns child.left.get_output_columns() right_columns child.right.get_output_columns() predicate node.predicate # 分解谓词 left_pred, right_pred, join_pred self._split_predicate( predicate, left_columns, right_columns ) new_left child.left new_right child.right if left_pred is not None: new_left LogicalFilter(childchild.left, predicateleft_pred) if right_pred is not None: new_right LogicalFilter(childchild.right, predicateright_pred) new_join LogicalJoin( leftnew_left, rightnew_right, join_typechild.join_type, conditionjoin_pred or child.condition ) if join_pred is None and left_pred is None and right_pred is None: return None # 无法下推 return new_join return None def _merge_predicates(self, p1: Expression, p2: Expression) - Expression: 合并两个谓词为AND from sql.ast import Expression, ExpressionType return Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftp1, rightp2 ) def _predicate_uses_only_columns(self, predicate: Expression, columns: List[Expression]) - bool: 检查谓词是否只使用了指定列 used_columns self._extract_columns(predicate) allowed {c.value for c in columns if hasattr(c, value)} return all(c in allowed for c in used_columns) def _extract_columns(self, expr: Expression) - set: 从表达式中提取所有引用的列名 from sql.ast import ExpressionType columns set() if expr.expr_type ExpressionType.COLUMN_REF: columns.add(expr.value) elif expr.expr_type ExpressionType.BINARY_OP: columns.update(self._extract_columns(expr.left)) columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.UNARY_OP: columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.FUNCTION_CALL: for arg in expr.args: columns.update(self._extract_columns(arg)) return columns def _split_predicate(self, predicate: Expression, left_cols: List[str], right_cols: List[str]) - tuple: 将连接条件分解为 - 只涉及左表的条件 - 只涉及右表的条件 - 涉及两表的连接条件 from sql.ast import ExpressionType if predicate.expr_type ExpressionType.BINARY_OP: if predicate.operator AND: left_part, right_part, join_part self._split_predicate( predicate.left, left_cols, right_cols ) l2, r2, j2 self._split_predicate( predicate.right, left_cols, right_cols ) # 合并结果 left_result self._combine_and(left_part, l2) right_result self._combine_and(right_part, r2) join_result self._combine_and(join_part, j2) return left_result, right_result, join_result else: # 二元操作, , 等 left_used self._extract_columns(predicate.left) right_used self._extract_columns(predicate.right) all_used left_used | right_used left_set set(left_cols) right_set set(right_cols) if all_used.issubset(left_set): return predicate, None, None elif all_used.issubset(right_set): return None, predicate, None else: return None, None, predicate return None, None, predicate def _combine_and(self, p1: Optional[Expression], p2: Optional[Expression]) - Optional[Expression]: 用AND组合两个表达式 if p1 is None: return p2 if p2 is None: return p1 from sql.ast import Expression, ExpressionType return Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftp1, rightp2 )3.3 投影下推class ProjectionPushdown(OptimizationRule): 投影下推 尽早减少列的数量减少数据传输和处理的开销 例 SELECT name FROM (SELECT id, name, age FROM users) t → SELECT name FROM users def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalProjection): return None child node.child # 需要的列 needed_columns set() for col_expr in node.columns: needed_columns.update(self._extract_columns(col_expr)) # 如果子节点是Scan缩小扫描的列 if isinstance(child, LogicalScan): if set(child.output_columns) ! needed_columns: return LogicalScan( table_namechild.table_name, output_columnslist(needed_columns), aliaschild.alias ) # 如果子节点是Projection合并 if isinstance(child, LogicalProjection): # 只需要父投影需要的列 filtered_child_cols [ c for c in child.columns if hasattr(c, value) and c.value in needed_columns ] if filtered_child_cols: return LogicalProjection( childchild.child, columnsfiltered_child_cols ) return None def _extract_columns(self, expr) - set: 提取表达式引用的列 from sql.ast import ExpressionType if expr.expr_type ExpressionType.COLUMN_REF: return {expr.value} elif expr.expr_type ExpressionType.BINARY_OP: return self._extract_columns(expr.left) | self._extract_columns(expr.right) elif expr.expr_type ExpressionType.UNARY_OP: return self._extract_columns(expr.right) elif expr.expr_type ExpressionType.FUNCTION_CALL: cols set() for arg in expr.args: cols.update(self._extract_columns(arg)) return cols return set()3.4 常量折叠class ConstantFolding(OptimizationRule): 常量折叠 在编译时计算常量表达式减少运行时开销 例 WHERE age 10 5 → WHERE age 15 SELECT 2 * 3 1 → SELECT 7 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if isinstance(node, LogicalFilter): new_predicate self._fold_constants(node.predicate) if new_predicate is not node.predicate: return LogicalFilter(childnode.child, predicatenew_predicate) if isinstance(node, LogicalProjection): new_columns [self._fold_constants(c) for c in node.columns] if new_columns ! node.columns: return LogicalProjection(childnode.child, columnsnew_columns) return None def _fold_constants(self, expr) - Expression: 递归折叠常量表达式 from sql.ast import Expression, ExpressionType if expr.expr_type ExpressionType.BINARY_OP: left self._fold_constants(expr.left) right self._fold_constants(expr.right) # 如果两边都是常量直接计算 if (left.expr_type ExpressionType.LITERAL and right.expr_type ExpressionType.LITERAL): try: result self._eval_op(left.value, expr.operator, right.value) return Expression(expr_typeExpressionType.LITERAL, valueresult) except: pass if left is not expr.left or right is not expr.right: return Expression( expr_typeExpressionType.BINARY_OP, operatorexpr.operator, leftleft, rightright ) elif expr.expr_type ExpressionType.UNARY_OP: operand self._fold_constants(expr.right) if (operand.expr_type ExpressionType.LITERAL and expr.operator -): return Expression(expr_typeExpressionType.LITERAL, value-operand.value) return expr def _eval_op(self, left_val, op, right_val): 计算二元操作 if op : return left_val right_val if op -: return left_val - right_val if op *: return left_val * right_val if op /: return left_val / right_val if op : return left_val right_val if op !: return left_val ! right_val if op : return left_val right_val if op : return left_val right_val if op : return left_val right_val if op : return left_val right_val raise ValueError(fUnknown operator: {op})3.5 其他优化规则class EliminateRedundantProjection(OptimizationRule): 消除冗余投影 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalProjection): return None child node.child # 如果子节点也是Projection且列相同消除一层 if isinstance(child, LogicalProjection): child_cols {c.value for c in child.columns if hasattr(c, value)} parent_cols {c.value for c in node.columns if hasattr(c, value)} if child_cols parent_cols: return child # 如果投影保留了所有列消除投影 if isinstance(child, LogicalScan): child_cols set(child.output_columns) parent_cols {c.value for c in node.columns if hasattr(c, value)} if child_cols parent_cols: return child return None class MergeFilters(OptimizationRule): 合并连续的过滤条件 def apply(self, node: LogicalNode) - Optional[LogicalNode]: if not isinstance(node, LogicalFilter): return None child node.child if isinstance(child, LogicalFilter): from sql.ast import Expression, ExpressionType merged Expression( expr_typeExpressionType.BINARY_OP, operatorAND, leftchild.predicate, rightnode.predicate ) return LogicalFilter(childchild.child, predicatemerged) return None四、成本估算4.1 统计信息收集# sql/optimizer/statistics.py from dataclasses import dataclass, field from typing import Dict, Optional dataclass class ColumnStats: 列统计信息 distinct_values: int 0 null_count: int 0 min_value: Optional[object] None max_value: Optional[object] None avg_width: float 0 dataclass class TableStats: 表统计信息 row_count: int 0 page_count: int 0 columns: Dict[str, ColumnStats] field(default_factorydict) class StatisticsCollector: 统计信息收集器 def __init__(self, catalog, storage): self.catalog catalog self.storage storage self.stats: Dict[str, TableStats] {} def collect(self, table_name: str): 收集表的统计信息 schema self.catalog.get_table_schema(table_name) table_stats TableStats( row_count0, page_count0, columns{} ) for col in schema.columns: table_stats.columns[col.name] ColumnStats() # 扫描数据收集统计 distinct_values {col.name: set() for col in schema.columns} row_count 0 pages self.catalog.get_table_pages(table_name) for page_id in pages: page self.storage.read_page(page_id) if page is None: continue table_stats.page_count 1 for slot in range(page.num_records): record page.read_record(slot) if record is None: continue row_count 1 # 解析记录并更新统计 # ... (简化处理) table_stats.row_count row_count for col_name, values in distinct_values.items(): if col_name in table_stats.columns: table_stats.columns[col_name].distinct_values len(values) self.stats[table_name] table_stats return table_stats def get_table_stats(self, table_name: str) - Optional[TableStats]: 获取表的统计信息 return self.stats.get(table_name) def estimate_selectivity(self, predicate) - float: 估计谓词的选择率 返回 0~1 之间的值表示满足条件的行比例 from sql.ast import ExpressionType if predicate.expr_type ExpressionType.BINARY_OP: if predicate.operator AND: return (self.estimate_selectivity(predicate.left) * self.estimate_selectivity(predicate.right)) elif predicate.operator OR: sel1 self.estimate_selectivity(predicate.left) sel2 self.estimate_selectivity(predicate.right) return sel1 sel2 - sel1 * sel2 elif predicate.operator : return 0.01 # 等值条件通常选择性较低 elif predicate.operator in (, , , ): return 0.333 # 范围条件通常选择1/3 elif predicate.operator !: return 0.667 return 0.5 # 默认选择率4.2 成本模型class CostModel: 成本模型 # 成本常数 SEQ_PAGE_COST 1.0 # 顺序读一页的成本 RANDOM_PAGE_COST 10.0 # 随机读一页的成本 CPU_TUPLE_COST 0.01 # 处理一行的CPU成本 INDEX_LOOKUP_COST 2.0 # 索引查找成本 def __init__(self, statistics: StatisticsCollector): self.statistics statistics def estimate_scan_cost(self, table_name: str) - float: 估计全表扫描成本 stats self.statistics.get_table_stats(table_name) if stats is None: return 1000 # 默认成本 # I/O成本 io_cost stats.page_count * self.SEQ_PAGE_COST # CPU成本 cpu_cost stats.row_count * self.CPU_TUPLE_COST return io_cost cpu_cost def estimate_index_cost(self, table_name: str, index_name: str, predicate) - float: 估计索引扫描成本 stats self.statistics.get_table_stats(table_name) if stats is None: return 500 selectivity self.statistics.estimate_selectivity(predicate) estimated_rows stats.row_count * selectivity # B树查找成本 tree_height 3 # 假设树高3层 lookup_cost tree_height * self.RANDOM_PAGE_COST # 读取数据页成本 data_cost estimated_rows * self.RANDOM_PAGE_COST / 100 # 假设每页100行 return lookup_cost data_cost def estimate_join_cost(self, left_rows: int, right_rows: int) - float: 估计连接成本 # Nested Loop Join nested_loop_cost left_rows * right_rows * self.CPU_TUPLE_COST # Hash Join hash_join_cost (left_rows right_rows) * self.CPU_TUPLE_COST * 2 return min(nested_loop_cost, hash_join_cost)五、索引选择5.1 索引匹配class IndexSelector: 索引选择器 def __init__(self, catalog, cost_model: CostModel): self.catalog catalog self.cost_model cost_model def choose_best_access_method(self, table_name: str, predicate) - tuple: 选择最佳访问方法 返回(access_method, estimated_cost) access_method: seq_scan 或 (index_scan, index_name) # 全表扫描的成本 seq_cost self.cost_model.estimate_scan_cost(table_name) best_method (seq_scan, None) best_cost seq_cost # 检查是否有可用索引 indexes self.catalog.get_table_indexes(table_name) for index in indexes: # 检查谓词是否能匹配索引 if self._can_use_index(predicate, index.columns): index_cost self.cost_model.estimate_index_cost( table_name, index.name, predicate ) if index_cost best_cost: best_cost index_cost best_method (index_scan, index.name) return best_method, best_cost def _can_use_index(self, predicate, index_columns: List[str]) - bool: 检查谓词是否能使用索引 from sql.ast import ExpressionType if predicate is None: return False # 提取谓词中涉及的列 used_columns self._extract_columns(predicate) # 检查是否匹配索引的前缀列 for i, col in enumerate(index_columns): if col in used_columns: return True return False def _extract_columns(self, expr) - set: 提取表达式中的列 from sql.ast import ExpressionType columns set() if expr.expr_type ExpressionType.COLUMN_REF: columns.add(expr.value) elif expr.expr_type ExpressionType.BINARY_OP: columns.update(self._extract_columns(expr.left)) columns.update(self._extract_columns(expr.right)) elif expr.expr_type ExpressionType.UNARY_OP: columns.update(self._extract_columns(expr.right)) return columns六、完整优化器# sql/optimizer/optimizer.py class Optimizer: 查询优化器 def __init__(self, catalog, storage): self.catalog catalog self.storage storage self.statistics StatisticsCollector(catalog, storage) self.cost_model CostModel(self.statistics) self.rule_optimizer RuleBasedOptimizer() self.index_selector IndexSelector(catalog, self.cost_model) def optimize(self, logical_plan: LogicalNode) - LogicalNode: 优化逻辑计划 步骤 1. 应用规则优化 2. 选择物理实现 3. 返回优化后的计划 # 1. 规则优化 optimized self.rule_optimizer.optimize(logical_plan) # 2. 索引选择 optimized self._choose_indexes(optimized) # 3. 连接顺序优化 optimized self._optimize_join_order(optimized) return optimized def _choose_indexes(self, plan: LogicalNode) - LogicalNode: 为扫描节点选择索引 if isinstance(plan, LogicalFilter): # 检查是否可以改用索引扫描 if isinstance(plan.child, LogicalScan): table_name plan.child.table_name method, cost self.index_selector.choose_best_access_method( table_name, plan.predicate ) if method[0] index_scan: # 替换为索引扫描 return IndexScanNode( table_nametable_name, index_namemethod[1], predicateplan.predicate ) # 递归处理子节点 new_child self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalFilter(childnew_child, predicateplan.predicate) elif isinstance(plan, LogicalJoin): new_left self._choose_indexes(plan.left) new_right self._choose_indexes(plan.right) if new_left is not plan.left or new_right is not plan.right: return LogicalJoin( leftnew_left, rightnew_right, join_typeplan.join_type, conditionplan.condition ) elif isinstance(plan, LogicalProjection): new_child self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalProjection(childnew_child, columnsplan.columns) return plan def _optimize_join_order(self, plan: LogicalNode) - LogicalNode: 优化连接顺序 # 收集所有参与连接的表 tables self._collect_tables(plan) if len(tables) 2: return plan # 使用贪心算法选择连接顺序 # ... (简化实现) return plan def _collect_tables(self, plan: LogicalNode) - List[str]: 收集计划中涉及的所有表 if isinstance(plan, LogicalScan): return [plan.table_name] elif isinstance(plan, LogicalJoin): return (self._collect_tables(plan.left) self._collect_tables(plan.right)) tables [] for child in plan.get_children(): tables.extend(self._collect_tables(child)) return tables class IndexScanNode(LogicalNode): 索引扫描节点物理计划 def __init__(self, table_name: str, index_name: str, predicateNone): self.table_name table_name self.index_name index_name self.predicate predicate def get_output_columns(self): return [] def get_children(self): return []七、完整演示def test_optimizer(): print( * 60) print( 查询优化器测试) print( * 60) from sql.ast import Expression, ExpressionType # 构建逻辑计划 # SELECT name, age FROM users WHERE age 25 AND age 50 # 1. 原始计划 scan LogicalScan(table_nameusers, output_columns[id, name, age]) age_col Expression(expr_typeExpressionType.COLUMN_REF, valueage) lit_25 Expression(expr_typeExpressionType.LITERAL, value25) lit_50 Expression(expr_typeExpressionType.LITERAL, value50) cond1 Expression(expr_typeExpressionType.BINARY_OP, operator, leftage_col, rightlit_25) cond2 Expression(expr_typeExpressionType.BINARY_OP, operator, leftage_col, rightlit_50) where Expression(expr_typeExpressionType.BINARY_OP, operatorAND, leftcond1, rightcond2) filter_node LogicalFilter(childscan, predicatewhere) name_col Expression(expr_typeExpressionType.COLUMN_REF, valuename) age_col2 Expression(expr_typeExpressionType.COLUMN_REF, valueage) projection LogicalProjection( childfilter_node, columns[name_col, age_col2] ) print(\n 原始逻辑计划:) print_plan(projection, 0) # 2. 应用优化 optimizer RuleBasedOptimizer() optimized optimizer.optimize(projection) print(\n 优化后逻辑计划:) print_plan(optimized, 0) # 3. 测试常量折叠 print(\n 常量折叠测试:) fold_expr Expression( expr_typeExpressionType.BINARY_OP, operator, leftExpression(expr_typeExpressionType.LITERAL, value10), rightExpression(expr_typeExpressionType.LITERAL, value5) ) folding ConstantFolding() result folding._fold_constants(fold_expr) print(f 10 5 {result.value}) # 4. 测试谓词下推 print(\n⬇️ 谓词下推测试:) inner_scan LogicalScan(table_nameusers, output_columns[id, name, age]) inner_age Expression(expr_typeExpressionType.COLUMN_REF, valueage) inner_cond Expression(expr_typeExpressionType.BINARY_OP, operator, leftinner_age, rightlit_25) inner_filter LogicalFilter(childinner_scan, predicateinner_cond) outer_proj LogicalProjection( childinner_filter, columns[name_col, age_col2] ) outer_age Expression(expr_typeExpressionType.COLUMN_REF, valueage) outer_cond Expression(expr_typeExpressionType.BINARY_OP, operator, leftouter_age, rightlit_50) outer_filter LogicalFilter(childouter_proj, predicateouter_cond) print( 优化前: Filter(age50) → Projection → Filter(age25) → Scan) pushed PredicatePushdown().apply(outer_filter) if pushed: print( 优化后:, end ) print_plan(pushed, 0) def print_plan(node: LogicalNode, indent: int): 打印逻辑计划 prefix * indent if isinstance(node, LogicalScan): print(f{prefix}Scan({node.table_name}, cols{node.output_columns})) elif isinstance(node, LogicalFilter): print(f{prefix}Filter({simplify_expr(node.predicate)})) print_plan(node.child, indent 1) elif isinstance(node, LogicalProjection): cols [c.value for c in node.columns if hasattr(c, value)] print(f{prefix}Projection({cols})) print_plan(node.child, indent 1) elif isinstance(node, LogicalJoin): print(f{prefix}Join({node.join_type})) print_plan(node.left, indent 1) print_plan(node.right, indent 1) elif isinstance(node, LogicalAggregate): print(f{prefix}Aggregate(group{node.group_by}, agg{node.aggregates})) print_plan(node.child, indent 1) elif isinstance(node, LogicalSort): print(f{prefix}Sort({node.order_by})) print_plan(node.child, indent 1) elif isinstance(node, LogicalLimit): print(f{prefix}Limit({node.limit}, offset{node.offset})) print_plan(node.child, indent 1) else: print(f{prefix}{type(node).__name__}) def simplify_expr(expr) - str: 简化表达式显示 from sql.ast import ExpressionType if expr.expr_type ExpressionType.BINARY_OP: return f({simplify_expr(expr.left)} {expr.operator} {simplify_expr(expr.right)}) elif expr.expr_type ExpressionType.COLUMN_REF: return expr.value elif expr.expr_type ExpressionType.LITERAL: return str(expr.value) return str(expr) if __name__ __main__: test_optimizer()八、总结这一讲实现了MiniDB的查询优化器逻辑计划与物理执行分离便于应用优化规则优化规则谓词下推、投影下推、常量折叠、消除冗余成本估算统计信息收集、I/O成本和CPU成本模型索引选择根据谓词匹配合适的索引规则引擎反复应用规则直到收敛现在MiniDB能做出聪明的执行决策了。下一讲将实现网络层与客户端让MiniDB成为一个可以通过网络访问的数据库服务。