diff --git a/AGENTS.md b/AGENTS.md
index b81c686..9627082 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -28,5 +28,5 @@
 - 改动说明只在交流中记录。
 
 ## 当前重点
-- 主线已完成：高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测 → M8 事务保存点 → M9 角色继承 → M10 转授权 → M11 FROM 派生表与投影别名 → M12 `WITH` / CTE → M13 比较运算符补全 → M14 ANY / ALL 量词比较。
+- 主线已完成：高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测 → M8 事务保存点 → M9 角色继承 → M10 转授权 → M11 FROM 派生表与投影别名 → M12 `WITH` / CTE → M13 比较运算符补全 → M14 ANY / ALL 量词比较 → M15 优化器直方图。
 - 进度与计数见 `agent_credientials/project-plan.txt`，限制与风险见 `agent_credientials/known-issues.txt`。
diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Cost.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Cost.hs
index bdcdd7e..b38e3e7 100644
--- a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Cost.hs
+++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Cost.hs
@@ -1,8 +1,8 @@
-module ChuSQL.Core.Engine.Algebra.Cost (tableMetaOf, probeCost, pointRows, rangeRows, indexCheaper) where
+module ChuSQL.Core.Engine.Algebra.Cost (tableMetaOf, probeCost, pointRows, rangeRows, selectivityOf, indexCheaper) where
 
 import ChuSQL.Core.Model
 
--- 统计与成本估算：行数、列级不同值数与索引探路页数。
+-- 统计与成本估算：行数、不同值数、列直方图与索引探路页数。
 
 -- | 表统计，拿不到就是 Nothing
 tableMetaOf :: Database -> String -> Maybe TableMeta
@@ -20,9 +20,42 @@ pointRows m col = case [d | (c, d, _) <- metaDistinct m, c == col, d > 1] of
   where
     rows = max 0 (metaRowCount m)
 
--- | 范围扫描的估算命中行数：没有直方图，按四分之一算
-rangeRows :: TableMeta -> Int
-rangeRows m = max 0 (metaRowCount m) `div` 4
+-- | 范围扫描的估算命中行数：有直方图按桶算，没有按四分之一
+rangeRows :: TableMeta -> String -> Maybe (Value, Bool) -> Maybe (Value, Bool) -> Int
+rangeRows m col lo hi = case histogramFor m col of
+    Just h -> estimateRows h lo hi
+    Nothing -> max 0 (metaRowCount m) `div` 4
+
+-- | 直方图估算区间命中行数：按桶算交集占比
+estimateRows :: Histogram -> Maybe (Value, Bool) -> Maybe (Value, Bool) -> Int
+estimateRows h lo hi = max 0 (min total (round (sum (zipWith overlap [0 ..] (histBuckets h)))))
+  where
+    total = sum (histBuckets h)
+    width = if histHigh h > histLow h then (histHigh h - histLow h) / fromIntegral (length (histBuckets h)) else 0
+    lower = maybe (-1 / 0) id (boundOf lo)
+    upper = maybe (1 / 0) id (boundOf hi)
+    overlap i count
+        | width <= 0 = if lower <= histLow h && histLow h <= upper then fromIntegral count else 0
+        | otherwise = fromIntegral count * hit / width
+      where
+        start = histLow h + width * fromIntegral i
+        hit = max 0 (min (start + width) upper - max start lower)
+
+-- | 区间选择率：直方图命中的比例；没有直方图给 Nothing
+selectivityOf :: TableMeta -> String -> Maybe (Value, Bool) -> Maybe (Value, Bool) -> Maybe Double
+selectivityOf m col lo hi = case histogramFor m col of
+    Nothing -> Nothing
+    Just h
+        | total <= 0 -> Nothing
+        | otherwise -> Just (fromIntegral (estimateRows h lo hi) / fromIntegral total)
+      where
+        total = sum (histBuckets h)
+
+-- | 边界值转成参与估值的小数；非数值给 Nothing
+boundOf :: Maybe (Value, Bool) -> Maybe Double
+boundOf (Just (VInt n, _)) = Just (fromIntegral n)
+boundOf (Just (VFloat d, _)) = Just d
+boundOf _ = Nothing
 
 -- | 有索引且探路加命中比全表扫便宜
 indexCheaper :: TableMeta -> String -> Int -> Bool
diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/JoinOrder.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/JoinOrder.hs
index 3c060d3..5eab017 100644
--- a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/JoinOrder.hs
+++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/JoinOrder.hs
@@ -1,6 +1,6 @@
 module ChuSQL.Core.Engine.Algebra.JoinOrder (reorderJoins) where
 
-import ChuSQL.Core.Engine.Algebra.Cost (pointRows, rangeRows, tableMetaOf)
+import ChuSQL.Core.Engine.Algebra.Cost (pointRows, rangeRows, selectivityOf, tableMetaOf)
 import ChuSQL.Core.Engine.Algebra.Expr (colsInExpr, hasDivision)
 import ChuSQL.Core.Engine.Algebra.Op (RelOp (..), relOpCols)
 import ChuSQL.Core.Engine.Syntax.AST (Expr (..), JoinKind (..))
@@ -45,7 +45,7 @@ leafRows :: Database -> RelOp -> Maybe Int
 leafRows db op = case op of
     Scan _ t _ -> rowCountOf db t
     Lookup _ t c _ -> pointRows <$> tableMetaOf db t <*> pure c
-    Range _ t _ _ _ -> rangeRows <$> tableMetaOf db t
+    Range _ t c lo hi -> (\m -> rangeRows m c lo hi) <$> tableMetaOf db t
     Filter p x -> scaleRows db p x <$> leafRows db x
     Project _ x -> leafRows db x
     _ -> Nothing
@@ -54,12 +54,20 @@ leafRows db op = case op of
 rowCountOf :: Database -> String -> Maybe Int
 rowCountOf db t = metaRowCount <$> tableMetaOf db t
 
