diff --git a/README.md b/README.md index 6df3814..342d116 100644 --- a/README.md +++ b/README.md @@ -43,16 +43,16 @@ > 如果找到了较好的Rust+Haskell集成方法也会及时替换 # 性能基准 - **测试环境**:Windows / GHC 9.10.3 / `-O2` **数据规模**:`users`、`orders` 各 500 行,两表配对共 25 万对 -**计时方式**:CPU 时间,自适应批量(累计超过 0.3 秒后折算成单次耗时) +**计时方式**:CPU 时间;每批至少 8 次迭代、累计超过 0.3 秒后固定批次,取 3 批中位数(避开时钟量化误差) | 查询 | 结果行数 | 未优化 | 优化后 | 加速 | | ------------------------------------------ | -------: | -------: | ----------: | --------: | -| `JOIN` + `WHERE u.age > 90` | 45 | 242.2 ms | **16.6 ms** | **14.6x** | -| `JOIN`(无 WHERE,只有投影) | 500 | 210.9 ms | 195.3 ms | 1.1x | -| `JOIN` + `WHERE` + `ORDER BY` + `LIMIT 10` | 10 | 164.1 ms | **14.2 ms** | **11.6x** | -| 单表 `WHERE age > 90`(对照组) | 45 | 0.175 ms | 0.290 ms | 0.6x | +| `JOIN` + `WHERE u.age > 90` | 45 | 156.3 ms | **12.2 ms** | **12.8x** | +| `JOIN`(无 WHERE,只有投影) | 500 | 169.9 ms | 146.5 ms | 1.2x | +| `JOIN` + `WHERE` + `ORDER BY` + `LIMIT 10` | 10 | 175.8 ms | **13.7 ms** | **12.9x** | +| 单表 `WHERE age > 90`(对照组) | 45 | 0.183 ms | 0.275 ms | 0.7x | -> 最近更新:2026-09-22 +> 跑法:`cd benchmark/chusql-engine && stack run`(可选参数:`用户行数 订单数 单批目标秒数`,默认 `500 500 0.3`) +> 最近更新:2026-09-24 diff --git a/benchmark/chusql-engine/src/Main.hs b/benchmark/chusql-engine/src/Main.hs index 44f1fa9..0448db3 100644 --- a/benchmark/chusql-engine/src/Main.hs +++ b/benchmark/chusql-engine/src/Main.hs @@ -7,6 +7,7 @@ import ChuSQL.Model import ChuSQL.Syntax.Parser (parseStatement) import Control.Exception (evaluate) import Data.IORef (IORef, newIORef, readIORef) +import Data.List (sort) import System.CPUTime (getCPUTime) import System.Environment (getArgs) import Text.Printf (printf) @@ -79,14 +80,33 @@ timed act = do t1 <- getCPUTime return (fromIntegral (t1 - t0) / 1e12) --- | 加倍批量次数再折算单次 +-- | 每批至少跑几次(减少 CPU 时钟量化误差) +minBatch :: Int +minBatch = 8 + +-- | 取几批中位数 +rounds :: Int +rounds = 3 + +-- | 跑一批 k 次,返回总耗时 +timeBatch :: Int -> IO Int -> IO Double +timeBatch k act = timed (mapM_ (const act) [1 .. k]) + +-- | 中位数 +median :: [Double] -> Double +median xs = sort xs !! (length xs `div` 2) + +-- | 自适应批量(够长且至少 8 次),取 3 批中位数 benchOp :: Double -> IO Int -> IO (Double, Int) -benchOp target act = loop 1 +benchOp target act = do + k <- pickBatch 1 + ts <- mapM (const (timeBatch k act)) [1 .. rounds] + return (median ts / fromIntegral k, k) where - -- \| 批次翻倍直到超过目标 - loop k = do - t <- timed (mapM_ (const act) [1 .. k]) - if t < target then loop (k * 2) else return (t / fromIntegral k, k) + -- \| 批次翻倍直到累计超过目标 + pickBatch k = do + t <- timeBatch k act + if t < target || k < minBatch then pickBatch (k * 2) else return k -- | 打印一个用例的对比 report :: IORef String -> Database -> Double -> String -> IO () @@ -105,7 +125,12 @@ report saltRef db target sql = do (tOp * 1000) speedup mismatch - printf " (实测批次:未优化 %d 次/批,优化后 %d 次/批)\n\n" kUn kOp + printf + " (%d 次/批 × %d 批取中位数:未优化 %d 次/批,优化后 %d 次/批)\n\n" + minBatch + rounds + kUn + kOp queries :: [String] queries = @@ -128,7 +153,7 @@ main = do target = pick 2 (0.3 :: Double) db = [("users", mkUsers n), ("orders", mkOrders n m)] printf "数据规模:users = %d 行,orders = %d 行\n" n m - printf "计时方式:CPU 时间;每个用例自动加倍批量次数,直到一批累计耗时超过 %.2f 秒,再折算成单次耗时\n" target + printf "计时方式:CPU 时间;每批至少 %d 次迭代、累计超过 %.2f 秒后固定批次,取 %d 批中位数\n" minBatch target rounds printf "(先跑一遍预热,避免把首次构造数据的开销算进去)\n\n" saltRef <- newIORef "" _ <- runOnce saltRef db warmupSql False diff --git a/chusql-engine/src/ChuSQL/Algebra/Eval.hs b/chusql-engine/src/ChuSQL/Algebra/Eval.hs index 714a309..9db8993 100644 --- a/chusql-engine/src/ChuSQL/Algebra/Eval.hs +++ b/chusql-engine/src/ChuSQL/Algebra/Eval.hs @@ -6,11 +6,12 @@ import ChuSQL.Algebra.Sort (sortRows) import ChuSQL.Model import ChuSQL.Storage (MonadStorage (..)) import ChuSQL.Syntax.AST (Expr (..)) -import Control.Monad (filterM) +import Control.Monad (filterM, foldM) -- 执行:按算子树算出结果行(Scan 阶段会加别名前缀)。 -- * 工具 + -- | 只保留清单里的列 project :: [String] -> Row -> Row project ["*"] row = row @@ -31,7 +32,21 @@ evalScan db mAlias tbl = do let prefix = maybe "" (++ ".") mAlias Right (map (addPrefix prefix) (tableRows table)) +-- | 把两边的行按条件配对(边配边筛,不先建整张积表) +filterPairs :: (Row -> Either String Bool) -> [Row] -> [Row] -> Either String [Row] +filterPairs cond lrows rrows = go lrows [] + where + go [] acc = Right (reverse acc) + go (l : ls) acc = do + acc' <- foldM (keep l) acc rrows + go ls acc' + keep l acc r = do + let row = l ++ r + ok <- cond row + pure (if ok then row : acc else acc) + -- * 求值 + -- | 纯求值(不需要存储) evalRelOp :: Database -> RelOp -> Either String [Row] evalRelOp db (Scan mAlias tbl) = evalScan db mAlias tbl @@ -53,37 +68,41 @@ evalRelOp db (Limit n op) = do evalRelOp db (Join left right cond) = do lrows <- evalRelOp db left rrows <- evalRelOp db right - let cross = [l ++ r | l <- lrows, r <- rrows] - filterM (evalCondForRow cond) cross + filterPairs (evalCondForRow cond) lrows rrows -- | 单子求值:Lookup 问存储 -evalRelOpM :: (MonadStorage m) => Database -> RelOp -> m (Either String [Row]) -evalRelOpM db (Scan a t) = pure (evalScan db a t) -evalRelOpM _ (Lookup t k) = do +evalRelOpM :: (MonadStorage m) => RelOp -> m (Either String [Row]) +evalRelOpM (Scan mAlias t) = do + result <- scan t + pure $ do + rows <- result + case mAlias of + Nothing -> Right rows + Just a -> Right (map (addPrefix (a ++ ".")) rows) +evalRelOpM (Lookup t k) = do result <- lookupByKey t k pure $ case result of Left e -> Left e Right Nothing -> Right [] - Right (Just r) -> Right [r] -evalRelOpM db (Filter p x) = do - result <- evalRelOpM db x + Right (Just row) -> Right [row] +evalRelOpM (Filter p x) = do + result <- evalRelOpM x pure $ do rows <- result filterM (evalCondForRow p) rows -evalRelOpM db (Project cols x) = do - result <- evalRelOpM db x - pure (map (project cols) <$> result) -evalRelOpM db (Sort specs x) = do - result <- evalRelOpM db x - pure (sortRows specs <$> result) -evalRelOpM db (Limit n x) = do - result <- evalRelOpM db x - pure (take n <$> result) -evalRelOpM db (Join l r c) = do - ls <- evalRelOpM db l - rs <- evalRelOpM db r +evalRelOpM (Project cols x) = do + result <- evalRelOpM x + pure (fmap (map (project cols)) result) +evalRelOpM (Sort spec x) = do + result <- evalRelOpM x + pure (fmap (sortRows spec) result) +evalRelOpM (Limit n x) = do + result <- evalRelOpM x + pure (fmap (take n) result) +evalRelOpM (Join l r cond) = do + ls <- evalRelOpM l + rs <- evalRelOpM r pure $ do - leftRows <- ls - rightRows <- rs - let cross = [x ++ y | x <- leftRows, y <- rightRows] - filterM (evalCondForRow c) cross + left <- ls + right <- rs + filterPairs (evalCondForRow cond) left right diff --git a/chusql-engine/src/ChuSQL/Algebra/Sort.hs b/chusql-engine/src/ChuSQL/Algebra/Sort.hs index 0dbb3a0..5e64cd7 100644 --- a/chusql-engine/src/ChuSQL/Algebra/Sort.hs +++ b/chusql-engine/src/ChuSQL/Algebra/Sort.hs @@ -19,21 +19,26 @@ compareValue (VStr a) (VStr b) = compare a b compareValue (VBool a) (VBool b) = compare a b compareValue _ _ = EQ --- | 按排序要求比较两行 -compareRows :: [(String, SortDir)] -> Row -> Row -> Ordering -compareRows [] _ _ = EQ -compareRows ((col, dir) : rest) r1 r2 = - case (lookup col r1, lookup col r2) of +-- | 取出排序键(按 spec 顺序,每行只查一次) +sortKey :: [(String, SortDir)] -> Row -> [Maybe Value] +sortKey spec row = [lookup col row | (col, _) <- spec] + +-- | 比较两个排序键 +compareKeys :: [(String, SortDir)] -> [Maybe Value] -> [Maybe Value] -> Ordering +compareKeys [] _ _ = EQ +compareKeys ((_, dir) : rest) (a : as) (b : bs) = + case (a, b) of (Just v1, Just v2) -> case compareValue v1 v2 of - EQ -> compareRows rest r1 r2 - o -> - if dir == Desc - then flipOrdering o - else o - _ -> compareRows rest r1 r2 + EQ -> compareKeys rest as bs + o -> if dir == Desc then flipOrdering o else o + _ -> compareKeys rest as bs +compareKeys _ _ _ = EQ --- | 排序若干行 +-- | 排序若干行(先取键再排,避免每次比较都查列) sortRows :: [(String, SortDir)] -> [Row] -> [Row] sortRows [] rows = rows -sortRows spec rows = sortBy (compareRows spec) rows +sortRows spec rows = map snd (sortBy cmp (map decorate rows)) + where + decorate row = (sortKey spec row, row) + cmp (k1, _) (k2, _) = compareKeys spec k1 k2 diff --git a/chusql-engine/src/ChuSQL/Engine.hs b/chusql-engine/src/ChuSQL/Engine.hs index 03caff6..892fab5 100644 --- a/chusql-engine/src/ChuSQL/Engine.hs +++ b/chusql-engine/src/ChuSQL/Engine.hs @@ -14,6 +14,7 @@ import Control.Monad (filterM) -- 引擎入口:语义检查 + 分发执行,以及内存实现与泛型版本。 -- * 对外入口 + -- | 跑一条语句(内存实现) runStatement :: Database -> Statement -> Either String (Database, [Row]) runStatement db q = do @@ -25,35 +26,36 @@ runStatement db q = do rowsOf :: Either String (Database, [Row]) -> Either String [Row] rowsOf = fmap snd - -- | 泛型入口:先检查再执行 runStatementM :: (MonadStorage m) => Statement -> m (Either String [Row]) runStatementM q = do db <- snapshot case check db q of Left err -> pure (Left err) - Right _ -> runStatementUncheckedM q + Right _ -> runStatementUncheckedM db q -- * 分发执行 + -- | 按语句类型分发(已检查过) -runStatementUncheckedM :: (MonadStorage m) => Statement -> m (Either String [Row]) -runStatementUncheckedM q@Select{} = do - db <- snapshot +runStatementUncheckedM :: (MonadStorage m) => Database -> Statement -> m (Either String [Row]) +runStatementUncheckedM db q@Select{} = case translate q of Left e -> pure (Left e) - Right relOp -> evalRelOpM db (optimize db relOp) - -runStatementUncheckedM (Insert tbl cols vals) = + Right relOp -> evalRelOpM (optimize db relOp) +runStatementUncheckedM _ (Insert tbl cols vals) = case mapM (\e -> evalExpr e []) vals of Left err -> pure (Left err) Right values | length cols /= length values -> pure (Left "column count does not match value count") | otherwise -> do - result <- insert tbl (zip cols values) + let row = zip cols values + key = case lookup "id" row of + Just (VInt k) -> Just k + _ -> Nothing + result <- insert tbl row key pure (result >> Right []) - -runStatementUncheckedM (Delete tbl mWhere) = do +runStatementUncheckedM _ (Delete tbl mWhere) = do rowsResult <- scan tbl case rowsResult of Left err -> pure (Left err) @@ -70,8 +72,7 @@ runStatementUncheckedM (Delete tbl mWhere) = do -- \| 保留 = 条件不成立 shouldKeep :: Expr -> Row -> Either String Bool shouldKeep e row = not <$> evalCondForRow e row - -runStatementUncheckedM (Update tbl assigns mWhere) = do +runStatementUncheckedM _ (Update tbl assigns mWhere) = do rowsResult <- scan tbl case rowsResult of Left err -> pure (Left err) @@ -90,15 +91,15 @@ runStatementUncheckedM (Update tbl assigns mWhere) = do Nothing -> Right True Just e -> evalCondForRow e row if keep then applyUpdates asgns row else Right row - -runStatementUncheckedM (CreateTable name cols) = do +runStatementUncheckedM _ (CreateTable name cols) = do result <- createTable name cols pure (result >> Right []) -runStatementUncheckedM (DropTable name) = do +runStatementUncheckedM _ (DropTable name) = do result <- dropTable name pure (result >> Right []) -- * 更新辅助 + -- | 依次求值并覆盖列 applyUpdates :: [(String, Expr)] -> Row -> Either String Row applyUpdates [] row = Right row diff --git a/chusql-engine/src/ChuSQL/Storage.hs b/chusql-engine/src/ChuSQL/Storage.hs index 8ee9a26..b06a0e5 100644 --- a/chusql-engine/src/ChuSQL/Storage.hs +++ b/chusql-engine/src/ChuSQL/Storage.hs @@ -12,7 +12,7 @@ class (Monad m) => MonadStorage m where scan :: String -> m (Either String [Row]) -- \| 追加一行 - insert :: String -> Row -> m (Either String ()) + insert :: String -> Row -> Maybe Int -> m (Either String ()) -- \| 整表替换 replaceAll :: String -> [Row] -> m (Either String ()) diff --git a/chusql-engine/src/ChuSQL/Storage/IPC.hs b/chusql-engine/src/ChuSQL/Storage/IPC.hs index facf87c..3f31e00 100644 --- a/chusql-engine/src/ChuSQL/Storage/IPC.hs +++ b/chusql-engine/src/ChuSQL/Storage/IPC.hs @@ -7,17 +7,19 @@ module ChuSQL.Storage.IPC ( Response (..), SchemaColumn (..), sendRequest, + closeConnection, doListTables, doLookupByKey, - doInsertKeyed, + doInsertWith, doDescribeTable, - doCreateTable, doDropTable, + doListCatalog, ) where import ChuSQL.Model import ChuSQL.Storage -import Control.Exception (IOException, bracket, try) +import Control.Concurrent.MVar (MVar, modifyMVar, modifyMVar_, newMVar) +import Control.Exception (IOException, try) import Data.Aeson (FromJSON (..), ToJSON (..), eitherDecodeStrict, encode, object, withObject, (.:), (.=)) import Data.Aeson qualified as A import Data.Aeson.Key qualified as K @@ -25,15 +27,16 @@ import Data.Aeson.KeyMap qualified as KM import Data.Aeson.Types (Parser) import Data.ByteString.Char8 qualified as BSC import Data.ByteString.Lazy qualified as BL -import Data.List (nub) import Data.Scientific (floatingOrInteger) import Data.Text qualified as T import System.Environment (lookupEnv) import System.IO +import System.IO.Unsafe (unsafePerformIO) -- IPC 存储实现:每个存储方法翻成管道上的一条 JSON 请求。 -- * 存储实现 + -- | 命名管道上的存储 newtype IPCStorage a = IPCStorage { runIPCStorage :: IO a @@ -54,45 +57,80 @@ instance Monad IPCStorage where a <- m runIPCStorage (k a) +-- * 全局连接缓存 + +-- | 全局句柄缓存:按管道路径缓存,路径变了就换连接。 +{-# NOINLINE connectionRef #-} +connectionRef :: MVar (Maybe (FilePath, Handle)) +connectionRef = unsafePerformIO (newMVar Nothing) + +-- | 打开管道 +openPipe :: FilePath -> IO Handle +openPipe path = do + h <- openFile path ReadWriteMode + hSetBuffering h LineBuffering + hSetNewlineMode h noNewlineTranslation + pure h + +-- | 关掉缓存里的句柄(没有就什么都不做) +closeCached :: Maybe (FilePath, Handle) -> IO () +closeCached Nothing = pure () +closeCached (Just (_, h)) = hClose h + +-- | 拿一个可用句柄;路径没变就复用,变了就换。 +getConnection :: FilePath -> IO Handle +getConnection path = modifyMVar connectionRef $ \cached -> case cached of + Just (p, h) + | p == path -> pure (Just (p, h), h) + _ -> do + _ <- try (closeCached cached) :: IO (Either IOException ()) + h <- openPipe path + pure (Just (path, h), h) + +-- | 丢掉缓存的连接(出错后调用,下次会重连) +dropConnection :: IO () +dropConnection = modifyMVar_ connectionRef $ \cached -> do + _ <- try (closeCached cached) :: IO (Either IOException ()) + pure Nothing + +-- | 关闭并清空连接。可选,进程退出前调用。 +closeConnection :: IO () +closeConnection = dropConnection + -- * 管道 + -- | 管道路径 pipePath :: IO String pipePath = do name <- maybe "chusql-storage" id <$> lookupEnv "CHUSQL_PIPE" pure ("\\\\.\\pipe\\" ++ name) --- | 开管道跑闭包 -withPipe :: (Handle -> IO a) -> IO a -withPipe = bracket openPipe hClose - where - -- \| 打开管道句柄 - openPipe = do - path <- pipePath - h <- openFile path ReadWriteMode - hSetBuffering h LineBuffering - hSetNewlineMode h noNewlineTranslation - pure h - --- | 发一条请求读一条响应 +-- | 发一条请求读一条响应;管道不通就回 RespError,并丢掉缓存等下次重连。 sendRequest :: Request -> IO Response sendRequest req = do - result <- try (withPipe roundTrip) :: IO (Either IOException Response) - pure $ case result of - Left e -> RespError ("pipe error: " ++ show e) - Right r -> r - where - roundTrip h = do - BL.hPutStr h (encode req) - BSC.hPutStr h "\n" - hFlush h - line <- BSC.hGetLine h - let cleaned = BSC.dropWhileEnd (== '\r') line - case eitherDecodeStrict cleaned of - Left err -> pure (RespError ("decode: " ++ err)) - Right r -> pure r + path <- pipePath + result <- try (roundTrip req path) :: IO (Either IOException Response) + case result of + Left e -> do + dropConnection + pure (RespError ("pipe error: " ++ show e)) + Right r -> pure r +-- | 一条请求-响应往返(复用缓存里的句柄) +roundTrip :: Request -> FilePath -> IO Response +roundTrip req path = do + h <- getConnection path + BL.hPutStr h (encode req) + BSC.hPutStr h "\n" + hFlush h + line <- BSC.hGetLine h + let cleaned = BSC.dropWhileEnd (== '\r') line + case eitherDecodeStrict cleaned of + Left err -> pure (RespError ("decode: " ++ err)) + Right r -> pure r -- * 编解码 + -- | 一列的 schema data SchemaColumn = SchemaColumn { scName :: String @@ -112,6 +150,7 @@ instance FromJSON SchemaColumn where pure (SchemaColumn n t) -- * 请求 + -- | 请求类型 data Request = ReqPing @@ -123,6 +162,7 @@ data Request | ReqDescribeTable String | ReqCreateTable String [SchemaColumn] | ReqDropTable String + | ReqListCatalog -- | 请求编码 instance ToJSON Request where @@ -163,7 +203,7 @@ instance ToJSON Request where ] toJSON (ReqDropTable t) = object ["method" .= ("drop_table" :: T.Text), "table" .= t] - + toJSON ReqListCatalog = object ["method" .= ("list_catalog" :: T.Text)] -- | 响应类型 data Response @@ -173,6 +213,7 @@ data Response | RespSchema [SchemaColumn] Int | RespOk | RespError String + | RespCatalog [(String, [SchemaColumn], Int)] -- | 响应解码 instance FromJSON Response where @@ -188,7 +229,18 @@ instance FromJSON Response where cols <- o .: "columns" cnt <- o .: "row_count" pure (RespSchema cols cnt) + "catalog" -> do + xs <- o .: "schemas" + parsed <- mapM parseTableSchema xs + pure (RespCatalog parsed) other -> fail ("unknown status: " ++ T.unpack other) + where + -- \| 解析一条 list_catalog 记录。 + parseTableSchema = withObject "TableSchema" $ \o -> do + t <- o .: "table" + cols <- o .: "columns" + cnt <- o .: "row_count" + pure (t, cols, cnt) -- | 值编码成 JSON valueToJSON :: Value -> A.Value @@ -221,8 +273,8 @@ rowFromJSON (A.Object o) = mapM toPair (KM.toList o) pure (K.toString k, val) rowFromJSON _ = fail "row must be a JSON object" - -- * 内部封装 + -- | 发 Scan doScan :: String -> IO (Either String [Row]) doScan t = do @@ -232,25 +284,10 @@ doScan t = do RespError e -> Left e _ -> Left "unexpected response to scan" --- | 发 Insert(带 id 键) -doInsert :: String -> Row -> IO (Either String ()) -doInsert t r = do - resp <- sendRequest (ReqInsert t r (rowKey r)) - pure $ case resp of - RespOk -> Right () - RespError e -> Left e - _ -> Left "unexpected response to insert" - --- | 取行里的 id 作索引键 -rowKey :: Row -> Maybe Int -rowKey r = case lookup "id" r of - Just (VInt k) -> Just k - _ -> Nothing - --- | 发 Insert(指定键) -doInsertKeyed :: String -> Row -> Int -> IO (Either String ()) -doInsertKeyed t r k = do - resp <- sendRequest (ReqInsert t r (Just k)) +-- | 发 Insert,可带索引键。 +doInsertWith :: String -> Row -> Maybe Int -> IO (Either String ()) +doInsertWith t r mk = do + resp <- sendRequest (ReqInsert t r mk) pure $ case resp of RespOk -> Right () RespError e -> Left e @@ -274,6 +311,15 @@ doListTables = do RespError e -> Left e _ -> Left "unexpected response to list_tables" +-- | 一次拿全库 schema(表名 + 列 + 行数)。 +doListCatalog :: IO (Either String [(String, [SchemaColumn], Int)]) +doListCatalog = do + resp <- sendRequest ReqListCatalog + pure $ case resp of + RespCatalog xs -> Right xs + RespError e -> Left e + _ -> Left "unexpected response to list_catalog" + -- | 发 LookupByIndex doLookupByKey :: String -> Int -> IO (Either String (Maybe Row)) doLookupByKey t k = do @@ -312,6 +358,7 @@ doDropTable t = do _ -> Left "unexpected response to drop_table" -- * 表结构 + -- | 线上 schema 转本地列 schemaToColumns :: [SchemaColumn] -> [(String, Column)] schemaToColumns = map go @@ -321,45 +368,15 @@ schemaToColumns = map go go (SchemaColumn n "bool") = (n, TBool) go (SchemaColumn n _) = (n, TStr) --- | 从行推断列和类型 -inferColumns :: [Row] -> [(String, Column)] -inferColumns rows = [(n, inferType n) | n <- names] - where - -- \| 出现过的列名 - names = nub (concatMap (map fst) rows) - - -- \| 按第一个值推断类型 - inferType n = case [v | r <- rows, Just v <- [lookup n r]] of - (VInt _ : _) -> TInt - (VStr _ : _) -> TStr - (VBool _ : _) -> TBool - [] -> TStr - --- | 扫一张表拼成 Table -loadTable :: String -> IO Table -loadTable t = do - schemaR <- doDescribeTable t - case schemaR of - Right (cols, _n) -> do - rowsR <- doScan t - case rowsR of - Right rows -> pure (Table t (schemaToColumns cols) rows) - Left _ -> pure (Table t (schemaToColumns cols) []) - Left _ -> do - rowsR <- doScan t - case rowsR of - Right rows -> pure (Table t (inferColumns rows) rows) - Left _ -> pure (Table t [] []) - - -- * MonadStorage 实例 + -- | 存储方法全走管道 instance MonadStorage IPCStorage where -- \| 发 Scan scan t = IPCStorage (doScan t) -- \| 发 Insert - insert t r = IPCStorage (doInsert t r) + insert t r mk = IPCStorage (doInsertWith t r mk) -- \| 发 ReplaceAll replaceAll t rs = IPCStorage (doReplaceAll t rs) @@ -380,12 +397,12 @@ instance MonadStorage IPCStorage where -- \| 列表 + 逐表扫描拼库 snapshot = IPCStorage $ do - result <- doListTables + result <- doListCatalog case result of Left _ -> pure [] - Right ts -> mapM loadDbEntry ts + Right xs -> mapM loadEntry xs where - -- \| 加载一个 (表名, 表) 对 - loadDbEntry t = do - tbl <- loadTable t - pure (t, tbl) + -- \| 加载一张表(schema 来自 catalog,行来自 scan) + loadEntry (t, cols, _n) = do + rows <- doScan t + pure (t, Table t (schemaToColumns cols) (either (const []) id rows)) diff --git a/chusql-engine/src/ChuSQL/Storage/Memory.hs b/chusql-engine/src/ChuSQL/Storage/Memory.hs index 0e9b46f..41f1bf1 100644 --- a/chusql-engine/src/ChuSQL/Storage/Memory.hs +++ b/chusql-engine/src/ChuSQL/Storage/Memory.hs @@ -2,10 +2,12 @@ module ChuSQL.Storage.Memory (MemoryStorage (runMemoryStorage)) where import ChuSQL.Model import ChuSQL.Storage +import Data.List (find) -- 内存实现:状态是 Database,错误通道是 Either String。 -- * 实例 + -- | 内存存储:Database 上的状态 newtype MemoryStorage a = MemoryStorage {runMemoryStorage :: Database -> Either String (a, Database)} @@ -37,7 +39,7 @@ instance MonadStorage MemoryStorage where Right (Right (tableRows tbl), db) -- \| 追加一行 - insert t r = MemoryStorage $ \db -> do + insert t r _ = MemoryStorage $ \db -> do tbl <- lookupTable db t Right (Right (), replaceTable t tbl{tableRows = tableRows tbl ++ [r]} db) @@ -67,11 +69,10 @@ instance MonadStorage MemoryStorage where snapshot = MemoryStorage $ \db -> Right (db, db) -- * 工具 + -- | 按 id 列找一行 findRow :: Table -> Int -> Maybe Row -findRow tbl k = case filter matches (tableRows tbl) of - (r : _) -> Just r - [] -> Nothing +findRow tbl k = find matches (tableRows tbl) where -- \| 这行的 id 是否等于 k matches r = lookup "id" r == Just (VInt k) diff --git a/chusql-engine/src/ChuSQL/Syntax/AST.hs b/chusql-engine/src/ChuSQL/Syntax/AST.hs index 7dff886..9398e35 100644 --- a/chusql-engine/src/ChuSQL/Syntax/AST.hs +++ b/chusql-engine/src/ChuSQL/Syntax/AST.hs @@ -5,6 +5,7 @@ import ChuSQL.Model (Column (..)) -- 语法树:语句、数据来源、条件表达式、排序方向。 -- * 语句 + -- | 一条语句:查 / 加 / 删 / 改 / 建表 / 删表 data Statement = Select @@ -28,6 +29,7 @@ data FromClause deriving (Show, Eq) -- * 表达式 + -- | 条件表达式 data Expr = Col String @@ -42,6 +44,7 @@ data Expr deriving (Show, Eq) -- * 排序 + -- | 排序方向 data SortDir = Asc @@ -49,6 +52,7 @@ data SortDir deriving (Show, Eq) -- * 构造 + -- | 造一个最简单的 SELECT makeSelect :: [String] -> String -> Maybe Expr -> Statement makeSelect cols tbl w = diff --git a/chusql-engine/test/Spec.hs b/chusql-engine/test/Spec.hs index f377aea..b899f85 100644 --- a/chusql-engine/test/Spec.hs +++ b/chusql-engine/test/Spec.hs @@ -11,8 +11,9 @@ import ChuSQL.Storage.IPC (IPCStorage (runIPCStorage), Request (ReqPing), Respon import ChuSQL.Syntax.AST import ChuSQL.Syntax.Parser import Control.Concurrent (threadDelay) -import Control.Exception (IOException, bracket, try) +import Control.Exception (IOException, bracket, finally, try) import Data.List (isInfixOf, sortOn) +import Data.Unique (hashUnique, newUnique) import System.CPUTime (getCPUTime) import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, getTemporaryDirectory, removePathForcibly) import System.Environment (getEnvironment, lookupEnv, setEnv, unsetEnv) @@ -340,7 +341,12 @@ main = hspec $ do it "returns Left when a value has the wrong type for comparison" $ do runStatement testDB (Insert "users" ["name"] [Col "other"]) `shouldSatisfy` isLeft - + it "writes id column to index and queries it back" $ do + case parseStatement "INSERT INTO users (id, name, age) VALUES (42, 'Zoe', 20)" >>= runStatement testDB of + Left err -> expectationFailure err + Right (db', _) -> do + rowsOf (parseStatement "SELECT name FROM users WHERE id = 42" >>= runStatement db') + `shouldBe` Right [[("name", VStr "Zoe")]] describe "ChuSQL.Engine (DELETE)" $ do it "parses DELETE with WHERE" $ do parseStatement "DELETE FROM users WHERE age < 18" @@ -1102,7 +1108,7 @@ main = hspec $ do it "scan returns rows with all three value types after insert" $ do withTestServer $ \srv -> withServerEnv srv $ do let row = [("id", VInt 7), ("name", VStr "Alice"), ("flag", VBool True)] - runIPCStorage (insert "users" row) `shouldReturn` Right () + runIPCStorage (insert "users" row (Just 7)) `shouldReturn` Right () rows <- runIPCStorage (scan "users") (map sortRow <$> rows) `shouldBe` Right [sortRow row] @@ -1113,7 +1119,7 @@ main = hspec $ do it "scan returns all 20 inserted rows in order" $ do withTestServer $ \srv -> withServerEnv srv $ do - mapM_ (\i -> runIPCStorage (insert "many" [("id", VInt i)])) [1 .. 20 :: Int] + mapM_ (\i -> runIPCStorage (insert "many" [("id", VInt i)] (Just i))) [1 .. 20 :: Int] rows <- runIPCStorage (scan "many") fmap length rows `shouldBe` Right 20 fmap (sortOn show . map (lookup "id")) rows @@ -1121,7 +1127,7 @@ main = hspec $ do it "replaceAll leaves only the new rows" $ do withTestServer $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "t" [("id", VInt 1)]) `shouldReturn` Right () + runIPCStorage (insert "t" [("id", VInt 1)] (Just 1)) `shouldReturn` Right () runIPCStorage (replaceAll "t" [[("id", VInt 9)], [("id", VInt 8)]]) `shouldReturn` Right () rows <- runIPCStorage (scan "t") fmap (sortOn show . map (lookup "id")) rows @@ -1130,7 +1136,7 @@ main = hspec $ do it "lookupByKey finds a row inserted with an id over the pipe" $ do withTestServer $ \srv -> withServerEnv srv $ do let row = [("id", VInt 7), ("name", VStr "Zoe")] - runIPCStorage (insert "keyed" row) `shouldReturn` Right () + runIPCStorage (insert "keyed" row (Just 7)) `shouldReturn` Right () found <- runIPCStorage (lookupByKey "keyed" 7) fmap (fmap sortRow) found `shouldBe` Right (Just (sortRow row)) missing <- runIPCStorage (lookupByKey "keyed" 8) @@ -1138,7 +1144,7 @@ main = hspec $ do it "replaceAll rebuilds the index over the pipe" $ do withTestServer $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "rb" [("id", VInt 1)]) `shouldReturn` Right () + runIPCStorage (insert "rb" [("id", VInt 1)] (Just 1)) `shouldReturn` Right () runIPCStorage (replaceAll "rb" [[("id", VInt 9), ("name", VStr "Zoe")]]) `shouldReturn` Right () old <- runIPCStorage (lookupByKey "rb" 1) old `shouldBe` Right Nothing @@ -1148,8 +1154,8 @@ main = hspec $ do it "snapshot lists tables and scans each one to build the database" $ do withTestServer $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "a" [("id", VInt 1)]) `shouldReturn` Right () - runIPCStorage (insert "b" [("id", VInt 2)]) `shouldReturn` Right () + runIPCStorage (insert "a" [("id", VInt 1)] (Just 1)) `shouldReturn` Right () + runIPCStorage (insert "b" [("id", VInt 2)] (Just 2)) `shouldReturn` Right () db <- runIPCStorage snapshot map fst db `shouldBe` ["a", "b"] map (length . tableRows . snd) db `shouldBe` [1, 1] @@ -1157,8 +1163,8 @@ main = hspec $ do it "doListTables returns the tables that exist" $ do withTestServer $ \srv -> withServerEnv srv $ do - runIPCStorage (insert "t1" [("id", VInt 1)]) `shouldReturn` Right () - runIPCStorage (insert "t2" [("id", VInt 2)]) `shouldReturn` Right () + runIPCStorage (insert "t1" [("id", VInt 1)] (Just 1)) `shouldReturn` Right () + runIPCStorage (insert "t2" [("id", VInt 2)] (Just 2)) `shouldReturn` Right () doListTables `shouldReturn` Right ["t1", "t2"] it "rows survive a restart on the same data directory" $ do @@ -1168,7 +1174,7 @@ main = hspec $ do Right bin -> do srv1 <- startServer bin let dir = dataDir srv1 - withServerEnv srv1 $ runIPCStorage (insert "persist" [("id", VInt 42)]) `shouldReturn` Right () + withServerEnv srv1 $ runIPCStorage (insert "persist" [("id", VInt 42)] (Just 42)) `shouldReturn` Right () stopServer srv1 srv2 <- startServerAt bin dir rows <- withServerEnv srv2 $ runIPCStorage (scan "persist") @@ -1180,6 +1186,31 @@ main = hspec $ do withPipeName "chusql-no-such-server" $ do result <- runIPCStorage (scan "users") result `shouldSatisfy` isLeft + it "INSERT then SELECT WHERE id = k via IPC" $ do + withTestServer $ \srv -> withServerEnv srv $ do + _ <- + runIPCStorage + ( runStatementM + (CreateTable "ipc_idx" [("id", TInt), ("name", TStr)]) + ) + _ <- + runIPCStorage + ( runStatementM + (Insert "ipc_idx" ["id", "name"] [LitInt 42, LitStr "Zoe"]) + ) + result <- + runIPCStorage + ( runStatementM + ( Select + { selectCols = ["name"] + , selectFrom = FromTable Nothing "ipc_idx" + , selectWhere = Just (Eq (Col "id") (LitInt 42)) + , selectOrderBy = [] + , selectLimit = Nothing + } + ) + ) + result `shouldBe` Right [[("name", VStr "Zoe")]] it "parses CREATE TABLE with two columns" $ do parseStatement "CREATE TABLE users (id INT, name TEXT)" `shouldBe` Right (CreateTable "users" [("id", TInt), ("name", TStr)]) @@ -1280,7 +1311,7 @@ locateServer = do startServer :: FilePath -> IO IPCServer startServer bin = do - stamp <- show <$> getCPUTime + stamp <- uniqueStamp tmp <- getTemporaryDirectory let dir = tmp "chusql-hs-IPC" stamp createDirectoryIfMissing True dir @@ -1288,7 +1319,7 @@ startServer bin = do startServerAt :: FilePath -> FilePath -> IO IPCServer startServerAt bin dir = do - stamp <- show <$> getCPUTime + stamp <- uniqueStamp let name = "chusql-hs-test-" ++ stamp parentEnv <- getEnvironment (_, _, _, ph) <- @@ -1302,6 +1333,13 @@ startServerAt bin dir = do waitUntilReady srv 100 pure srv +-- | 每次调用都不同的后缀:CPU 时间精度不够,再加上一个唯一编号。 +uniqueStamp :: IO String +uniqueStamp = do + cpu <- getCPUTime + u <- hashUnique <$> newUnique + pure (show cpu ++ "-" ++ show u) + waitUntilReady :: IPCServer -> Int -> IO () waitUntilReady srv tries = do probe <- try (withServerEnv srv (sendRequest ReqPing)) :: IO (Either IOException Response) @@ -1337,10 +1375,9 @@ withServerEnv srv act = do oldDir <- lookupEnv "CHUSQL_DATA_DIR" setEnv "CHUSQL_PIPE" (pipeName srv) setEnv "CHUSQL_DATA_DIR" (dataDir srv) - r <- act - restore "CHUSQL_PIPE" oldPipe - restore "CHUSQL_DATA_DIR" oldDir - pure r + act `finally` do + restore "CHUSQL_PIPE" oldPipe + restore "CHUSQL_DATA_DIR" oldDir where restore k = maybe (unsetEnv k) (setEnv k) diff --git a/chusql-storage/src/bin/server.rs b/chusql-storage/src/bin/server.rs index a4387b9..18c397b 100644 --- a/chusql-storage/src/bin/server.rs +++ b/chusql-storage/src/bin/server.rs @@ -9,7 +9,7 @@ use chusql_storage::catalog::Catalog; use chusql_storage::config::{self, Config, Loaded}; use chusql_storage::heap::HeapTable; use chusql_storage::log; -use chusql_storage::protocol::{Request, Response, SchemaColumn}; +use chusql_storage::protocol::{Request, Response, SchemaColumn, TableSchemaWire}; use chusql_storage::wal::{Wal, WalOp}; use chusql_storage::{log_debug, log_error, log_info, log_warn}; use interprocess::local_socket::{ @@ -28,20 +28,22 @@ struct Server { cfg: Config, tables: Mutex>>>, write_lock: Mutex<()>, + catalog: Mutex, } impl Server { /// 新建服务器状态 fn new(cfg: Config) -> std::io::Result { std::fs::create_dir_all(&cfg.data_dir)?; + let catalog = Catalog::load(cfg.data_dir.join("catalog.json"))?; Ok(Server { cfg, tables: Mutex::new(HashMap::new()), write_lock: Mutex::new(()), + catalog: Mutex::new(catalog), }) } - // 路径 /// 表名 -> 数据文件 fn table_path(&self, table: &str) -> PathBuf { @@ -125,7 +127,9 @@ impl Server { table: &str, f: impl FnOnce(&mut HeapTable) -> std::io::Result, ) -> std::io::Result { - if !self.table_exists(table) { + // 已经在缓存里的表不用再查磁盘(省一次 metadata 系统调用) + let cached = self.tables.lock().unwrap().contains_key(table); + if !cached && !self.table_exists(table) { return Err(std::io::Error::new( std::io::ErrorKind::NotFound, format!("unknown table: {}", table), @@ -143,19 +147,21 @@ impl Server { None => self.with_table(table, |t| t.insert(row))?, Some(k) => self.with_indexed_table(table, |t| t.insert_keyed(row, *k))?, } - let mut c = Catalog::load(self.catalog_path())?; let row: chusql_storage::protocol::Row = row.clone().into_iter().collect(); - c.record_insert(table, &row); - c.save(self.catalog_path())?; + let mut c = self.catalog.lock().unwrap(); + if c.record_insert(table, &row) { + c.save(self.catalog_path())?; + } Ok(()) } WalOp::ReplaceAll { table, rows } => { self.with_indexed_table(table, |t| t.replace_all(rows))?; - let mut c = Catalog::load(self.catalog_path())?; let rows: Vec = rows.iter().map(|r| r.clone().into_iter().collect()).collect(); - c.record_replace_all(table, &rows); - c.save(self.catalog_path())?; + let mut c = self.catalog.lock().unwrap(); + if c.record_replace_all(table, &rows) { + c.save(self.catalog_path())?; + } Ok(()) } } @@ -298,6 +304,21 @@ impl Server { } } + Request::ListCatalog => { + log_debug!(pipe, "list_catalog"); + let c = self.catalog.lock().unwrap(); + let schemas: Vec = c + .all_tables() + .into_iter() + .map(|(name, ts)| TableSchemaWire { + table: name.to_string(), + columns: ts.columns.clone(), + row_count: ts.row_count, + }) + .collect(); + Response::Catalog { schemas } + } + Request::DescribeTable { table } => { log_debug!(pipe, "describe_table table={}", table); match self.describe_table(&table) { @@ -317,18 +338,13 @@ impl Server { Request::CreateTable { table, columns } => { log_debug!(pipe, "create_table table={} cols={}", table, columns.len()); - let mut c = match Catalog::load(self.catalog_path()) { - Ok(c) => c, - Err(e) => { + { + let c = self.catalog.lock().unwrap(); + if c.describe(&table).is_some() { return Response::Error { - message: format!("create_table {}: catalog: {}", table, e), - } + message: format!("create_table {}: table already exists", table), + }; } - }; - if c.describe(&table).is_some() { - return Response::Error { - message: format!("create_table {}: table already exists", table), - }; } if let Err(e) = self.with_table(&table, |_t| Ok(())) { @@ -337,6 +353,7 @@ impl Server { }; } + let mut c = self.catalog.lock().unwrap(); if let Err(e) = c.create_table(&table, columns) { return Response::Error { message: format!("create_table {}: {}", table, e), @@ -347,31 +364,27 @@ impl Server { message: format!("create_table {}: catalog save: {}", table, e), }; } - Response::Ok } + Request::DropTable { table } => { log_debug!(pipe, "drop_table table={}", table); - let mut c = match Catalog::load(self.catalog_path()) { - Ok(c) => c, - Err(e) => return Response::Error { - message: format!("drop_table {}: catalog: {}", table, e), - }, - }; - if let Err(e) = c.drop_table(&table) { - return Response::Error { - message: format!("drop_table {}: {}", table, e), - }; - } - if let Err(e) = c.save(self.catalog_path()) { - return Response::Error { - message: format!("drop_table {}: catalog save: {}", table, e), - }; + { + let mut c = self.catalog.lock().unwrap(); + if let Err(e) = c.drop_table(&table) { + return Response::Error { + message: format!("drop_table {}: {}", table, e), + }; + } + if let Err(e) = c.save(self.catalog_path()) { + return Response::Error { + message: format!("drop_table {}: catalog save: {}", table, e), + }; + } } self.tables.lock().unwrap().remove(&table); - let _ = std::fs::remove_file(self.table_path(&table)); let _ = std::fs::remove_file(self.index_path(&table)); @@ -384,12 +397,12 @@ impl Server { /// 表名:catalog 与目录取并集 fn list_tables(&self) -> std::io::Result> { let mut names = BTreeSet::new(); - - let c = Catalog::load(self.catalog_path())?; - for t in c.table_names() { - names.insert(t); + { + let c = self.catalog.lock().unwrap(); + for t in c.table_names() { + names.insert(t); + } } - let dir = &self.cfg.data_dir; if dir.exists() { for entry in std::fs::read_dir(dir)? { @@ -400,19 +413,17 @@ impl Server { } } } - Ok(names.into_iter().collect()) } /// 查一张表的 schema fn describe_table(&self, table: &str) -> std::io::Result<(Vec, u64)> { - let c = Catalog::load(self.catalog_path())?; + let c = self.catalog.lock().unwrap(); if let Some(schema) = c.describe(table) { return Ok((schema.columns.clone(), schema.row_count)); } - if self.tables.lock().unwrap().contains_key(table) - || std::fs::metadata(self.table_path(table)).is_ok() - { + drop(c); + if std::fs::metadata(self.table_path(table)).is_ok() { return Ok((Vec::new(), 0)); } Err(std::io::Error::new( @@ -422,6 +433,14 @@ impl Server { } } +impl Drop for Server { + fn drop(&mut self) { + if let Ok(c) = self.catalog.lock() { + let _ = c.save(self.catalog_path()); + } + } +} + // 连接 /// 处理一条连接 diff --git a/chusql-storage/src/btree.rs b/chusql-storage/src/btree.rs index 5d4e536..a9e06b5 100644 --- a/chusql-storage/src/btree.rs +++ b/chusql-storage/src/btree.rs @@ -250,16 +250,14 @@ impl DiskBTree { /// 读根页号 fn root(&mut self) -> io::Result> { - let meta = self.file.read_page(META_PAGE)?; - let r = read_u64(&meta, OFF_META_ROOT); + let r = self.file.with_page(META_PAGE, |p| read_u64(p, OFF_META_ROOT))?; Ok(if r == 0 { None } else { Some(r) }) } /// 写根页号 fn set_root(&mut self, r: Option) -> io::Result<()> { - let mut meta = self.file.read_page(META_PAGE)?; - write_u64(&mut meta, OFF_META_ROOT, r.unwrap_or(0)); - self.file.write_page(&meta) + self.file + .update_page(META_PAGE, |p| write_u64(p, OFF_META_ROOT, r.unwrap_or(0))) } /// 分配新页 @@ -292,8 +290,8 @@ impl DiskBTree { /// 递归查找 fn get_rec(&mut self, page_id: PageId, key: i64) -> io::Result> { - let p = self.file.read_page(page_id)?; - match read_node(&p, self.order) { + let order = self.order; + match self.file.with_page(page_id, |p| read_node(p, order))? { NodeData::Leaf { keys, values, .. } => { Ok(keys.binary_search(&key).ok().map(|i| values[i])) } @@ -345,9 +343,10 @@ impl DiskBTree { key: i64, value: u64, ) -> io::Result<(PageId, Option<(i64, PageId)>)> { - let p = self.file.read_page(page_id)?; - let limit = max_keys(self.order); - match read_node(&p, self.order) { + let order = self.order; + let node = self.file.with_page(page_id, |p| read_node(p, order))?; + let limit = max_keys(order); + match node { NodeData::Leaf { mut keys, mut values, next } => { match keys.binary_search(&key) { Ok(i) => values[i] = value, @@ -357,9 +356,9 @@ impl DiskBTree { } } if keys.len() <= limit { - let mut p2 = self.new_page(page_id); - write_node(&mut p2, &NodeData::Leaf { keys, values, next }, self.order); - self.file.write_page(&p2)?; + self.file.update_page(page_id, |p| { + write_node(p, &NodeData::Leaf { keys, values, next }, order) + })?; Ok((page_id, None)) } else { let mid = keys.len() / 2; @@ -372,17 +371,13 @@ impl DiskBTree { write_node( &mut rp, &NodeData::Leaf { keys: rkeys, values: rvalues, next }, - self.order, + order, ); self.file.write_page(&rp)?; - let mut lp = self.new_page(page_id); - write_node( - &mut lp, - &NodeData::Leaf { keys, values, next: right_id }, - self.order, - ); - self.file.write_page(&lp)?; + self.file.update_page(page_id, |p| { + write_node(p, &NodeData::Leaf { keys, values, next: right_id }, order) + })?; Ok((page_id, Some((sep, right_id)))) } @@ -394,22 +389,18 @@ impl DiskBTree { children[i] = new_child; match split { None => { - let mut p2 = self.new_page(page_id); - write_node(&mut p2, &NodeData::Internal { keys, children }, self.order); - self.file.write_page(&p2)?; + self.file.update_page(page_id, |p| { + write_node(p, &NodeData::Internal { keys, children }, order) + })?; Ok((page_id, None)) } Some((sep, right_id)) => { keys.insert(i, sep); children.insert(i + 1, right_id); if keys.len() <= limit { - let mut p2 = self.new_page(page_id); - write_node( - &mut p2, - &NodeData::Internal { keys, children }, - self.order, - ); - self.file.write_page(&p2)?; + self.file.update_page(page_id, |p| { + write_node(p, &NodeData::Internal { keys, children }, order) + })?; Ok((page_id, None)) } else { let mid = keys.len() / 2; @@ -423,17 +414,13 @@ impl DiskBTree { write_node( &mut rp, &NodeData::Internal { keys: rkeys, children: rchildren }, - self.order, + order, ); self.file.write_page(&rp)?; - let mut lp = self.new_page(page_id); - write_node( - &mut lp, - &NodeData::Internal { keys, children }, - self.order, - ); - self.file.write_page(&lp)?; + self.file.update_page(page_id, |p| { + write_node(p, &NodeData::Internal { keys, children }, order) + })?; Ok((page_id, Some((up, right_id)))) } @@ -454,8 +441,8 @@ impl DiskBTree { /// 中序遍历 fn walk(&mut self, page_id: PageId, out: &mut Vec<(i64, u64)>) -> io::Result<()> { - let p = self.file.read_page(page_id)?; - match read_node(&p, self.order) { + let order = self.order; + match self.file.with_page(page_id, |p| read_node(p, order))? { NodeData::Leaf { keys, values, .. } => { for i in 0..keys.len() { out.push((keys[i], values[i])); diff --git a/chusql-storage/src/catalog.rs b/chusql-storage/src/catalog.rs index 5188b9a..076a403 100644 --- a/chusql-storage/src/catalog.rs +++ b/chusql-storage/src/catalog.rs @@ -39,6 +39,11 @@ impl Catalog { serde_json::from_slice(&bytes).map_err(io::Error::other) } + /// 返回所有表的 (表名, schema)。 + pub fn all_tables(&self) -> Vec<(&str, &TableSchema)> { + self.tables.iter().map(|(k, v)| (k.as_str(), v)).collect() + } + /// 写回文件 pub fn save>(&self, path: P) -> io::Result<()> { let bytes = serde_json::to_vec_pretty(self).map_err(io::Error::other)?; @@ -98,22 +103,27 @@ impl Catalog { } } /// 记一次插入:补列 + 行数 +1 - pub fn record_insert(&mut self, table: &str, row: &Row) { + pub fn record_insert(&mut self, table: &str, row: &Row) -> bool { let entry = self.tables.entry(table.to_string()).or_default(); + let mut changed = false; for (k, v) in row { if entry.columns.iter().any(|c| &c.name == k) { continue; } if let Some(ty) = infer_type(v) { entry.columns.push(SchemaColumn { name: k.clone(), ty }); + changed = true; } } entry.row_count += 1; + changed } + /// 记一次整表替换:列照补,行数重置 - pub fn record_replace_all(&mut self, table: &str, rows: &[Row]) { + pub fn record_replace_all(&mut self, table: &str, rows: &[Row]) -> bool { let entry = self.tables.entry(table.to_string()).or_default(); + let mut changed = false; for r in rows { for (k, v) in r { if entry.columns.iter().any(|c| &c.name == k) { @@ -121,10 +131,12 @@ impl Catalog { } if let Some(ty) = infer_type(v) { entry.columns.push(SchemaColumn { name: k.clone(), ty }); + changed = true; } } } entry.row_count = rows.len() as u64; + changed } /// 删表 diff --git a/chusql-storage/src/config.rs b/chusql-storage/src/config.rs index 1a3893f..5da95fb 100644 --- a/chusql-storage/src/config.rs +++ b/chusql-storage/src/config.rs @@ -11,7 +11,7 @@ use crate::log::Level; pub const DEFAULT_CONFIG_PATH: &str = "chusql-storage.toml"; /// 默认池容量(页数) -pub const DEFAULT_POOL_SIZE: usize = 64; +pub const DEFAULT_POOL_SIZE: usize = 1024; /// 默认页大小 pub const DEFAULT_PAGE_SIZE: usize = 4096; /// 默认 B+ 树 order diff --git a/chusql-storage/src/heap.rs b/chusql-storage/src/heap.rs index ebb6733..f7cc6df 100644 --- a/chusql-storage/src/heap.rs +++ b/chusql-storage/src/heap.rs @@ -177,10 +177,12 @@ impl HeapTable { let n = self.file.num_pages()?; if n > 0 { let last_id = n - 1; - let mut p = self.file.read_page(last_id)?; - if let Some(slot) = HeapPage::insert_tuple(&mut p, &tuple) { - self.file.write_page(&p)?; - return Ok((last_id, slot)); + let mut slot = None; + self.file.update_page(last_id, |p| { + slot = HeapPage::insert_tuple(p, &tuple); + })?; + if let Some(s) = slot { + return Ok((last_id, s)); } } @@ -215,8 +217,8 @@ impl HeapTable { /// 按位置读一行 pub fn read_at(&mut self, page_id: PageId, slot: u16) -> io::Result> { - let p = self.file.read_page(page_id)?; - match HeapPage::get_tuple(&p, slot) { + let bytes = self.file.with_page(page_id, |p| HeapPage::get_tuple(p, slot))?; + match bytes { None => Ok(None), Some(bytes) => { let row: Row = serde_json::from_slice(&bytes).map_err(io::Error::other)?; @@ -230,13 +232,18 @@ impl HeapTable { let n = self.file.num_pages()?; let mut rows = Vec::new(); for i in 0..n { - let p = self.file.read_page(i)?; - for slot in HeapPage::iter_slots(&p) { - if let Some(bytes) = HeapPage::get_tuple(&p, slot) { - let row: Row = serde_json::from_slice(&bytes) - .map_err(io::Error::other)?; - rows.push(row); + let tuples = self.file.with_page(i, |p| { + let mut out = Vec::new(); + for slot in HeapPage::iter_slots(p) { + if let Some(bytes) = HeapPage::get_tuple(p, slot) { + out.push(bytes); + } } + out + })?; + for bytes in tuples { + let row: Row = serde_json::from_slice(&bytes).map_err(io::Error::other)?; + rows.push(row); } } Ok(rows) diff --git a/chusql-storage/src/page.rs b/chusql-storage/src/page.rs index 763ad3c..f2f51f7 100644 --- a/chusql-storage/src/page.rs +++ b/chusql-storage/src/page.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::fs::{File, OpenOptions}; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::path::Path; @@ -41,7 +41,8 @@ pub struct PageFile { page_size: usize, capacity: usize, cache: HashMap, - order: VecDeque, + last_used: HashMap, + clock: u64, dirty: HashSet, hits: u64, misses: u64, @@ -70,7 +71,8 @@ impl PageFile { page_size, capacity, cache: HashMap::new(), - order: VecDeque::new(), + last_used: HashMap::new(), + clock: 0, dirty: HashSet::new(), hits: 0, misses: 0, @@ -116,26 +118,28 @@ impl PageFile { /// 标记为最近使用 fn touch(&mut self, id: PageId) { - if let Some(pos) = self.order.iter().position(|&x| x == id) { - self.order.remove(pos); - } - self.order.push_back(id); + self.clock += 1; + self.last_used.insert(id, self.clock); } /// 淘汰一个最冷的页 fn evict_one(&mut self) -> io::Result { - while let Some(victim) = self.order.pop_front() { - if let Some(p) = self.cache.get(&victim) { - if self.dirty.contains(&victim) { - let p = p.clone(); - self.write_raw(&p)?; - self.dirty.remove(&victim); - } - self.cache.remove(&victim); - return Ok(true); - } + let victim = self + .cache + .keys() + .min_by_key(|id| self.last_used.get(id).copied().unwrap_or(0)) + .copied(); + let Some(victim) = victim else { + return Ok(false); + }; + if self.dirty.contains(&victim) { + let p = self.cache[&victim].clone(); + self.write_raw(&p)?; + self.dirty.remove(&victim); } - Ok(false) + self.cache.remove(&victim); + self.last_used.remove(&victim); + Ok(true) } /// 直接写文件 @@ -166,20 +170,38 @@ impl PageFile { /// 读一页(走缓存) pub fn read_page(&mut self, id: PageId) -> io::Result { - if let Some(p) = self.cache.get(&id) { - let p = p.clone(); + self.note_access(id)?; + Ok(self.cache[&id].clone()) + } + + /// 只读地用一页(走缓存,不整页拷贝) + pub fn with_page(&mut self, id: PageId, f: impl FnOnce(&Page) -> R) -> io::Result { + self.note_access(id)?; + Ok(f(&self.cache[&id])) + } + + /// 就地改一页(走缓存,不整页拷贝) + pub fn update_page(&mut self, id: PageId, f: impl FnOnce(&mut Page) -> R) -> io::Result { + self.note_access(id)?; + let r = f(self.cache.get_mut(&id).expect("cached above")); + self.dirty.insert(id); + Ok(r) + } + + /// 记一次访问:命中加 hits,未命中读进来并记 misses + fn note_access(&mut self, id: PageId) -> io::Result<()> { + if self.cache.contains_key(&id) { self.hits += 1; - self.touch(id); - return Ok(p); + } else { + self.misses += 1; + let p = self.read_raw(id)?; + if self.cache.len() >= self.capacity { + self.evict_one()?; + } + self.cache.insert(id, p); } - self.misses += 1; - let p = self.read_raw(id)?; - if self.cache.len() >= self.capacity { - self.evict_one()?; - } - self.cache.insert(id, p.clone()); - self.order.push_back(id); - Ok(p) + self.touch(id); + Ok(()) } /// 写一页(走缓存) @@ -257,7 +279,7 @@ impl PageFile { /// 清空文件 pub fn truncate(&mut self) -> io::Result<()> { self.cache.clear(); - self.order.clear(); + self.last_used.clear(); self.dirty.clear(); self.file.set_len(0)?; self.file.seek(SeekFrom::Start(0))?; diff --git a/chusql-storage/src/protocol.rs b/chusql-storage/src/protocol.rs index 44e6bac..7af16eb 100644 --- a/chusql-storage/src/protocol.rs +++ b/chusql-storage/src/protocol.rs @@ -43,6 +43,7 @@ pub enum Request { columns: Vec, }, DropTable { table: String }, + ListCatalog } #[derive(Debug, Serialize, Deserialize)] @@ -59,4 +60,13 @@ pub enum Response { }, Ok, Error { message: String }, + Catalog { schemas: Vec }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TableSchemaWire { + pub table: String, + pub columns: Vec, + #[serde(default)] + pub row_count: u64, } diff --git a/chusql-storage/src/wal.rs b/chusql-storage/src/wal.rs index 5f7251e..2b2afab 100644 --- a/chusql-storage/src/wal.rs +++ b/chusql-storage/src/wal.rs @@ -63,7 +63,7 @@ impl Wal { .append(true) .open(&self.path)?; f.write_all(&bytes)?; - f.sync_all()?; + f.sync_data()?; Ok(()) }