查询优化器：从 System R 到 Cascades

-- orders:    10,000,000 行
-- customers:    500,000 行
-- products:     100,000 行

SELECT c.name, p.title, SUM(o.amount)
FROM   orders o
  JOIN customers c ON o.cust_id = c.id
  JOIN products  p ON o.prod_id = p.id
WHERE  c.country = 'CN'
  AND  p.category = 'electronics'
GROUP BY c.name, p.title;

SELECT * FROM orders o JOIN customers c ON o.cust_id = c.id
ORDER BY c.name;

# 每个子问题维护多个计划（有趣顺序）
best_plans = {}  # key: (table_set, order) -> plan
# ({customers}, order_by_name) -> IndexScan(customers, name_idx)
# ({customers}, no_order)      -> SeqScan(customers)

-- PostgreSQL 中查看统计信息
SELECT attname, n_distinct, most_common_vals, most_common_freqs
FROM   pg_stats
WHERE  tablename = 'orders' AND attname = 'status';

-- 输出示例：
-- most_common_vals:  {pending, shipped, delivered, cancelled}
-- most_common_freqs: {0.35, 0.30, 0.25, 0.10}

WHERE city = 'Beijing' AND country = 'CN'

-- PostgreSQL 多列统计
CREATE STATISTICS city_country_stats (dependencies)
    ON city, country FROM addresses;

ANALYZE addresses;

-- PostgreSQL 默认代价参数
SHOW seq_page_cost;       -- 1.0（基准）
SHOW random_page_cost;    -- 4.0（随机I/O是顺序的4倍）
SHOW cpu_tuple_cost;      -- 0.01
SHOW cpu_index_tuple_cost;-- 0.005
SHOW cpu_operator_cost;   -- 0.0025
SHOW parallel_tuple_cost; -- 0.1
SHOW effective_cache_size;-- 4GB（影响索引扫描代价估算）

-- 关键代价常数
-- server_cost:  row_evaluate_cost=0.1, key_compare_cost=0.05
-- engine_cost:  io_block_read_cost=1.0, memory_block_read_cost=0.25

UPDATE mysql.engine_cost SET cost_value = 0.5
WHERE  cost_name = 'io_block_read_cost';
FLUSH OPTIMIZER_COSTS;

def dp_join_order(tables, edges):
    """
    tables: 表的列表
    edges:  连接谓词 (表i, 表j) -> 谓词
    返回:   最优计划及其代价
    """
    n = len(tables)
    dp = {}  # bitmask -> (cost, plan)

    # 基础：单表
    for i in range(n):
        mask = 1 << i
        dp[mask] = (scan_cost(tables[i]), ScanPlan(tables[i]))

    # 枚举所有子集大小从2到n
    for size in range(2, n + 1):
        for subset in combinations(range(n), size):
            mask = sum(1 << i for i in subset)
            best = (float('inf'), None)

            # 枚举所有非空真子集作为分割
            sub = mask
            while sub > 0:
                complement = mask & ~sub
                if complement > 0 and sub in dp and complement in dp:
                    # 检查 sub 和 complement 之间是否有连接谓词
                    if has_join_predicate(sub, complement, edges):
                        cost = dp[sub][0] + dp[complement][0] \
                             + join_cost(dp[sub], dp[complement])
                        if cost < best[0]:
                            best = (cost, JoinPlan(dp[sub][1],
                                                   dp[complement][1]))
                sub = (sub - 1) & mask

            if best[1] is not None:
                dp[mask] = best

    full_mask = (1 << n) - 1
    return dp[full_mask]

def greedy_join_order(tables, edges):
    """贪心：每次选代价最低的两张表连接"""
    remaining = set(range(len(tables)))
    plans = {i: ScanPlan(tables[i]) for i in remaining}

    while len(remaining) > 1:
        best_cost = float('inf')
        best_pair = None

        for i in remaining:
            for j in remaining:
                if i < j and has_join_predicate(i, j, edges):
                    cost = join_cost(plans[i], plans[j])
                    if cost < best_cost:
                        best_cost = cost
                        best_pair = (i, j)

        if best_pair is None:
            break

        i, j = best_pair
        new_id = min(remaining)
        plans[new_id] = JoinPlan(plans[i], plans[j])
        remaining.discard(i)
        remaining.discard(j)
        remaining.add(new_id)

    return plans[min(remaining)]

-- PostgreSQL 遗传优化参数
SHOW geqo;                -- on
SHOW geqo_threshold;      -- 12
SHOW geqo_effort;         -- 5 (1-10)
SHOW geqo_pool_size;      -- 0 (自动，通常为 2*effort)
SHOW geqo_generations;    -- 0 (自动)
SHOW geqo_selection_bias; -- 2.0

