diff --git a/AGENTS.md b/AGENTS.md index bcdb0c5..aaca64b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,5 +28,5 @@ - 改动说明只在交流中记录。 ## 当前重点 -- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测。 +- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测 → M8 事务保存点。 - 进度与计数见 `agent_credientials/project-plan.txt`,限制与风险见 `agent_credientials/known-issues.txt`。 diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine.hs b/chusql-core/engine/src/ChuSQL/Core/Engine.hs index 57a4902..c4d666d 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine.hs @@ -137,6 +137,9 @@ runStatementUncheckedM _ ShowDatabases = runStatementM ShowDatabases runStatementUncheckedM _ BeginTransaction = pure (Left "BEGIN is executed by the session") runStatementUncheckedM _ CommitTransaction = pure (Left "COMMIT is executed by the session") runStatementUncheckedM _ RollbackTransaction = pure (Left "ROLLBACK is executed by the session") +runStatementUncheckedM _ Savepoint{} = pure (Left "SAVEPOINT is executed by the session") +runStatementUncheckedM _ RollbackToSavepoint{} = pure (Left "ROLLBACK TO is executed by the session") +runStatementUncheckedM _ ReleaseSavepoint{} = pure (Left "RELEASE is executed by the session") -- | 只有需要读全表才能判断约束时才扫表 existingRows :: (MonadStorage m) => String -> Table -> m (Either String [Row]) diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Semantic.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Semantic.hs index 7f9fed4..1fef105 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Semantic.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Semantic.hs @@ -437,6 +437,10 @@ checkResolvedWith db outer q = case q of BeginTransaction -> Right () CommitTransaction -> Right () RollbackTransaction -> Right () + -- | 保存点名字只是会话层的记号,不做列检查 + Savepoint _ -> Right () + RollbackToSavepoint _ -> Right () + ReleaseSavepoint _ -> Right () where tableCols' = map fst . tableCols diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/AST.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/AST.hs index fa6f53b..1e5201f 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/AST.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/AST.hs @@ -59,6 +59,12 @@ data Statement | CommitTransaction -- | 回滚事务 | RollbackTransaction + -- | 建保存点:SAVEPOINT 名字 + | Savepoint String + -- | 回滚到保存点:ROLLBACK TO [SAVEPOINT] 名字 + | RollbackToSavepoint String + -- | 释放保存点:RELEASE [SAVEPOINT] 名字 + | ReleaseSavepoint String deriving (Show, Eq) -- | 连接类型:内连接与左外连接 diff --git a/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/Parser.hs b/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/Parser.hs index d028d2e..f6b38fc 100644 --- a/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/Parser.hs +++ b/chusql-core/engine/src/ChuSQL/Core/Engine/Syntax/Parser.hs @@ -619,7 +619,14 @@ transactionStatement = try (BeginTransaction <$ (keyword "start" *> keyword "transaction")) <|> (BeginTransaction <$ (keyword "begin" *> optional (keyword "transaction"))) <|> (CommitTransaction <$ (keyword "commit" *> optional (keyword "transaction"))) + <|> try (RollbackToSavepoint <$> rollbackToName) <|> (RollbackTransaction <$ (keyword "rollback" *> optional (keyword "transaction"))) + <|> (Savepoint <$> (keyword "savepoint" *> identifier)) + <|> (ReleaseSavepoint <$> (keyword "release" *> optional (keyword "savepoint") *> identifier)) + where + -- | ROLLBACK TO [SAVEPOINT] 名字 + rollbackToName = + keyword "rollback" *> keyword "to" *> optional (keyword "savepoint") *> identifier -- | 解析总入口 parseStatement :: String -> Either String Statement diff --git a/chusql-core/engine/test/Spec.hs b/chusql-core/engine/test/Spec.hs index 490fe56..0dff7d5 100644 --- a/chusql-core/engine/test/Spec.hs +++ b/chusql-core/engine/test/Spec.hs @@ -286,6 +286,17 @@ main = hspec $ do parseStatement "COMMIT TRANSACTION" `shouldBe` Right CommitTransaction parseStatement "ROLLBACK" `shouldBe` Right RollbackTransaction parseStatement "rollback transaction" `shouldBe` Right RollbackTransaction + parseStatement "SAVEPOINT spot" `shouldBe` Right (Savepoint "spot") + parseStatement "savepoint spot" `shouldBe` Right (Savepoint "spot") + parseStatement "ROLLBACK TO spot" `shouldBe` Right (RollbackToSavepoint "spot") + parseStatement "ROLLBACK TO SAVEPOINT spot" `shouldBe` Right (RollbackToSavepoint "spot") + parseStatement "RELEASE spot" `shouldBe` Right (ReleaseSavepoint "spot") + parseStatement "RELEASE SAVEPOINT spot" `shouldBe` Right (ReleaseSavepoint "spot") + + it "requires a name on savepoint statements" $ do + parseStatement "SAVEPOINT" `shouldSatisfy` isLeft + parseStatement "ROLLBACK TO" `shouldSatisfy` isLeft + parseStatement "RELEASE" `shouldSatisfy` isLeft it "parses the bare words only as whole words" $ do parseStatement "BEGINNER" `shouldSatisfy` isLeft @@ -297,7 +308,7 @@ main = hspec $ do (runStatement [("users", users)] =<< parseStatement sql) `shouldSatisfy` either (isInfixOf "executed by the session") (const False) ) - ["BEGIN", "COMMIT", "ROLLBACK"] + ["BEGIN", "COMMIT", "ROLLBACK", "SAVEPOINT spot", "ROLLBACK TO spot", "RELEASE spot"] it "puts transaction errors under the protocol category" $ do errorCode "BEGIN is executed by the session" `shouldBe` "query_error" diff --git a/chusql-server/src/ChuSQL/Server/Session.hs b/chusql-server/src/ChuSQL/Server/Session.hs index 99ce9a3..2079929 100644 --- a/chusql-server/src/ChuSQL/Server/Session.hs +++ b/chusql-server/src/ChuSQL/Server/Session.hs @@ -69,10 +69,11 @@ data Session = Session , ssTransaction :: IORef (Maybe Transaction) } --- | 一条会话里的显式事务:开事务时的快照与事务内改到的状态 +-- | 一条会话里的显式事务:开事务时的快照、事务内改到的状态与保存点栈 data Transaction = Transaction { txBase :: Database , txStaged :: Database + , txSavepoints :: [(String, Database)] } -- | 开一个会话:账号与角色服务、配置策略都和 Web 端一致 @@ -191,6 +192,9 @@ runStatementCoded session sql = case parseStatement (T.unpack sql) of Right BeginTransaction -> beginTransaction session Right CommitTransaction -> commitTransaction session Right RollbackTransaction -> rollbackTransaction session + Right (Savepoint name) -> savepointTransaction session (T.pack name) + Right (RollbackToSavepoint name) -> rollbackToSavepoint session (T.pack name) + Right (ReleaseSavepoint name) -> releaseSavepoint session (T.pack name) Right statement | Just command <- accountCommand statement -> runAccount session command | Just command <- privilegeCommand statement -> runPrivilege session command @@ -249,7 +253,7 @@ beginTransaction session = do case snapshotResult of Left err -> pure (Left (engineErrorCode err)) Right db -> do - writeIORef (ssTransaction session) (Just (Transaction db db)) + writeIORef (ssTransaction session) (Just (Transaction db db [])) pure (Right (emptyResult (Just current))) -- | 提交:把快照差分交给存储层应用,成功才结束事务 @@ -285,6 +289,54 @@ rollbackTransaction session = do writeIORef (ssTransaction session) Nothing pure (Right (emptyResult (Just current))) +-- | 建保存点:把事务当前状态压进保存点栈 +savepointTransaction :: Session -> Text -> IO (Either SessionError QueryResult) +savepointTransaction session name = do + current <- readIORef (ssCurrent session) + existing <- readIORef (ssTransaction session) + case existing of + Nothing -> pure (Left (SessionError "query_error" "no transaction in progress")) + Just transaction -> do + let stack = txSavepoints transaction ++ [(T.unpack name, txStaged transaction)] + writeIORef (ssTransaction session) (Just transaction {txSavepoints = stack}) + pure (Right (emptyResult (Just current))) + +-- | 回滚到保存点:事务状态退回保存点,该保存点保留 +rollbackToSavepoint :: Session -> Text -> IO (Either SessionError QueryResult) +rollbackToSavepoint session name = do + current <- readIORef (ssCurrent session) + existing <- readIORef (ssTransaction session) + case existing of + Nothing -> pure (Left (SessionError "query_error" "no transaction in progress")) + Just transaction -> case savepointIndex name (txSavepoints transaction) of + Nothing -> pure (Left (SessionError "query_error" ("no such savepoint: " <> name))) + Just index -> do + let (_, staged) = txSavepoints transaction !! index + writeIORef + (ssTransaction session) + (Just transaction {txStaged = staged, txSavepoints = take (index + 1) (txSavepoints transaction)}) + pure (Right (emptyResult (Just current))) + +-- | 释放保存点:该保存点与它之后的保存点一起丢掉 +releaseSavepoint :: Session -> Text -> IO (Either SessionError QueryResult) +releaseSavepoint session name = do + current <- readIORef (ssCurrent session) + existing <- readIORef (ssTransaction session) + case existing of + Nothing -> pure (Left (SessionError "query_error" "no transaction in progress")) + Just transaction -> case savepointIndex name (txSavepoints transaction) of + Nothing -> pure (Left (SessionError "query_error" ("no such savepoint: " <> name))) + Just index -> do + writeIORef (ssTransaction session) (Just transaction {txSavepoints = take index (txSavepoints transaction)}) + pure (Right (emptyResult (Just current))) + +-- | 最近一次同名保存点的下标 +savepointIndex :: Text -> [(String, Database)] -> Maybe Int +savepointIndex name entries = + case [index | (index, (label, _)) <- zip [0 ..] entries, T.pack label == name] of + [] -> Nothing + hits -> Just (last hits) + -- | 事务里跑一条数据语句:只动内存快照,不碰存储层 runInTransaction :: Session -> Statement -> IO (Either SessionError QueryResult) runInTransaction session statement = do diff --git a/chusql-server/test/Spec.hs b/chusql-server/test/Spec.hs index b3c21a6..dd72e73 100644 --- a/chusql-server/test/Spec.hs +++ b/chusql-server/test/Spec.hs @@ -788,6 +788,66 @@ transactionSpec = describe "server transactions" $ do ids <- sessionIds second ids `shouldMatchList` [1, 2, 3, 5] + it "keeps the work before a savepoint and drops the work after it" $ withMemorySession $ \session -> do + _ <- mustSql session "begin" + _ <- mustSql session "update users set age = 31 where id = 1" + _ <- mustSql session "savepoint spot" + _ <- mustSql session "insert into users (id, name, age) values (6, 'six', 26)" + _ <- mustSql session "delete from users where id = 2" + _ <- mustSql session "rollback to spot" + inside <- sessionIds session + inside `shouldMatchList` [1, 2, 3, 4, 5] + sessionAge session 1 `shouldReturn` Just 31 + _ <- mustSql session "commit" + ids <- sessionIds session + ids `shouldMatchList` [1, 2, 3, 4, 5] + sessionAge session 1 `shouldReturn` Just 31 + + it "lets a savepoint be used again after rolling back to it" $ withMemorySession $ \session -> do + _ <- mustSql session "begin" + _ <- mustSql session "insert into users (id, name, age) values (6, 'six', 26)" + _ <- mustSql session "savepoint spot" + _ <- mustSql session "insert into users (id, name, age) values (7, 'seven', 27)" + _ <- mustSql session "rollback to spot" + _ <- mustSql session "insert into users (id, name, age) values (8, 'eight', 28)" + ids <- sessionIds session + ids `shouldMatchList` [1, 2, 3, 4, 5, 6, 8] + + it "drops the savepoints above the one rolled back to" $ withMemorySession $ \session -> do + _ <- mustSql session "begin" + _ <- mustSql session "savepoint one" + _ <- mustSql session "savepoint two" + _ <- mustSql session "rollback to one" + missing <- runStatementCoded session "rollback to two" + failureMessage missing `shouldBe` "no such savepoint: two" + + it "drops a released savepoint and the ones after it" $ withMemorySession $ \session -> do + _ <- mustSql session "begin" + _ <- mustSql session "savepoint one" + _ <- mustSql session "savepoint two" + released <- mustSql session "release one" + qrRowCount released `shouldBe` 0 + first <- runStatementCoded session "rollback to one" + failureMessage first `shouldBe` "no such savepoint: one" + second <- runStatementCoded session "rollback to two" + failureMessage second `shouldBe` "no such savepoint: two" + + it "clears the savepoints on COMMIT" $ withMemorySession $ \session -> do + _ <- mustSql session "begin" + _ <- mustSql session "savepoint spot" + _ <- mustSql session "commit" + missing <- runStatementCoded session "rollback to spot" + failureMessage missing `shouldBe` "no transaction in progress" + + it "needs a transaction and a known name for savepoints" $ withMemorySession $ \session -> do + outside <- runStatementCoded session "savepoint spot" + failureMessage outside `shouldBe` "no transaction in progress" + _ <- mustSql session "begin" + unknown <- runStatementCoded session "rollback to nowhere" + failureMessage unknown `shouldBe` "no such savepoint: nowhere" + gone <- runStatementCoded session "release nowhere" + failureMessage gone `shouldBe` "no such savepoint: nowhere" + it "needs a current database to start" $ do db <- newMVar testDb settings <- tempSettingsPath "transaction-bare" @@ -882,6 +942,26 @@ ipcConcurrencySpec = describe "server backend on the real storage" $ do rows <- rowsOf (beStatement alpha "SELECT id FROM t") intColumn "id" rows `shouldBe` [1] + it "rolls back to a savepoint on the real storage" $ withIpcStorage $ do + base <- ipcBackend + mustRun (beStatement base "CREATE DATABASE alpha") + let alpha = beWithDatabase base "alpha" + settings <- tempSettingsPath "transaction-ipc-savepoint" + removeIfExists settings + session <- newSession alpha (T.pack testDatabaseName) settings + _ <- mustSql session "use alpha" + _ <- mustSql session "create table t (id int)" + _ <- mustSql session "begin" + _ <- mustSql session "insert into t (id) values (1)" + _ <- mustSql session "savepoint spot" + _ <- mustSql session "insert into t (id) values (2)" + _ <- mustSql session "rollback to spot" + inside <- mustSql session "select * from t" + intValues (qrRows inside) `shouldBe` [1] + _ <- mustSql session "commit" + rows <- rowsOf (beStatement alpha "SELECT id FROM t") + intColumn "id" rows `shouldBe` [1] + it "refuses the second COMMIT on the real storage" $ withIpcStorage $ do base <- ipcBackend mustRun (beStatement base "CREATE DATABASE alpha") diff --git a/docs/architecture.md b/docs/architecture.md index 4bbdec2..727e8e0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -49,7 +49,7 @@ chusql-server 语句里的库前缀和表限定符由引擎在语义层解析:存储层只接受精确的表名键,因此 `FROM sales.orders` 与选定 `sales` 后的 `FROM orders` 指向同一张表。 -显式事务在会话层实现:`BEGIN` 把当前数据库整库读成内存快照,事务内的语句都在快照上执行;`COMMIT` 先把快照、基线与库里现值按行 `id` 逐键比对,检测到写写冲突就报 `serialization_failure` 并保留事务;没有冲突才把差分用一条 `apply_transaction` 请求交给存储层应用;`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交:事务里看不到别的事务未提交或后来提交的改动,提交对外一次性可见,冲突由调用方回滚或重试。 +显式事务在会话层实现:`BEGIN` 把当前数据库整库读成内存快照,事务内的语句都在快照上执行;`SAVEPOINT` 把当前快照压进会话里的保存点栈,`ROLLBACK TO` 退回该点并保留它,`RELEASE` 丢掉该点与其后的保存点;`COMMIT` 先把快照、基线与库里现值按行 `id` 逐键比对,检测到写写冲突就报 `serialization_failure` 并保留事务;没有冲突才把差分用一条 `apply_transaction` 请求交给存储层应用;`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交:事务里看不到别的事务未提交或后来提交的改动,提交对外一次性可见,冲突由调用方回滚或重试。 WAL 负责崩溃恢复:每条记录带自增 LSN,一组写操作与一条提交标记一次落盘;每组提交在数据和字典都刷盘之后把 `wal.checkpoint` 推进到这条提交的 LSN,含义是「到此为止的改动都已落盘」,再按这个边界丢掉日志前缀——所以日志里通常只有当组在飞的记录。进程重启时只重放检查点之后的记录,按标记分组取已提交的组、丢弃没有标记的整组,重放前逐条做幂等检查作为第二道保险,结束后再把检查点推到最后一条提交。LSN 从检查点与文件尾里较大的那个续号,跨重启单调,`SystemStatus` 应答里的 `last_lsn` 报告该检查点 LSN。建表、删表、建索引、删索引与数据改动走同一条 WAL,恢复时一并重放。 diff --git a/docs/commands.md b/docs/commands.md index b78a175..9a4b45e 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -48,10 +48,12 @@ csql -f json -e "SELECT count(*) FROM orders" 事务: - `BEGIN`(或 `START TRANSACTION`)开始显式事务,`COMMIT` 提交,`ROLLBACK` 回滚。 +- `SAVEPOINT 名字` 在事务里记一个回滚点,`ROLLBACK TO 名字`(可写 `ROLLBACK TO SAVEPOINT 名字`)退回该点但保留它,`RELEASE 名字` 丢掉该点与它之后的回滚点。 - 事务内只允许数据语句;DDL 与 `USE` 会被拒绝,`SHOW DATABASES` 这类不碰表数据的语句照常可用。 - 事务里的读都看开事务那一刻的快照:别的连接在事务期间提交的改动不会出现在事务里,本会话未提交的改动别的连接也看不见。 -- 同一连接同时只能有一个事务,没有保存点与嵌套事务;开事务前要先选好数据库。 +- 同一连接同时只能有一个事务,没有嵌套事务;回滚点只在事务内有效,事务结束就没了。开事务前要先选好数据库。 - 提交时按行 `id` 写回本事务改动过的行,事务没碰过的行不受影响;没有整数 `id` 的表按整表替换提交。 +- 提交前会比对基线:本事务改过的行如果已被别的连接改掉或删掉,提交报 `serialization_failure` 并保留事务,由调用方 `ROLLBACK` 或重试。 - 提交是持久的:整批改动与一条提交标记一次落盘,提交后进程被强杀也会在重启时重放;提交前崩溃则整批丢弃,不会只落一半。 ### 交互模式