--- | 过滤之后的大致行数：等值看不同值数，其他按四分之一
+-- | 行数估算：等值看不同值数，区间看直方图，其余四分之一
 scaleRows :: Database -> Expr -> RelOp -> Int -> Int
 scaleRows db p x rows = case p of
     Eq (Col k) _ -> case distinctOf db x k of
         Just d | d > 1 -> max 1 (rows `div` d)
         _ -> quarter
+    Gt (Col k) (LitInt v) -> pick k (Just (VInt v, False)) Nothing
+    Gt (Col k) (LitStr v) -> pick k (Just (VStr v, False)) Nothing
+    GtE (Col k) (LitInt v) -> pick k (Just (VInt v, True)) Nothing
+    GtE (Col k) (LitStr v) -> pick k (Just (VStr v, True)) Nothing
+    Lt (Col k) (LitInt v) -> pick k Nothing (Just (VInt v, False))
+    Lt (Col k) (LitStr v) -> pick k Nothing (Just (VStr v, False))
+    LtE (Col k) (LitInt v) -> pick k Nothing (Just (VInt v, True))
+    LtE (Col k) (LitStr v) -> pick k Nothing (Just (VStr v, True))
     Gt _ _ -> quarter
     Lt _ _ -> quarter
     GtE _ _ -> quarter
@@ -70,6 +78,11 @@ scaleRows db p x rows = case p of
     _ -> quarter
   where
     quarter = max 1 (rows `div` 4)
+    pick k lo hi = case leafSource x of
+        Just (alias, t)
+            | Just m <- tableMetaOf db t
+            , Just s <- selectivityOf m (unqualify alias k) lo hi -> max 1 (round (fromIntegral rows * s))
+        _ -> quarter
 
 -- | 叶子对应的别名与表名
 leafSource :: RelOp -> Maybe (Maybe String, String)
diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Optimize.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Optimize.hs
index f66f19e..9c21a57 100644
--- a/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Optimize.hs
+++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Algebra/Optimize.hs
@@ -159,10 +159,10 @@ worthLookup db t col = case tableMetaOf db t of
     Just m -> indexCheaper m col (pointRows m col)
 
 -- | 范围扫描值不值得改写；没有统计就按老规矩改
-worthRange :: Database -> String -> String -> Bool
-worthRange db t col = case tableMetaOf db t of
+worthRange :: Database -> String -> String -> Maybe (Value, Bool) -> Maybe (Value, Bool) -> Bool
+worthRange db t col lo hi = case tableMetaOf db t of
     Nothing -> True
-    Just m -> indexCheaper m col (rangeRows m)
+    Just m -> indexCheaper m col (rangeRows m col lo hi)
 
 -- | 单节点重写
 rewriteNode :: Database -> RelOp -> RelOp