-- 找出每个部门工资最高的员工
SELECT e.name, e.salary, e.dept_id
FROM   employees e
WHERE  e.salary = (
    SELECT MAX(e2.salary)
    FROM   employees e2
    WHERE  e2.dept_id = e.dept_id  -- 关联引用
);

-- 去关联后
SELECT e.name, e.salary, e.dept_id
FROM   employees e
  JOIN (
    SELECT dept_id, MAX(salary) AS max_sal
    FROM   employees
    GROUP BY dept_id
  ) m ON e.dept_id = m.dept_id AND e.salary = m.max_sal;

-- 使用窗口函数（无需子查询）
SELECT name, salary, dept_id
FROM (
    SELECT name, salary, dept_id,
           MAX(salary) OVER (PARTITION BY dept_id) AS max_sal
    FROM   employees
) t
WHERE salary = max_sal;

-- 原始
SELECT c.name
FROM   customers c
WHERE  EXISTS (
    SELECT 1 FROM orders o
    WHERE  o.cust_id = c.id AND o.amount > 1000
);

-- 去关联为 Semi Join
SELECT c.name
FROM   customers c
  SEMI JOIN orders o ON o.cust_id = c.id AND o.amount > 1000;

-- 原始
SELECT c.name
FROM   customers c
WHERE  NOT EXISTS (
    SELECT 1 FROM orders o WHERE o.cust_id = c.id
);

-- 去关联为 Anti Join
SELECT c.name
FROM   customers c
  ANTI JOIN orders o ON o.cust_id = c.id;

"""
mini_optimizer.py -- 自底向上查询优化器
支持: SeqScan, IndexScan, NLJoin, HashJoin, MergeJoin
"""
from __future__ import annotations
import itertools
from dataclasses import dataclass, field
from typing import Optional

# ============================
# 数据结构
# ============================

@dataclass
class TableStats:
    """表的统计信息"""
    name: str
    rows: int
    pages: int
    indexes: dict[str, "IndexStats"] = field(default_factory=dict)

@dataclass
class IndexStats:
    """索引的统计信息"""
    column: str
    distinct: int
    pages: int
    ordered: bool = True

@dataclass
class JoinEdge:
    """连接条件"""
    left_table: str
    left_col: str
    right_table: str
    right_col: str
    selectivity: float = 0.1

@dataclass
class PhysicalPlan:
    """物理执行计划"""
    op: str
    cost: float
    rows: float
    order: Optional[str] = None   # 输出排序键
    children: list["PhysicalPlan"] = field(default_factory=list)
    info: str = ""

    def __repr__(self):
        return self._fmt(0)

    def _fmt(self, indent: int) -> str:
        prefix = "  " * indent
        line = f"{prefix}{self.op}  (cost={self.cost:.1f}, rows={self.rows:.0f}"
        if self.order:
            line += f", order={self.order}"
        line += f")  {self.info}"
        parts = [line]
        for c in self.children:
            parts.append(c._fmt(indent + 1))
        return "\n".join(parts)

# ============================
# 代价常数
# ============================

SEQ_PAGE_COST      = 1.0
RANDOM_PAGE_COST   = 4.0
CPU_TUPLE_COST     = 0.01
CPU_INDEX_COST     = 0.005
CPU_OPERATOR_COST  = 0.0025
HASH_BUILD_COST    = 0.02
HASH_PROBE_COST    = 0.01
SORT_CPU_FACTOR    = 0.05

# ============================
# 单表访问路径
# ============================

def seq_scan(table: TableStats) -> PhysicalPlan:
    cost = SEQ_PAGE_COST * table.pages + CPU_TUPLE_COST * table.rows
    return PhysicalPlan(
        op="SeqScan", cost=cost, rows=table.rows,
        info=f"on {table.name}"
    )

def index_scan(table: TableStats, idx_name: str,
               idx: IndexStats, selectivity: float = 1.0) -> PhysicalPlan:
    est_rows = table.rows * selectivity
    pages_fetched = min(table.pages, int(est_rows * table.pages / table.rows) + 1)
    cost = (RANDOM_PAGE_COST * min(idx.pages, int(est_rows / idx.distinct * idx.pages) + 1)
            + SEQ_PAGE_COST * pages_fetched
            + CPU_INDEX_COST * est_rows
            + CPU_TUPLE_COST * est_rows)
    return PhysicalPlan(
        op="IndexScan", cost=cost, rows=est_rows,
        order=idx.column if idx.ordered else None,
        info=f"on {table.name} using {idx_name}"
    )

# ============================
# 排序
# ============================

