当前位置：首页 > news >正文

第8讲：查询优化器

news 2026/8/13 10:02:12

上一讲我们实现了查询执行引擎，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_factory=list) 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 node

3.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(child=child.child, predicate=merged) # 如果子节点是Projection，尝试下推 if isinstance(child, LogicalProjection): # 检查谓词是否只引用了投影中保留的列 if self._predicate_uses_only_columns(node.predicate, child.columns): # 将过滤下推到投影之下 new_filter = LogicalFilter(child=child.child, predicate=node.predicate) return LogicalProjection(child=new_filter, columns=child.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(child=child.left, predicate=left_pred) if right_pred is not None: new_right = LogicalFilter(child=child.right, predicate=right_pred) new_join = LogicalJoin( left=new_left, right=new_right, join_type=child.join_type, condition=join_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_type=ExpressionType.BINARY_OP, operator='AND', left=p1, right=p2 ) 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_type=ExpressionType.BINARY_OP, operator='AND', left=p1, right=p2 )

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_name=child.table_name, output_columns=list(needed_columns), alias=child.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( child=child.child, columns=filtered_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(child=node.child, predicate=new_predicate) if isinstance(node, LogicalProjection): new_columns = [self._fold_constants(c) for c in node.columns] if new_columns != node.columns: return LogicalProjection(child=node.child, columns=new_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_type=ExpressionType.LITERAL, value=result) except: pass if left is not expr.left or right is not expr.right: return Expression( expr_type=ExpressionType.BINARY_OP, operator=expr.operator, left=left, right=right ) 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_type=ExpressionType.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(f"Unknown 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_type=ExpressionType.BINARY_OP, operator='AND', left=child.predicate, right=node.predicate ) return LogicalFilter(child=child.child, predicate=merged) 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_factory=dict) 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_count=0, page_count=0, 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_name=table_name, index_name=method[1], predicate=plan.predicate ) # 递归处理子节点 new_child = self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalFilter(child=new_child, predicate=plan.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( left=new_left, right=new_right, join_type=plan.join_type, condition=plan.condition ) elif isinstance(plan, LogicalProjection): new_child = self._choose_indexes(plan.child) if new_child is not plan.child: return LogicalProjection(child=new_child, columns=plan.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, predicate=None): 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_name='users', output_columns=['id', 'name', 'age']) age_col = Expression(expr_type=ExpressionType.COLUMN_REF, value='age') lit_25 = Expression(expr_type=ExpressionType.LITERAL, value=25) lit_50 = Expression(expr_type=ExpressionType.LITERAL, value=50) cond1 = Expression(expr_type=ExpressionType.BINARY_OP, operator='>', left=age_col, right=lit_25) cond2 = Expression(expr_type=ExpressionType.BINARY_OP, operator='<', left=age_col, right=lit_50) where = Expression(expr_type=ExpressionType.BINARY_OP, operator='AND', left=cond1, right=cond2) filter_node = LogicalFilter(child=scan, predicate=where) name_col = Expression(expr_type=ExpressionType.COLUMN_REF, value='name') age_col2 = Expression(expr_type=ExpressionType.COLUMN_REF, value='age') projection = LogicalProjection( child=filter_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_type=ExpressionType.BINARY_OP, operator='+', left=Expression(expr_type=ExpressionType.LITERAL, value=10), right=Expression(expr_type=ExpressionType.LITERAL, value=5) ) folding = ConstantFolding() result = folding._fold_constants(fold_expr) print(f" 10 + 5 = {result.value}") # 4. 测试谓词下推 print("\n⬇️ 谓词下推测试:") inner_scan = LogicalScan(table_name='users', output_columns=['id', 'name', 'age']) inner_age = Expression(expr_type=ExpressionType.COLUMN_REF, value='age') inner_cond = Expression(expr_type=ExpressionType.BINARY_OP, operator='>', left=inner_age, right=lit_25) inner_filter = LogicalFilter(child=inner_scan, predicate=inner_cond) outer_proj = LogicalProjection( child=inner_filter, columns=[name_col, age_col2] ) outer_age = Expression(expr_type=ExpressionType.COLUMN_REF, value='age') outer_cond = Expression(expr_type=ExpressionType.BINARY_OP, operator='<', left=outer_age, right=lit_50) outer_filter = LogicalFilter(child=outer_proj, predicate=outer_cond) print(" 优化前: Filter(age<50) → Projection → Filter(age>25) → 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()