@@ -171,45 +171,53 @@ rewriteNode db (Filter (Eq (Col k) (LitInt v)) (Scan mAlias t Nothing))
 rewriteNode db (Filter (Eq (Col k) (LitStr v)) (Scan mAlias t Nothing))
     | worthLookup db t (unqualify mAlias k) = Lookup mAlias t (unqualify mAlias k) (VStr v)
 rewriteNode db (Filter (And (Gt (Col k1) (LitInt lo)) (Lt (Col k2) (LitInt hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VInt lo, False)) (Just (VInt hi, False)) =
         Range mAlias t (unqualify mAlias k1) (Just (VInt lo, False)) (Just (VInt hi, False))
 rewriteNode db (Filter (And (Gt (Col k1) (LitStr lo)) (Lt (Col k2) (LitStr hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VStr lo, False)) (Just (VStr hi, False)) =
         Range mAlias t (unqualify mAlias k1) (Just (VStr lo, False)) (Just (VStr hi, False))
 rewriteNode db (Filter (Gt (Col k) (LitInt v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) (Just (VInt v, False)) Nothing
+    | worthRange db t (unqualify mAlias k) (Just (VInt v, False)) Nothing =
+        Range mAlias t (unqualify mAlias k) (Just (VInt v, False)) Nothing
 rewriteNode db (Filter (Gt (Col k) (LitStr v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) (Just (VStr v, False)) Nothing
+    | worthRange db t (unqualify mAlias k) (Just (VStr v, False)) Nothing =
+        Range mAlias t (unqualify mAlias k) (Just (VStr v, False)) Nothing
 rewriteNode db (Filter (Lt (Col k) (LitInt v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) Nothing (Just (VInt v, False))
+    | worthRange db t (unqualify mAlias k) Nothing (Just (VInt v, False)) =
+        Range mAlias t (unqualify mAlias k) Nothing (Just (VInt v, False))
 rewriteNode db (Filter (Lt (Col k) (LitStr v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) Nothing (Just (VStr v, False))
+    | worthRange db t (unqualify mAlias k) Nothing (Just (VStr v, False)) =
+        Range mAlias t (unqualify mAlias k) Nothing (Just (VStr v, False))
 rewriteNode db (Filter (And (GtE (Col k1) (LitInt lo)) (LtE (Col k2) (LitInt hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VInt lo, True)) (Just (VInt hi, True)) =
         Range mAlias t (unqualify mAlias k1) (Just (VInt lo, True)) (Just (VInt hi, True))
 rewriteNode db (Filter (And (GtE (Col k1) (LitInt lo)) (Lt (Col k2) (LitInt hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VInt lo, True)) (Just (VInt hi, False)) =
         Range mAlias t (unqualify mAlias k1) (Just (VInt lo, True)) (Just (VInt hi, False))
 rewriteNode db (Filter (And (Gt (Col k1) (LitInt lo)) (LtE (Col k2) (LitInt hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VInt lo, False)) (Just (VInt hi, True)) =
         Range mAlias t (unqualify mAlias k1) (Just (VInt lo, False)) (Just (VInt hi, True))
 rewriteNode db (Filter (And (GtE (Col k1) (LitStr lo)) (LtE (Col k2) (LitStr hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VStr lo, True)) (Just (VStr hi, True)) =
         Range mAlias t (unqualify mAlias k1) (Just (VStr lo, True)) (Just (VStr hi, True))
 rewriteNode db (Filter (And (GtE (Col k1) (LitStr lo)) (Lt (Col k2) (LitStr hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VStr lo, True)) (Just (VStr hi, False)) =
         Range mAlias t (unqualify mAlias k1) (Just (VStr lo, True)) (Just (VStr hi, False))
 rewriteNode db (Filter (And (Gt (Col k1) (LitStr lo)) (LtE (Col k2) (LitStr hi))) (Scan mAlias t Nothing))
-    | k1 == k2 && worthRange db t (unqualify mAlias k1) =
+    | k1 == k2 && worthRange db t (unqualify mAlias k1) (Just (VStr lo, False)) (Just (VStr hi, True)) =
         Range mAlias t (unqualify mAlias k1) (Just (VStr lo, False)) (Just (VStr hi, True))
 rewriteNode db (Filter (GtE (Col k) (LitInt v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) (Just (VInt v, True)) Nothing
+    | worthRange db t (unqualify mAlias k) (Just (VInt v, True)) Nothing =
+        Range mAlias t (unqualify mAlias k) (Just (VInt v, True)) Nothing
 rewriteNode db (Filter (GtE (Col k) (LitStr v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) (Just (VStr v, True)) Nothing
+    | worthRange db t (unqualify mAlias k) (Just (VStr v, True)) Nothing =
+        Range mAlias t (unqualify mAlias k) (Just (VStr v, True)) Nothing
 rewriteNode db (Filter (LtE (Col k) (LitInt v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) Nothing (Just (VInt v, True))
+    | worthRange db t (unqualify mAlias k) Nothing (Just (VInt v, True)) =
+        Range mAlias t (unqualify mAlias k) Nothing (Just (VInt v, True))
 rewriteNode db (Filter (LtE (Col k) (LitStr v)) (Scan mAlias t Nothing))
-    | worthRange db t (unqualify mAlias k) = Range mAlias t (unqualify mAlias k) Nothing (Just (VStr v, True))
+    | worthRange db t (unqualify mAlias k) Nothing (Just (VStr v, True)) =
+        Range mAlias t (unqualify mAlias k) Nothing (Just (VStr v, True))
 rewriteNode db (Filter p (Join kind l r c))
     | not (hasDivision p || planHasDivision (Join kind l r c)) =
         case kind of
diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs
index ef73166..b7c46b4 100644
--- a/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs
+++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Storage/IPC.hs
@@ -401,7 +401,7 @@ instance MonadStorage IPCStorage where
 
 -- | 线上表信息转引擎侧统计
 metaOf :: TableInfo -> TableMeta
-metaOf info = TableMeta (tiRows info) (tiStats info) (tiIndexes info)
+metaOf info = TableMeta (tiRows info) (tiStats info) (tiHistograms info) (tiIndexes info)
 
 -- | 没设库名就照单全收；设了库名只收这个库的表，键名去掉库名前缀
 visibleTables :: Maybe String -> TableInfo -> [(String, Table)]
diff --git a/chusql-core/engine/test/Spec.hs b/chusql-core/engine/test/Spec.hs
index ef1cf9a..9dbb328 100644
--- a/chusql-core/engine/test/Spec.hs
+++ b/chusql-core/engine/test/Spec.hs
@@ -2,6 +2,7 @@
 
 module Main where
 
+import ChuSQL.Core.Engine.Algebra.Cost (rangeRows, selectivityOf)
 import ChuSQL.Core.Engine.Algebra.Eval (evalRelOp, evalRelOpM)
 import ChuSQL.Core.Engine.Algebra.Expr (colsInExpr, evalExpr)
 import ChuSQL.Core.Engine.Algebra.Op (RelOp (..), renderPlan)
@@ -71,7 +72,7 @@ chainA =
         { tableName = "a"
         , tableCols = [("id", TInt), ("name", TStr)]
         , tableRows = [[("id", VInt n), ("name", VStr ("a" ++ show n))] | n <- [1 .. 6]]
-        , tableMeta = Just (TableMeta 6 [("id", 6, False)] ["id"])
+        , tableMeta = Just (TableMeta 6 [("id", 6, False)] [] ["id"])
         }
 
 -- | 连接链测试表 b：4 行，a_id 各不相同
@@ -81,7 +82,7 @@ chainB =
         { tableName = "b"
         , tableCols = [("id", TInt), ("a_id", TInt)]
         , tableRows = [[("id", VInt n), ("a_id", VInt n)] | n <- [1 .. 4]]
-        , tableMeta = Just (TableMeta 4 [("id", 4, False), ("a_id", 4, False)] ["id"])
+        , tableMeta = Just (TableMeta 4 [("id", 4, False), ("a_id", 4, False)] [] ["id"])
         }
 
 -- | 连接链测试表 c：2 行，b_id 各不相同
@@ -91,7 +92,7 @@ chainC =
         { tableName = "c"
         , tableCols = [("id", TInt), ("b_id", TInt)]
         , tableRows = [[("id", VInt n), ("b_id", VInt n)] | n <- [1, 2]]
-        , tableMeta = Just (TableMeta 2 [("id", 2, False), ("b_id", 2, False)] ["id"])
+        , tableMeta = Just (TableMeta 2 [("id", 2, False), ("b_id", 2, False)] [] ["id"])
         }
 
 -- | 三张表组成的连接链测试库
@@ -551,7 +552,7 @@ main = hspec $ do
             executeOperator OpNe [VInt 1, VFloat 1.0] `shouldBe` Right (VBool False)
 
         it "rewrites >= and <= into an inclusive index range" $ do
-            let stat rows distinct indexes = TableMeta rows [(c, d, False) | (c, d) <- distinct] indexes
+            let stat rows distinct indexes = TableMeta rows [(c, d, False) | (c, d) <- distinct] [] indexes
                 dbWith m = [("users", Table "users" [("id", TInt), ("name", TStr)] [] (Just m))]
                 indexed = dbWith (stat 10000 [("id", 10000)] ["id"])
                 low = Filter (GtE (Col "id") (LitInt 2)) (Scan Nothing "users" Nothing)
@@ -1509,6 +1510,44 @@ main = hspec $ do
                     [ [("name", VStr "Carol")]
                     ]
 
+    describe "ChuSQL.Core.Engine.Algebra.Cost (直方图)" $ do
+        it "builds a 16-bucket histogram from the numeric values" $ do
+            let hist = histogramOf [VInt n | n <- [0 .. 99]]
+            fmap histLow hist `shouldBe` Just 0
+            fmap histHigh hist `shouldBe` Just 99
+            fmap (length . histBuckets) hist `shouldBe` Just 16
+            fmap (sum . histBuckets) hist `shouldBe` Just 100
+            -- 非数值与 NULL 不参与；只有一个值都落第 0 桶
+            histogramOf [VStr "a", VNull] `shouldBe` Nothing
+            histogramOf [VInt 5, VNull, VStr "a"] `shouldBe` Just (Histogram 5 5 (1 : replicate 15 0))
+
+        it "estimates a range from the histogram buckets" $ do
+            let hist = Histogram 0 1600 (replicate 16 100)
+                meta = TableMeta 1600 [] [("id", hist)] ["id"]
+                plain = TableMeta 1600 [] [] []
+            rangeRows meta "id" (Just (VInt 0, True)) (Just (VInt 100, True)) `shouldBe` 100
+            rangeRows meta "id" Nothing Nothing `shouldBe` 1600
+            selectivityOf meta "id" (Just (VInt 0, True)) (Just (VInt 100, True)) `shouldBe` Just 0.0625
+            -- 没有直方图退回四分之一
+            rangeRows plain "id" Nothing Nothing `shouldBe` 400
+            selectivityOf plain "id" Nothing Nothing `shouldBe` Nothing
+
+        it "uses the histogram to reject a range that would scan most rows" $ do
+            let buckets = replicate 8 63 ++ replicate 8 62
+                withHist = TableMeta 1000 [("id", 1000, False)] [("id", Histogram 0 1600 buckets)] ["id"]
+                plain = TableMeta 1000 [("id", 1000, False)] [] ["id"]
+                dbWith m = [("users", Table "users" [("id", TInt), ("name", TStr)] [] (Just m))]
+                narrow = Filter (And (GtE (Col "id") (LitInt 10)) (LtE (Col "id") (LitInt 20))) (Scan Nothing "users" Nothing)
+                wide = Filter (Gt (Col "id") (LitInt 0)) (Scan Nothing "users" Nothing)
+            -- 直方图说这个窄范围只命中十几行：改写
+            optimize (dbWith withHist) narrow
+                `shouldBe` Range Nothing "users" "id" (Just (VInt 10, True)) (Just (VInt 20, True))
+            -- 直方图说这条范围覆盖全表：索引不比全表扫便宜，不改写
+            optimize (dbWith withHist) wide `shouldBe` wide
+            -- 没有直方图按四分之一算，仍然改写
+            optimize (dbWith plain) wide
+                `shouldBe` Range Nothing "users" "id" (Just (VInt 0, False)) Nothing
+
     describe "ChuSQL.Core.Engine.Algebra.Optimize" $ do
         it "keeps single-table WHERE results identical" $ do
             sameResultAsUnoptimized "SELECT name FROM users WHERE age > 18"
@@ -1559,7 +1598,7 @@ main = hspec $ do
                 `shouldBe` Right (Just (LitBool False))
 
         it "picks the index only when the statistics make it cheaper" $ do
-            let stat rows distinct indexes = TableMeta rows [(c, d, False) | (c, d) <- distinct] indexes
+            let stat rows distinct indexes = TableMeta rows [(c, d, False) | (c, d) <- distinct] [] indexes
                 dbWith m = [("users", Table "users" [("id", TInt), ("name", TStr)] [] (Just m))]
                 unknown = [("users", Table "users" [("id", TInt), ("name", TStr)] [] Nothing)]
                 point = Filter (Eq (Col "id") (LitInt 1)) (Scan Nothing "users" Nothing)
@@ -2689,7 +2728,7 @@ rowShards width rows
 
 -- | 声明 1000 行的大表
 bigTable :: Table
-bigTable = users {tableName = "big", tableMeta = Just (TableMeta 1000 [("id", 1000, False)] ["id"])}
+bigTable = users {tableName = "big", tableMeta = Just (TableMeta 1000 [("id", 1000, False)] [] ["id"])}
 
 -- | 只装大表的库
 bigDB :: Database
diff --git a/chusql-core/model/src/ChuSQL/Core/Model.hs b/chusql-core/model/src/ChuSQL/Core/Model.hs
index 2849bec..d6d2f0b 100644
--- a/chusql-core/model/src/ChuSQL/Core/Model.hs
+++ b/chusql-core/model/src/ChuSQL/Core/Model.hs
@@ -10,8 +10,11 @@ module ChuSQL.Core.Model
     , Value (..)
     , Row
     , TableMeta (..)
+    , Histogram (..)
     , Table (..)
     , Database
+    , histogramOf
+    , histogramFor
     , colNames
     , colType
     , lookupTable
@@ -42,7 +45,7 @@ module ChuSQL.Core.Model
 import Data.Char (isDigit, isSpace, toLower)
 import Data.Hashable (Hashable (..))
 import Data.List (isSuffixOf, stripPrefix)
-import Data.Maybe (fromMaybe)
+import Data.Maybe (fromMaybe, mapMaybe)
 
 -- 数据模型：列类型、列约束、值、行、表、数据库，外加查表小工具。
 
@@ -441,10 +444,19 @@ instance Hashable Value where
 -- | 一行数据
 type Row = [(String, Value)]
 
--- | 一张表的统计：行数、各列不同值数（capped 表示只数到上限）、有索引的列
+-- | 一列的等宽直方图：区间下界、上界与每桶计数
+data Histogram = Histogram
+    { histLow :: Double
+    , histHigh :: Double
+    , histBuckets :: [Int]
+    }
+    deriving (Show, Eq)
+
+-- | 一张表的统计：行数、各列不同值数、列直方图与索引
 data TableMeta = TableMeta
     { metaRowCount :: Int
     , metaDistinct :: [(String, Int, Bool)]
+    , metaHistograms :: [(String, Histogram)]
     , metaIndexes :: [String]
     }
     deriving (Show)
@@ -461,6 +473,36 @@ data Table = Table
 -- | 库：库名到表的映射
 type Database = [(String, Table)]
 
+-- | 这一列的直方图
+histogramFor :: TableMeta -> String -> Maybe Histogram
+histogramFor m col = lookup col (metaHistograms m)
+
+-- | 数值列按 16 个等宽桶统计，无数值给 Nothing
+histogramOf :: [Value] -> Maybe Histogram
+histogramOf values = case mapMaybe numericOf values of
+    [] -> Nothing
+    vs -> let lo = minimum vs; hi = maximum vs in Just (Histogram lo hi (bucketCounts lo hi vs))
+
+-- | 值转成用于分桶的数；非数值给 Nothing
+numericOf :: Value -> Maybe Double
+numericOf (VInt n) = Just (fromIntegral n)
+numericOf (VFloat d) = Just d
+numericOf _ = Nothing
+
+-- | 每个桶里落了多少个值
+bucketCounts :: Double -> Double -> [Double] -> [Int]
+bucketCounts lo hi vs = [length [v | v <- vs, bucketIndexOf lo hi v == i] | i <- [0 .. bucketTotal - 1]]
+
+-- | 值落在第几个桶；区间退化时都算第 0 桶
+bucketIndexOf :: Double -> Double -> Double -> Int
+bucketIndexOf lo hi v
+    | hi <= lo = 0
+    | otherwise = max 0 (min (bucketTotal - 1) (floor ((v - lo) / (hi - lo) * fromIntegral bucketTotal) :: Int))
+
+-- | 直方图的桶数
+bucketTotal :: Int
+bucketTotal = 16
+
 -- | 取表的列名
 colNames :: Table -> [String]
 colNames = map fst . tableCols
diff --git a/chusql-core/model/src/ChuSQL/Core/Protocol.hs b/chusql-core/model/src/ChuSQL/Core/Protocol.hs
index 027ca57..af564a4 100644
--- a/chusql-core/model/src/ChuSQL/Core/Protocol.hs
+++ b/chusql-core/model/src/ChuSQL/Core/Protocol.hs
@@ -15,7 +15,7 @@ module ChuSQL.Core.Protocol (
     queryResultJson,
 ) where
 
-import ChuSQL.Core.Model (Row, Value (..))
+import ChuSQL.Core.Model (Histogram (..), Row, Value (..))
 import Data.Aeson (FromJSON (..), ToJSON (..), object, withObject, (.:), (.:?), (.!=), (.=))
 import qualified Data.Aeson as A
 import qualified Data.Aeson.Key as K
@@ -244,6 +244,7 @@ data TableInfo = TableInfo
     , tiRows :: Int
     , tiIndexes :: [String]
     , tiStats :: [(String, Int, Bool)]
+    , tiHistograms :: [(String, Histogram)]
     }
     deriving (Show, Eq)
 
@@ -255,11 +256,21 @@ instance ToJSON TableInfo where
             , "columns" .= tiColumns info
             , "row_count" .= tiRows info
             , "indexes" .= map (\name -> object ["column" .= name]) (tiIndexes info)
-            , "stats" .= map statJson (tiStats info)
+            , "stats" .= map (statJson (tiHistograms info)) (tiStats info)
             ]
       where
-        statJson (name, distinct, capped) =
-            object ["name" .= name, "distinct" .= distinct, "capped" .= capped]
+        statJson hists (name, distinct, capped) =
+            object (statFields ++ histFields (lookup name hists))
+          where
+            statFields = ["name" .= name, "distinct" .= distinct, "capped" .= capped]
+            histFields Nothing = []
+            histFields (Just h) = ["lo" .= histLow h, "hi" .= histHigh h, "hist" .= histBuckets h]
+
+-- | 线上直方图还原；区间或桶缺了就当作没有
+histogramFromWire :: Maybe Double -> Maybe Double -> [Int] -> Maybe Histogram
+histogramFromWire (Just low) (Just high) counts
+    | not (null counts) = Just (Histogram low high counts)
+histogramFromWire _ _ _ = Nothing
 
 -- | JSON 解回表信息
 instance FromJSON TableInfo where
@@ -270,14 +281,19 @@ instance FromJSON TableInfo where
         indexes <- o .:? "indexes" .!= []
         stats <- o .:? "stats" .!= []
         indexColumns <- mapM (\v -> withObject "IndexWire" (.: "column") v) (indexes :: [A.Value])
-        parsedStats <- mapM parseStat (stats :: [A.Value])
-        pure (TableInfo table columns count indexColumns parsedStats)
+        parsed <- mapM parseStat (stats :: [A.Value])
+        let parsedStats = [(name, distinct, capped) | (name, distinct, capped, _) <- parsed]
+            parsedHistograms = [(name, h) | (name, _, _, Just h) <- parsed]
+        pure (TableInfo table columns count indexColumns parsedStats parsedHistograms)
       where
         parseStat = withObject "ColumnStat" $ \o -> do
             name <- o .: "name"
             distinct <- o .: "distinct"
             capped <- o .:? "capped" .!= False
-            pure (name, distinct, capped)
+            low <- o .:? "lo"
+            high <- o .:? "hi"
+            counts <- o .:? "hist" .!= []
+            pure (name, distinct, capped, histogramFromWire low high counts)
 
 -- | 一条账号记录
 data Account = Account
diff --git a/chusql-core/storage/src/catalog.rs b/chusql-core/storage/src/catalog.rs
index 6c9cacb..9869a96 100644
--- a/chusql-core/storage/src/catalog.rs
+++ b/chusql-core/storage/src/catalog.rs
@@ -12,6 +12,7 @@ use crate::protocol::{Row, SchemaColumn};
 // 数据字典：记每张表有哪些列、什么类型、多少行、有哪些索引，以及列级统计。
 
 const STATS_DISTINCT_CAP: u64 = 4096;
+const HIST_BUCKETS: usize = 16;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct IndexSchema {
@@ -23,6 +24,12 @@ pub struct ColumnStat {
     pub distinct: u64,
     #[serde(default)]
     pub capped: bool,
+    #[serde(default)]
+    pub lo: Option<f64>,
+    #[serde(default)]
+    pub hi: Option<f64>,
+    #[serde(default)]
+    pub hist: Vec<u64>,
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -152,6 +159,14 @@ impl Catalog {
                 if seen.remove(&fold(name, value_hash(v))) {
                     stat.distinct = stat.distinct.saturating_sub(1);
                 }
+                hist_remove(stat, v);
+            }
+        }
+        if entry.row_count == 0 {
+            for stat in entry.stats.values_mut() {
+                stat.lo = None;
+                stat.hi = None;
+                stat.hist.clear();
             }
         }
     }
@@ -187,6 +202,7 @@ impl Catalog {
                     seen.clear();
                 }
             }
+            hist_add(stat, v);
         }
     }
 
@@ -397,3 +413,63 @@ fn fold(column: &str, value_hash: u64) -> u64 {
     value_hash.hash(&mut h);
     h.finish()
 }
+
+/// 值落在第几个桶；区间退化时都算第 0 桶
+fn hist_index(lo: f64, hi: f64, v: f64) -> usize {
+    if hi <= lo {
+        return 0;
+    }
+    let scaled = ((v - lo) / (hi - lo) * HIST_BUCKETS as f64).floor();
+    scaled.clamp(0.0, (HIST_BUCKETS - 1) as f64) as usize
+}
+
+/// 区间拓宽时把旧桶按中点并进来，只保总量
+fn hist_merge(old: &[u64], old_lo: f64, old_hi: f64, new_lo: f64, new_hi: f64) -> Vec<u64> {
+    let mut out = vec![0u64; HIST_BUCKETS];
+    for (i, count) in old.iter().enumerate() {
+        if *count == 0 {
+            continue;
+        }
+        let mid = if old_hi > old_lo {
+            old_lo + (old_hi - old_lo) * (i as f64 + 0.5) / old.len() as f64
+        } else {
+            old_lo
+        };
+        out[hist_index(new_lo, new_hi, mid)] += *count;
+    }
+    out
+}
+
+/// 把一个值并进直方图；值跑出区间就拓宽区间
+fn hist_add(stat: &mut ColumnStat, v: &serde_json::Value) {
+    let Some(x) = v.as_f64() else { return };
+    match (stat.lo, stat.hi) {
+        (Some(lo), Some(hi)) if stat.hist.len() == HIST_BUCKETS => {
+            if x < lo || x > hi {
+                let (nlo, nhi) = (lo.min(x), hi.max(x));
+                stat.hist = hist_merge(&stat.hist, lo, hi, nlo, nhi);
+                stat.lo = Some(nlo);
+                stat.hi = Some(nhi);
+            }
+            let index = hist_index(stat.lo.unwrap_or(x), stat.hi.unwrap_or(x), x);
+            stat.hist[index] += 1;
+        }
+        _ => {
+            stat.lo = Some(x);
+            stat.hi = Some(x);
+            stat.hist = vec![0u64; HIST_BUCKETS];
+            stat.hist[0] = 1;
+        }
+    }
+}
+
+/// 从直方图里减掉一个值；值在区间外就不动
+fn hist_remove(stat: &mut ColumnStat, v: &serde_json::Value) {
+    let Some(x) = v.as_f64() else { return };
+    let (Some(lo), Some(hi)) = (stat.lo, stat.hi) else { return };
+    if x < lo || x > hi || stat.hist.len() != HIST_BUCKETS {
+        return;
+    }
+    let index = hist_index(lo, hi, x);
+    stat.hist[index] = stat.hist[index].saturating_sub(1);
+}
diff --git a/chusql-core/storage/src/protocol.rs b/chusql-core/storage/src/protocol.rs
index 4529d21..a6f32fa 100644
--- a/chusql-core/storage/src/protocol.rs
+++ b/chusql-core/storage/src/protocol.rs
@@ -58,12 +58,18 @@ fn default_true() -> bool {
     true
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct ColumnStatWire {
     pub name: String,
     pub distinct: u64,
     #[serde(default)]
     pub capped: bool,
+    #[serde(default)]
+    pub lo: Option<f64>,
+    #[serde(default)]
+    pub hi: Option<f64>,
+    #[serde(default)]
+    pub hist: Vec<u64>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
diff --git a/chusql-core/storage/src/runtime.rs b/chusql-core/storage/src/runtime.rs
index 96db318..01501b7 100644
--- a/chusql-core/storage/src/runtime.rs
+++ b/chusql-core/storage/src/runtime.rs
@@ -1602,6 +1602,9 @@ fn stat_wires(schema: &crate::catalog::TableSchema) -> Vec<ColumnStatWire> {
             name: name.clone(),
             distinct: s.distinct,
             capped: s.capped,
+            lo: s.lo,
+            hi: s.hi,
+            hist: s.hist.clone(),
         })
         .collect()
 }
diff --git a/chusql-core/storage/tests/catalog_test.rs b/chusql-core/storage/tests/catalog_test.rs
index 1a60f54..a7cdd7b 100644
--- a/chusql-core/storage/tests/catalog_test.rs
+++ b/chusql-core/storage/tests/catalog_test.rs
@@ -83,6 +83,47 @@ fn remove_column_drops_definition_stats_and_index() {
     assert!(c.remove_column("nope", "code").is_err());
 }
 
+/// 直方图随插入累计、随删除递减，值跑出区间就拓宽
+#[test]
+fn stats_track_a_histogram_for_numeric_columns() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("catalog.json");
+
+    let mut c = Catalog::load(&path).unwrap();
+    let rows: Vec<Row> = (0..100)
+        .map(|n| row(&[("id", json!(n)), ("name", json!(format!("u{}", n)))]))
+        .collect();
+    c.rebuild_stats("t", &rows);
+
+    let s = c.describe("t").unwrap();
+    let id = s.stats.get("id").unwrap();
+    assert_eq!(id.distinct, 100);
+    assert_eq!(id.lo, Some(0.0));
+    assert_eq!(id.hi, Some(99.0));
+    assert_eq!(id.hist.len(), 16);
+    assert_eq!(id.hist.iter().sum::<u64>(), 100);
+    let name = s.stats.get("name").unwrap();
+    assert!(name.hist.is_empty(), "字符串列不进直方图");
+    assert_eq!(name.lo, None);
+
+    c.record_insert("t", &row(&[("id", json!(1000)), ("name", json!("x"))]));
+    let s = c.describe("t").unwrap();
+    let id = s.stats.get("id").unwrap();
+    assert_eq!(id.lo, Some(0.0));
+    assert_eq!(id.hi, Some(1000.0));
+    assert_eq!(id.hist.iter().sum::<u64>(), 101);
+
+    c.record_delete("t", &[row(&[("id", json!(1000)), ("name", json!("x"))])]);
+    let s = c.describe("t").unwrap();
+    assert_eq!(s.stats.get("id").unwrap().hist.iter().sum::<u64>(), 100);
+
+    c.record_delete("t", &rows);
+    let s = c.describe("t").unwrap();
+    let id = s.stats.get("id").unwrap();
+    assert_eq!(id.lo, None, "删空之后直方图清掉");
+    assert_eq!(id.hist.iter().sum::<u64>(), 0);
+}
+
 /// 老类型名并到新写法，参数保留，未知类型原样
 #[test]
 fn normalize_type_maps_legacy_names() {
diff --git a/chusql-core/storage/tests/runtime_test.rs b/chusql-core/storage/tests/runtime_test.rs
index 8d57231..4c64de6 100644
--- a/chusql-core/storage/tests/runtime_test.rs
+++ b/chusql-core/storage/tests/runtime_test.rs
@@ -1554,6 +1554,11 @@ fn stats_report_distinct_values() {
     assert!(r.contains(r#""row_count":3"#), "row_count: {}", r);
     assert!(r.contains(r#""name":"age","distinct":2"#), "age has 2 distinct values: {}", r);
     assert!(r.contains(r#""name":"id","distinct":3"#), "id has 3 distinct values: {}", r);
+    assert!(
+        r.contains(r#""name":"age","distinct":2,"capped":false,"lo":30.0,"hi":41.0"#),
+        "age carries a histogram range: {}",
+        r
+    );
 
     send(&mut c, r#"{"method":"delete_keys","table":"stat_t","keys":[1]}"#);
     let r = send(&mut c, r#"{"method":"describe_table","table":"stat_t"}"#);
diff --git a/chusql-interface/src/ChuSQL/Interface/AccountTable.hs b/chusql-interface/src/ChuSQL/Interface/AccountTable.hs
index a668cb3..cb13632 100644
--- a/chusql-interface/src/ChuSQL/Interface/AccountTable.hs
+++ b/chusql-interface/src/ChuSQL/Interface/AccountTable.hs
@@ -43,4 +43,4 @@ accountColumnSpec name ty nullable autoIncrement primary unique =
 
 -- | 账号表的结构（排序校验用）
 systemTableInfo :: TableInfo
-systemTableInfo = TableInfo accountTableName accountColumns 0 [] []
+systemTableInfo = TableInfo accountTableName accountColumns 0 [] [] []
diff --git a/chusql-server/src/ChuSQL/Server/Backend.hs b/chusql-server/src/ChuSQL/Server/Backend.hs
index 5f6009f..2c86b34 100644
--- a/chusql-server/src/ChuSQL/Server/Backend.hs
+++ b/chusql-server/src/ChuSQL/Server/Backend.hs
@@ -319,7 +319,7 @@ exprTables expr = case expr of
     -- | 两侧都取表名
     both a b = exprTables a ++ exprTables b
 
--- | 一张表的线上信息（内存实现：没有索引，统计现算）
+-- | 一张表的线上信息（内存实现：没有索引，统计与直方图现算）
 tableInfoOf :: (String, Table) -> TableInfo
 tableInfoOf (name, tbl) =
     TableInfo
@@ -328,10 +328,14 @@ tableInfoOf (name, tbl) =
         , tiRows = length (tableRows tbl)
         , tiIndexes = []
         , tiStats = [(c, distinctOf c, False) | (c, _) <- tableCols tbl]
+        , tiHistograms = [(c, h) | (c, _) <- tableCols tbl, Just h <- [histogramOf (columnValues c)]]
         }
   where
+    -- | 一列的所有值
+    columnValues c = [v | r <- tableRows tbl, Just v <- [lookup c r]]
+
     -- | 一列的不同值个数
-    distinctOf c = length (nub [v | r <- tableRows tbl, Just v <- [lookup c r]])
+    distinctOf c = length (nub (columnValues c))
 
 -- | 本地列类型转线上字符串（和 IPC 那边同一套写法）
 wireType :: Column -> String
diff --git a/chusql-server/test/Spec.hs b/chusql-server/test/Spec.hs
index 3914d18..269a783 100644
--- a/chusql-server/test/Spec.hs
+++ b/chusql-server/test/Spec.hs
@@ -6,13 +6,14 @@ module Main (main) where
 import ChuSQL.Core.Engine.Storage.IPC (
     Account (..),
     Request (ReqAccountReset),
+    TableInfo (..),
     closeConnection,
     localStorageLink,
     setStorageLink,
  )
 import ChuSQL.Core.Engine.Syntax.AST (Statement)
 import ChuSQL.Core.Engine.Syntax.Parser (parseStatement)
-import ChuSQL.Core.Model (Database, Row, Table (..), Value (..), pattern TInt, pattern TStr)
+import ChuSQL.Core.Model (Database, Histogram (..), Row, Table (..), Value (..), pattern TInt, pattern TStr)
 import ChuSQL.Interface.Auth (hashPasswordWith)
 import ChuSQL.Interface.Protocol (ClientRequest (..), Grant (..), RoleView (..), ServerResponse (..), decodeRequest, encodeResponse)
 import ChuSQL.Interface.RateLimit (newRateLimiter)
@@ -307,6 +308,21 @@ tcpSpec = do
             asBool (at "truncated" value) `shouldBe` False
             asText (at "database" value) `shouldBe` "test"
 
+    describe "memory backend: catalog statistics" $ do
+        it "reports a histogram for a numeric column" $ do
+            db <- newMVar testDb
+            let backend = memoryBackend testDatabaseName db
+            infos <- beCatalog backend
+            case infos of
+                Left err -> expectationFailure err
+                Right tables -> case [i | i <- tables, tiTable i == "users"] of
+                    [] -> expectationFailure "the catalog is missing the users table"
+                    (info : _) -> do
+                        fmap histLow (lookup "id" (tiHistograms info)) `shouldBe` Just 1
+                        fmap histHigh (lookup "id" (tiHistograms info)) `shouldBe` Just 5
+                        fmap (sum . histBuckets) (lookup "id" (tiHistograms info)) `shouldBe` Just 5
+                        lookup "name" (tiHistograms info) `shouldBe` Nothing
+
     describe "server config ([server] in chusql.toml)" $ do
         it "falls back to the defaults when the file has no server section" $ do
             path <- tcpConfigPath "server-default"
diff --git a/docs/architecture.md b/docs/architecture.md
index 0cb52b9..a647c60 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -53,6 +53,8 @@ chusql-server
 
 量词比较走另一个真节点 `QuantCmp`：解析时普通比较运算符先读最长符号，发现后面是 `ANY` / `ALL` 就让给量词解析，因此 `age > ANY (SELECT …)` 与 `age > 25` 共用同一层优先级。求值时先把子查询的行取成字面量，再摊成 `AND`（`ALL`）/ `OR`（`ANY`）链交给既有三值逻辑，于是空子查询 `ANY` 为假、`ALL` 为真，子查询里的 `NULL` 按三值规则传播。节点不带取反位（本方言没有前缀 `NOT`），取反写相反的运算符，如 `<> ALL` 与 `NOT IN` 同义。
 
+优化器的成本估算用存储维护的统计：行数、列级不同值数，以及数值列的列级等宽直方图。直方图是 16 个桶加区间下界 / 上界，只统计 `int` 与 `float` 列（NULL 与非数值不进直方图）；存储侧随插入累计、随删除递减，值跑出区间就把旧桶按中点并进新区间，删到表空时清空。这些统计随 schema 下沉到引擎，`Cost.rangeRows` 有直方图就按桶与区间的交集估命中行数（端点按闭区间算），没有直方图退回原来的四分之一；`Cost.selectivityOf` 把这个比例交给连接顺序估算与索引改写的成本把关，因此一条覆盖大半张表的范围条件不会再被改写成索引点查。在线协议里每条列统计多出 `lo` / `hi` / `hist` 三个可选字段，缺省即「没有直方图」，老客户端与老字典都能照常读。
+
 显式事务在会话层实现：`BEGIN` 把当前数据库整库读成内存快照，事务内的语句都在快照上执行；`SAVEPOINT` 把当前快照压进会话里的保存点栈，`ROLLBACK TO` 退回该点并保留它，`RELEASE` 丢掉该点与其后的保存点；`COMMIT` 先把快照、基线与库里现值按行 `id` 逐键比对，检测到写写冲突就报 `serialization_failure` 并保留事务；没有冲突才把差分用一条 `apply_transaction` 请求交给存储层应用；`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交：事务里看不到别的事务未提交或后来提交的改动，提交对外一次性可见，冲突由调用方回滚或重试。
 
 WAL 负责崩溃恢复：每条记录带自增 LSN，一组写操作与一条提交标记一次落盘；每组提交在数据和字典都刷盘之后把 `wal.checkpoint` 推进到这条提交的 LSN，含义是「到此为止的改动都已落盘」，再按这个边界丢掉日志前缀——所以日志里通常只有当组在飞的记录。进程重启时只重放检查点之后的记录，按标记分组取已提交的组、丢弃没有标记的整组，重放前逐条做幂等检查作为第二道保险，结束后再把检查点推到最后一条提交。LSN 从检查点与文件尾里较大的那个续号，跨重启单调，`SystemStatus` 应答里的 `last_lsn` 报告该检查点 LSN。建表、删表、建索引、删索引与数据改动走同一条 WAL，恢复时一并重放。