def sort_plan(child: PhysicalPlan, sort_key: str) -> PhysicalPlan:
    if child.order == sort_key:
        return child
    n = max(child.rows, 1)
    import math
    sort_cost = SORT_CPU_FACTOR * n * math.log2(max(n, 2))
    total = child.cost + sort_cost
    return PhysicalPlan(
        op="Sort", cost=total, rows=child.rows,
        order=sort_key, children=[child],
        info=f"key={sort_key}"
    )

# ============================
# Join 算子
# ============================

def nested_loop_join(outer: PhysicalPlan, inner: PhysicalPlan,
                     edge: JoinEdge, out_rows: float) -> PhysicalPlan:
    cost = (outer.cost
            + outer.rows * inner.cost
            + CPU_TUPLE_COST * out_rows)
    return PhysicalPlan(
        op="NLJoin", cost=cost, rows=out_rows,
        children=[outer, inner],
        info=f"{edge.left_table}.{edge.left_col}={edge.right_table}.{edge.right_col}"
    )

def hash_join(build: PhysicalPlan, probe: PhysicalPlan,
              edge: JoinEdge, out_rows: float) -> PhysicalPlan:
    cost = (build.cost + probe.cost
            + HASH_BUILD_COST * build.rows
            + HASH_PROBE_COST * probe.rows
            + CPU_TUPLE_COST * out_rows)
    return PhysicalPlan(
        op="HashJoin", cost=cost, rows=out_rows,
        children=[build, probe],
        info=f"{edge.left_table}.{edge.left_col}={edge.right_table}.{edge.right_col}"
    )

def merge_join(left: PhysicalPlan, right: PhysicalPlan,
               edge: JoinEdge, out_rows: float) -> PhysicalPlan:
    left_sorted = sort_plan(left, f"{edge.left_table}.{edge.left_col}")
    right_sorted = sort_plan(right, f"{edge.right_table}.{edge.right_col}")
    merge_cost = CPU_OPERATOR_COST * (left_sorted.rows + right_sorted.rows)
    cost = left_sorted.cost + right_sorted.cost + merge_cost + CPU_TUPLE_COST * out_rows
    return PhysicalPlan(
        op="MergeJoin", cost=cost, rows=out_rows,
        order=f"{edge.left_table}.{edge.left_col}",
        children=[left_sorted, right_sorted],
        info=f"{edge.left_table}.{edge.left_col}={edge.right_table}.{edge.right_col}"
    )

# ============================
# 动态规划优化器
# ============================

class BottomUpOptimizer:
    def __init__(self, tables: list[TableStats], edges: list[JoinEdge],
                 interesting_orders: list[str] | None = None):
        self.tables = {t.name: t for t in tables}
        self.table_list = tables
        self.edges = edges
        self.interesting_orders = interesting_orders or []
        # dp[frozenset(table_names)] -> dict[order_key, PhysicalPlan]
        # order_key: None 表示不关心排序，字符串表示输出排序
        self.dp: dict[frozenset[str], dict[Optional[str], PhysicalPlan]] = {}

    def optimize(self) -> PhysicalPlan:
        self._init_single_tables()
        n = len(self.table_list)
        names = [t.name for t in self.table_list]

        for size in range(2, n + 1):
            for combo in itertools.combinations(names, size):
                table_set = frozenset(combo)
                self._optimize_join(table_set)

        full_set = frozenset(names)
        plans = self.dp.get(full_set, {})
        if not plans:
            raise ValueError("无法找到可行的连接计划")
        return min(plans.values(), key=lambda p: p.cost)

    def _init_single_tables(self):
        for t in self.table_list:
            plans: dict[Optional[str], PhysicalPlan] = {}
            ss = seq_scan(t)
            plans[None] = ss
            for idx_name, idx in t.indexes.items():
                isp = index_scan(t, idx_name, idx)
                order_key = f"{t.name}.{idx.column}"
                if order_key in self.interesting_orders:
                    if order_key not in plans or isp.cost < plans[order_key].cost:
                        plans[order_key] = isp
                if isp.cost < plans[None].cost:
                    plans[None] = isp
            self.dp[frozenset([t.name])] = plans

    def _find_edges(self, left_set: frozenset[str],
                    right_set: frozenset[str]) -> list[JoinEdge]:
        result = []
        for e in self.edges:
            if ((e.left_table in left_set and e.right_table in right_set) or
                (e.left_table in right_set and e.right_table in left_set)):
                result.append(e)
        return result

    def _optimize_join(self, table_set: frozenset[str]):
        best: dict[Optional[str], PhysicalPlan] = {}
        tables_list = sorted(table_set)

        for i in range(1, len(tables_list)):
            for left_combo in itertools.combinations(tables_list, i):
                left_set = frozenset(left_combo)
                right_set = table_set - left_set
                if not right_set:
                    continue

                if left_set not in self.dp or right_set not in self.dp:
                    continue

                join_edges = self._find_edges(left_set, right_set)
                if not join_edges:
                    continue

                edge = join_edges[0]
                sel = edge.selectivity
                for e in join_edges[1:]:
                    sel *= e.selectivity

                for l_order, l_plan in self.dp[left_set].items():
                    for r_order, r_plan in self.dp[right_set].items():
                        out_rows = l_plan.rows * r_plan.rows * sel

                        candidates = [
                            hash_join(l_plan, r_plan, edge, out_rows),
                            hash_join(r_plan, l_plan, edge, out_rows),
                            nested_loop_join(l_plan, r_plan, edge, out_rows),
                            merge_join(l_plan, r_plan, edge, out_rows),
                        ]

                        for plan in candidates:
                            order_key = plan.order
                            if order_key and order_key not in self.interesting_orders:
                                order_key = None

                            if None not in best or plan.cost < best[None].cost:
                                best[None] = plan
                            if order_key:
                                if (order_key not in best
                                        or plan.cost < best[order_key].cost):
                                    best[order_key] = plan

        if best:
            self.dp[table_set] = best

