v26.10.0-pre

This commit is contained in:
LIm1tZz7 committed 2026-10-04 14:54:17 +08:00
1 parent 59027684e6
commit fd5f320360
13 files changed
+149 -68

No files matched your search

+1 -1
View File
@@ -30,5 +30,5 @@
- 改动说明只在交流中记录。
## 当前重点
- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测 → M8 事务保存点 → M9 角色继承 → M10 转授权 → M11 FROM 派生表与投影别名 → M12 `WITH` / CTE → M13 比较运算符补全 → M14 ANY / ALL 量词比较 → M15 优化器直方图 → M16 子查询去相关 → M17 页内完整 compact。
- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断 → M7 事务写写冲突检测 → M8 事务保存点 → M9 角色继承 → M10 转授权 → M11 FROM 派生表与投影别名 → M12 `WITH` / CTE → M13 比较运算符补全 → M14 ANY / ALL 量词比较 → M15 优化器直方图 → M16 子查询去相关 → M17 页内完整 compact → 权限变更断开既有会话 → M18 发布链路真机验证。
- 进度与计数见 `agent_credientials/project-plan.txt`,限制与风险见 `agent_credientials/known-issues.txt`。
@@ -34,6 +34,8 @@ estimateRows h lo hi = max 0 (min total (round (sum (zipWith overlap [0 ..] (his
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)
-- | 第 i 个桶与区间重叠的行数
overlap :: Int -> Int -> Double
overlap i count
| width <= 0 = if lower <= histLow h && histLow h <= upper then fromIntegral count else 0
| otherwise = fromIntegral count * hit / width
@@ -6,7 +6,6 @@ import ChuSQL.Core.Engine.Syntax.AST
import ChuSQL.Core.Engine.Syntax.Parser (parseExpression)
import Data.Char (isAlphaNum, isAscii, toLower)
import Data.List (intercalate, isInfixOf, isPrefixOf, nub)
import Data.Maybe (fromMaybe)
-- 语义检查:表和列在不在、类型对不对,全部在执行前查。
@@ -766,7 +765,7 @@ resolveFrom db _ (FromSubquery mAlias stmt) = do
inner <- resolveSubqueryStatement db [] stmt
named <- nameDerivedColumns inner
Right (FromSubquery mAlias named)
resolveFrom db outer fromC@(FromJoin kind left right cond) = do
resolveFrom db outer (FromJoin kind left right cond) = do
l <- resolveFrom db outer left
r <- resolveFrom db outer right
env <- checkFrom db outer (FromJoin kind l r cond)
@@ -693,17 +693,17 @@ checkCteArity name cols out
-- | 把语句里对 CTE 的引用换成派生表
expandStatement :: [(String, Statement)] -> Statement -> Either String Statement
expandStatement env q = case q of
expandStatement env stmt = case stmt of
q@Select { selectFrom = fromC, selectWhere = mWhere } -> do
fromC' <- expandClause env fromC
w <- traverse (expandExpr env) mWhere
Right q { selectFrom = fromC', selectWhere = w }
q@SelectExpr { selectFrom = fromC, selectItems = items, selectWhere = mWhere } -> do
fromC' <- expandClause env fromC
items' <- traverse (\(label, e) -> fmap ((,) label) (expandExpr env e)) items
items' <- traverse (\(name, e) -> fmap ((,) name) (expandExpr env e)) items
w <- traverse (expandExpr env) mWhere
Right q { selectFrom = fromC', selectItems = items', selectWhere = w }
_ -> Right q
_ -> Right stmt
-- | 来源树里的 CTE 引用换成派生表
expandClause :: [(String, Statement)] -> FromClause -> Either String FromClause
+21 -26
View File
@@ -182,6 +182,11 @@ impl HeapPage {
/// 读槽位里的行字节
pub fn get_tuple(page: &Page, slot: u16) -> Option<Vec<u8>> {
Self::tuple_bytes(page, slot).map(<[u8]>::to_vec)
}
/// 借用槽位里的行字节
fn tuple_bytes(page: &Page, slot: u16) -> Option<&[u8]> {
let n = Self::slot_count(page);
if slot >= n {
return None;
@@ -192,7 +197,7 @@ impl HeapPage {
if len == 0 {
return None;
}
Some(page.data[off..off + len].to_vec())
Some(&page.data[off..off + len])
}
/// 所有槽位号
@@ -593,14 +598,12 @@ impl HeapTable {
/// 按位置读一行
pub fn read_at(&mut self, page_id: PageId, slot: u16) -> io::Result<Option<Row>> {
let bytes = self.file.with_page(page_id, |p| HeapPage::get_tuple(p, slot))?;
match bytes {
None => Ok(None),
Some(bytes) => {
let row = decode_row(&bytes, None, &self.hidden_columns)?;
Ok(Some(row))
}
}
let hidden = &self.hidden_columns;
self.file.with_page(page_id, |p| {
HeapPage::tuple_bytes(p, slot)
.map(|bytes| decode_row(bytes, None, hidden))
.transpose()
})?
}
/// 按位置删一行并返回它
@@ -631,13 +634,9 @@ impl HeapTable {
pub fn delete_by_keys(&mut self, keys: &[i64]) -> io::Result<Vec<Row>> {
let mut positions: Vec<(PageId, u16)> = Vec::new();
if self.has_index("id") {
if let Some(idx) = self.indexes.iter_mut().find(|i| i.column == "id") {
for &k in keys {
let pos = self
.indexes
.iter_mut()
.find(|i| i.column == "id")
.and_then(|i| i.tree.get(&key_of(k)).ok().flatten());
let pos = idx.tree.get(&key_of(k))?;
if let Some(p) = pos {
positions.push(unpack_position(p));
}
@@ -714,20 +713,16 @@ impl HeapTable {
) -> io::Result<Vec<(PageId, u16, Row)>> {
let n = self.file.num_pages()?;
let mut out = Vec::new();
let hidden = &self.hidden_columns;
for i in from..to.min(n) {
let tuples = self.file.with_page(i, |p| {
let mut v = Vec::new();
for slot in HeapPage::iter_slots(p) {
if let Some(bytes) = HeapPage::get_tuple(p, slot) {
v.push((slot, bytes));
self.file.with_page(i, |p| -> io::Result<()> {
for slot in 0..HeapPage::slot_count(p) {
if let Some(bytes) = HeapPage::tuple_bytes(p, slot) {
out.push((i, slot, decode_row(bytes, columns, hidden)?));
}
}
v
})?;
for (slot, bytes) in tuples {
let row = decode_row(&bytes, columns, &self.hidden_columns)?;
out.push((i, slot, row));
}
Ok(())
})??;
}
Ok(out)
}
+56
View File
@@ -15,6 +15,62 @@ fn row(pairs: &[(&str, serde_json::Value)]) -> Row {
m
}
/// 索引读失败时删除报错且保留数据
#[test]
fn delete_propagates_index_read_errors() {
let dir = tempfile::tempdir().unwrap();
let data = dir.path().join("t.db");
let index = dir.path().join("t.idx");
let mut t = HeapTable::open_indexed(&data, &index, DEFAULT_PAGE_SIZE, 4, 1).unwrap();
let original = row(&[("id", json!(1))]);
t.insert_row(&original).unwrap();
t.flush().unwrap();
let file = std::fs::OpenOptions::new().write(true).open(&index).unwrap();
file.set_len(DEFAULT_PAGE_SIZE as u64).unwrap();
let err = t.delete_by_keys(&[1]).unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
assert_eq!(t.scan().unwrap(), vec![original]);
}
/// 跨页投影与分片保持行序及隐藏列
#[test]
fn projected_shards_preserve_order_and_hidden_columns() {
let dir = tempfile::tempdir().unwrap();
let mut t = HeapTable::open(dir.path().join("t.db"), 256, 1).unwrap();
for i in 0..40 {
t.insert_row(&row(&[("id", json!(i)), ("name", json!("alice")), ("secret", json!(i))])).unwrap();
}
t.set_hidden_columns(["secret".to_string()].into_iter().collect());
let columns = vec!["id".to_string(), "secret".to_string()];
let expected: Vec<Row> = (0..40).map(|i| row(&[("id", json!(i))])).collect();
assert_eq!(t.scan_columns(Some(&columns)).unwrap(), expected);
let pages = t.num_pages().unwrap();
assert!(pages > 1);
let mut shards = t.scan_columns_pages(Some(&columns), 0, pages / 2).unwrap();
shards.extend(t.scan_columns_pages(Some(&columns), pages / 2, pages + 1).unwrap());
assert_eq!(shards, expected);
assert_eq!(t.read_at(0, 0).unwrap().unwrap(), row(&[("id", json!(0)), ("name", json!("alice"))]));
}
/// 扫描与点读均返回行解码错误
#[test]
fn malformed_rows_return_decode_errors() {
use chusql_core_storage::heap::HeapPage;
use chusql_core_storage::page::{Page, PageFile};
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("t.db");
let mut file = PageFile::open(&path, DEFAULT_PAGE_SIZE).unwrap();
let mut page = Page::new(0, DEFAULT_PAGE_SIZE);
HeapPage::insert_tuple(&mut page, b"{\"id\":1} trailing").unwrap();
file.write_page(&page).unwrap();
file.flush().unwrap();
drop(file);
let mut t = HeapTable::open(&path, DEFAULT_PAGE_SIZE, 1).unwrap();
assert!(t.scan().is_err());
assert!(t.scan_columns(Some(&[])).is_err());
assert!(t.read_at(0, 0).is_err());
}
/// 索引文件被清空(强杀或截断)后打开,会按堆数据补齐
#[test]
fn empty_index_file_is_rebuilt_from_rows() {
+32 -35
View File
@@ -49,8 +49,6 @@ import Network.Socket (
getSocketName,
listen,
setSocketOption,
shutdown,
ShutdownCmd (ShutdownBoth),
socket,
socketToHandle,
)
@@ -61,21 +59,31 @@ import System.IO.Error (isEOFError)
-- 在线连接与服务器环境
-- | 在线连接登记:账号名加一个「关掉这条连接」的动作
-- | 一条在线连接:登录账号与「已被断开」标记
data LiveEntry = LiveEntry
{ leWho :: IORef (Maybe Text)
, leDropped :: IORef Bool
}
-- | 在线连接登记:自增编号到连接条目
data LiveConnections = LiveConnections
{ lcNext :: IORef Int
, lcLive :: MVar (Map.Map Int (IORef (Maybe Text), IO ()))
, lcLive :: MVar (Map.Map Int LiveEntry)
}
-- | 建一张空的在线连接登记表
newLiveConnections :: IO LiveConnections
newLiveConnections = LiveConnections <$> newIORef 0 <*> newMVar Map.empty
-- | 新建一条连接条目
newLiveEntry :: IO LiveEntry
newLiveEntry = LiveEntry <$> newIORef Nothing <*> newIORef False
-- | 登记一条连接,返回注销动作
registerConnection :: LiveConnections -> IORef (Maybe Text) -> IO () -> IO (IO ())
registerConnection live who kick = do
registerConnection :: LiveConnections -> LiveEntry -> IO (IO ())
registerConnection live entry = do
n <- atomicModifyIORef' (lcNext live) (\i -> (i + 1, i))
modifyMVar_ (lcLive live) (pure . Map.insert n (who, kick))
modifyMVar_ (lcLive live) (pure . Map.insert n entry)
pure (modifyMVar_ (lcLive live) (pure . Map.delete n))
-- | 断开这些账号的在线连接,返回断开的条数
@@ -84,15 +92,15 @@ dropConnectionsOf live names = do
let wanted = map T.toLower (filter (not . T.null) names)
entries <- readMVar (lcLive live)
hits <- filterM (isOf wanted) (Map.elems entries)
mapM_ (tryKick . snd) hits
mapM_ markDropped hits
pure (length hits)
where
-- | 这条连接登录的账号在名单里吗
isOf wanted (who, _) = do
current <- readIORef who
isOf wanted entry = do
current <- readIORef (leWho entry)
pure (maybe False (\name -> T.toLower name `elem` wanted) current)
-- | 踢人失败不当错误:对端可能已经自己走了
tryKick kick = void (try kick :: IO (Either IOException ()))
-- | 只做标记:连接在下一次请求应答后收工
markDropped entry = writeIORef (leDropped entry) True
-- 服务器
@@ -191,48 +199,37 @@ reportConnection env conn = do
Left err -> hPutStrLn stderr ("connection failed: " ++ show err)
Right () -> pure ()
-- | 一条连接上跨请求的状态:登录账号、已被断开与注销动作
data Connection = Connection
{ cnWho :: IORef (Maybe Text)
, cnDropped :: IORef Bool
}
-- | 一个连接一个会话:一行一条 JSON,读到 EOF 就收工
serveConnection :: ServerEnv -> Socket -> IO ()
serveConnection env conn = do
h <- socketToHandle conn ReadWriteMode
hSetBuffering h LineBuffering
hSetNewlineMode h noNewlineTranslation
who <- newIORef Nothing
dropped <- newIORef False
let kick = do
writeIORef dropped True
void (try (shutdown conn ShutdownBoth) :: IO (Either IOException ()))
unregister <- registerConnection (srvLive env) who kick
let connection = Connection who dropped
entry <- newLiveEntry
unregister <- registerConnection (srvLive env) entry
session <- newSessionWith (srvBackend env) (srvRootName env) (srvSettingsFile env) (\names -> void (dropConnectionsOf (srvLive env) names))
outcome <- try (loop h session connection `finally` unregister) :: IO (Either IOException ())
outcome <- try (loop h session entry `finally` unregister) :: IO (Either IOException ())
case outcome of
Left err | not (isEOFError err) -> hPutStrLn stderr ("connection error: " ++ show err)
_ -> pure ()
void (try (hClose h) :: IO (Either IOException ()))
where
-- | 逐行读请求并应答
loop h session connection = do
loop h session entry = do
line <- BSC.hGetLine h
let cleaned = BSC.dropWhileEnd (== '\r') line
if BSC.length cleaned > scMaxMessage (srvConfig env)
then send h (RespError "too_large" "request line too large") >> loop h session connection
then send h (RespError "too_large" "request line too large") >> loop h session entry
else case decodeRequest cleaned of
Left message -> send h (RespError "bad_request" message) >> loop h session connection
Left message -> send h (RespError "bad_request" message) >> loop h session entry
Right request -> do
expired <- readIORef (cnDropped connection)
expired <- readIORef (leDropped entry)
if expired
then send h (RespError "unauthorized" "session was dropped by an account change")
else do
(response, keepGoing) <- dispatch env session connection request
(response, keepGoing) <- dispatch env session entry request
send h response
if keepGoing then loop h session connection else pure ()
if keepGoing then loop h session entry else pure ()
-- | 发一条响应并冲缓冲
send h response = do
BLC.hPutStr h (encodeResponse response)
@@ -240,15 +237,15 @@ serveConnection env conn = do
hFlush h
-- | 一条请求一条响应,quit 后停连接
dispatch :: ServerEnv -> Session -> Connection -> ClientRequest -> IO (ServerResponse, Bool)
dispatch env session connection request = case request of
dispatch :: ServerEnv -> Session -> LiveEntry -> ClientRequest -> IO (ServerResponse, Bool)
dispatch env session entry request = case request of
ReqHello version
| version == protocolVersion -> pure (RespHello protocolVersion, True)
| otherwise -> pure (RespError "bad_request" ("unsupported protocol version: " <> tshow version), True)
ReqLogin user password -> do
response <- login env session user password
case response of
RespLogin who _ -> writeIORef (cnWho connection) (Just who)
RespLogin who _ -> writeIORef (leWho entry) (Just who)
_ -> pure ()
pure (response, True)
ReqQuery sql -> do
+11 -1
View File
@@ -497,6 +497,8 @@ tcpServerSpec = do
allowed <- tcpQuery bob "select name from users"
asText (at "status" allowed) `shouldBe` "result"
_ <- tcpQuery admin "revoke reader from bob"
refused <- tcpQuery bob "select name from users"
asText (at "code" refused) `shouldBe` "unauthorized"
tcpClosed bob
it "disconnects every member of a role that was dropped" $
withTcpServer defaultServerConfig $ \_ handle ->
@@ -508,6 +510,8 @@ tcpServerSpec = do
withConnection handle $ \bob -> do
_ <- tcpLogin bob "bob" "bob-password-1"
_ <- tcpQuery admin "drop role reader"
refused <- tcpQuery bob "use test"
asText (at "code" refused) `shouldBe` "unauthorized"
tcpClosed bob
it "disconnects an account whose password was reset" $
withTcpServer defaultServerConfig $ \_ handle ->
@@ -517,6 +521,8 @@ tcpServerSpec = do
withConnection handle $ \bob -> do
_ <- tcpLogin bob "bob" "bob-password-1"
_ <- tcpQuery admin "alter user bob identified by 'bob-password-2'"
refused <- tcpQuery bob "use test"
asText (at "code" refused) `shouldBe` "unauthorized"
tcpClosed bob
withConnection handle $ \again -> do
relogin <- tcpLogin again "bob" "bob-password-2"
@@ -529,6 +535,8 @@ tcpServerSpec = do
withConnection handle $ \bob -> do
_ <- tcpLogin bob "bob" "bob-password-1"
_ <- tcpQuery admin "drop user bob"
refused <- tcpQuery bob "use test"
asText (at "code" refused) `shouldBe` "unauthorized"
tcpClosed bob
it "leaves the connections of untouched accounts alone" $
withTcpServer defaultServerConfig $ \_ handle ->
@@ -763,7 +771,9 @@ privilegeSpec = describe "server privileges" $ do
_ <- runPrivilegeCommand service root testDatabase (GrantRoleCommand "inner" ["reader"])
_ <- runPrivilegeCommand service root testDatabase (GrantRoleCommand "reader" ["alice"])
directly <- affectedAccounts service (RevokeRoleCommand "inner" ["bob"])
directly `shouldMatchList` ["inner", "bob", "reader", "alice"]
directly `shouldMatchList` ["bob"]
granted <- affectedAccounts service (GrantRoleCommand "writer" ["inner"])
granted `shouldMatchList` ["inner", "reader", "alice"]
dropped <- affectedAccounts service (DropRoleCommand "inner")
dropped `shouldMatchList` ["inner", "reader", "alice"]
affectedAccounts service (CreateRoleCommand "other") `shouldReturn` []
+4
View File
@@ -65,4 +65,8 @@ WAL 负责崩溃恢复:每条记录带自增 LSN,一组写操作与一条提
系统数据库及部分管理操作仅允许管理员访问。系统表统一用 `__system_` 前缀:账号表 `__system_users` 对语句通道隐藏,角色、授权、带 option 的授权、成员四张表(`__system_roles` / `__system_grants` / `__system_grant_options` / `__system_members`)只在 `system` 库里可见,权限服务自身也通过 SQL 读写它们;存储启动时把旧名 `sys_roles` / `sys_grants` / `sys_members` 连同数据与索引文件一起迁移过来。角色支持互相继承:`GRANT 角色 A TO 角色 B` 记一条 B 属于 A 的成员关系,判定权限时沿成员关系算出向上闭包再取并集;写入时拒绝成环与自继承,删角色会清掉它两个方向的成员关系。带 `WITH GRANT OPTION` 的授权记在第四张表 `__system_grant_options` 里(同样的角色、权限、对象三列,另开一张表省掉给 `__system_grants` 加列带来的系统表结构迁移):普通身份只要手里那条授权带 grant option 就能把同一项权限转授给别的角色,`REVOKE` 仍然只允许管理员,收回授权时连它的 grant option 一起收掉,但不级联收回别人已经转授出去的授权。授权命令由权限服务在会话层之外直接执行,`GRANT` 命令会先按当前库把对象补全成 `库.表` 再落盘,因此判断转授权资格与实际写入用的是同一个对象名。
账号与权限改动会断开受影响账号的在用连接。TCP 服务维护一张在线连接表(自增编号对应一条连接条目:登录账号与「已被断开」标记),会话在执行完 `CREATE` / `ALTER` / `DROP USER` 以及 `CREATE` / `DROP ROLE`、`GRANT` / `REVOKE` 之后算出受影响账号——角色类命令先沿成员关系把角色展开成具体账号名(删角色必须在命令落地前算,落地后成员关系就没了)——再给这些账号的连接打上断开标记。连接线程读到请求后先看标记,命中就回 `unauthorized`(`session was dropped by an account change`)并收工,因此空闲连接在客户端下一次请求时结束,不强行从别的线程中断阻塞读取:Windows 上 GHC 经 `socketToHandle` 得到的句柄,`hGetLine` / `hIsEOF` 阻塞期间既收不到 `throwTo` 也不受 `timeout` 约束,标记法不依赖异步异常,全平台行为一致。权限判定本身不缓存、每条语句现查,所以这里补的是「已登录的长连接不能继续用旧身份」,与不相关账号的连接无关。Web 侧走令牌,令牌登录时记下账号版本,改口令或删账号会递增版本,旧令牌因此自动失效。
服务监听地址、端口和请求限制可在 `chusql.toml` 的 `[server]` 分区中配置。
+5
View File
@@ -38,6 +38,11 @@ csql -d sales -e "SELECT * FROM orders LIMIT 5"
csql -f json -e "SELECT count(*) FROM orders"
```
在脚本里非交互使用(配合 `-e`):
- 口令从标准输入读取,行尾用 LF(`\n`)或者干脆不写行尾;CRLF 会把 `\r` 算进口令,结果是 `unauthorized`。
- `root's password: ` 这个提示写在标准输出里,解析输出时要先去掉它;错误信息写在标准错误,失败时退出码为 `1`。
表名与列名的写法:
- 选定数据库后语句里直接写表名即可,例如 `SELECT name FROM orders WHERE orders.id = 1`。
+4
View File
@@ -8,6 +8,10 @@ param(
[string[]]$Only
)
$ErrorActionPreference = 'Stop'
# 车道里的 GHC / stack 诊断带非 ASCII 字符,父子进程统一按 UTF-8 收
$env:GHC_CHARENC = 'UTF-8'
try { [Console]::OutputEncoding = [System.Text.Encoding]::UTF8 } catch { }
$OutputEncoding = [System.Text.Encoding]::UTF8
$LogDir = Join-Path $PSScriptRoot 'logs'
$TmpDir = Join-Path $PSScriptRoot 'tmp'
+4
View File
@@ -5,6 +5,10 @@ param(
[switch]$NoLint
)
$ErrorActionPreference = 'Continue'
# GHC / stack 的诊断带圆点等非 ASCII 字符,输出统一按 UTF-8,免得写管道时炸编码
$env:GHC_CHARENC = 'UTF-8'
try { [Console]::OutputEncoding = [System.Text.Encoding]::UTF8 } catch { }
$OutputEncoding = [System.Text.Encoding]::UTF8
$codes = @()
# 跑一个步骤:切目录、输出原样透传、记录退出码
+5
View File
@@ -11,6 +11,11 @@ $ErrorActionPreference = 'Stop'
# Windows 打包脚本:构建、装配、打 zip、出 sha256,产物落 releases\,命名与 CI 一致。
# stack / GHC 的诊断带圆点等非 ASCII 字符,输出统一按 UTF-8,免得写管道时炸编码
$env:GHC_CHARENC = 'UTF-8'
try { [Console]::OutputEncoding = [System.Text.Encoding]::UTF8 } catch { }
$OutputEncoding = [System.Text.Encoding]::UTF8
$repoRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path
if (-not $OutDir) { $OutDir = Join-Path $repoRoot 'releases' }
$OutDir = [System.IO.Path]::GetFullPath($OutDir)