# ============================
# 示例运行
# ============================

def demo():
    orders = TableStats("orders", rows=10_000_000, pages=200_000,
        indexes={"ord_cust_idx": IndexStats("cust_id", 500_000, 30_000),
                 "ord_prod_idx": IndexStats("prod_id", 100_000, 20_000)})

    customers = TableStats("customers", rows=500_000, pages=10_000,
        indexes={"cust_pk": IndexStats("id", 500_000, 1500)})

    products = TableStats("products", rows=100_000, pages=2_000,
        indexes={"prod_pk": IndexStats("id", 100_000, 300)})

    edges = [
        JoinEdge("orders", "cust_id", "customers", "id", selectivity=1/500_000),
        JoinEdge("orders", "prod_id", "products",  "id", selectivity=1/100_000),
    ]

    optimizer = BottomUpOptimizer(
        tables=[orders, customers, products],
        edges=edges,
        interesting_orders=["customers.id", "products.id"]
    )

    best = optimizer.optimize()
    print("===== 最优执行计划 =====")
    print(best)
    print(f"\n总代价: {best.cost:.1f}")

if __name__ == "__main__":
    demo()

EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT c.name, COUNT(o.id), SUM(o.amount)
FROM   customers c
  JOIN orders o ON o.cust_id = c.id
WHERE  c.country = 'CN'
GROUP BY c.name;

ANALYZE customers;
-- 对倾斜列提高精度
ALTER TABLE customers ALTER COLUMN country SET STATISTICS 1000;
ANALYZE customers;

SET enable_hashjoin = off;              -- 禁用特定算法测试
SET random_page_cost = 1.1;            -- SSD 场景
SET effective_cache_size = '32GB';
SET join_collapse_limit = 1;           -- 保持 FROM 子句中的连接顺序

SHOW CLUSTER SETTING sql.stats.automatic_collection.enabled;
SHOW STATISTICS FOR TABLE orders;
CREATE STATISTICS orders_stats FROM orders;  -- 手动触发

# Bao 的简化伪代码
class BaoOptimizer:
    def __init__(self, arms):
        self.arms = arms
        self.model = TreeCNN()
        self.experience = []

    def choose_arm(self, query):
        plans = [(arm, self.model(traditional_optimizer(query, hints=arm)))
                 for arm in self.arms]
        return min(plans, key=lambda x: x[1])[0]  # Thompson Sampling

    def feedback(self, query, arm, actual_time):
        plan = traditional_optimizer(query, hints=arm)
        self.experience.append((plan, actual_time))
        if len(self.experience) % RETRAIN_INTERVAL == 0:
            self.model.train(self.experience)

-- PostgreSQL 关键配置
-- 1. 让优化器了解你的硬件
SET random_page_cost = 1.1;           -- SSD
SET effective_cache_size = '24GB';    -- 可用内存的 75%
SET work_mem = '256MB';               -- 排序/哈希可用内存

-- 2. 统计信息精度
ALTER TABLE large_table ALTER COLUMN skewed_col SET STATISTICS 5000;
-- default_statistics_target = 100，对倾斜列可提高到 1000-10000

-- 3. 连接优化
SET join_collapse_limit = 12;   -- 超过 12 张表时不重排 explicit JOIN
SET from_collapse_limit = 12;   -- 超过 12 张表时不展平子查询
SET geqo_threshold = 14;        -- 超过 14 张表时启用遗传算法