ChuNa7a/ChuBlog
1
0 Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

v26.10.1 正式版

Browse Source 
修复部分问题，优化性能，进一步推进类型系统
This commit is contained in:
LIm1tZz7 committed 2026-10-05 22:22:23 +08:00
1 parent 48eec484b7
commit 8ce4239452
89 files changed
+5808 -884
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

chusql-core/storage/src/catalog.rs
+61 -1
View File
@@ -34,6 +34,8 @@ pub struct ColumnStat {
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct TableSchema {
#[serde(default)]
pub object_id: u64,
#[serde(default)]
pub system: bool,
pub columns: Vec<SchemaColumn>,
@@ -51,6 +53,10 @@ pub struct TableSchema {
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct Catalog {
#[serde(default)]
pub database_id: u64,
#[serde(default)]
pub object_high_water: u64,
#[serde(default)]
pub identity_high_water: i64,
tables: BTreeMap<String, TableSchema>,
@@ -68,7 +74,36 @@ impl Catalog {
if bytes.is_empty() {
return Ok(Catalog::default());
}
serde_json::from_slice(&bytes).map_err(io::Error::other)
let mut catalog: Self = serde_json::from_slice(&bytes).map_err(io::Error::other)?;
catalog.upgrade_object_ids()?;
Ok(catalog)
}
/// 补齐旧目录对象编号并校验唯一性
fn upgrade_object_ids(&mut self) -> io::Result<()> {
self.object_high_water = self.object_high_water.max(self.database_id);
let mut ids = BTreeSet::new();
for schema in self.tables.values() {
if schema.object_id != 0 && !ids.insert(schema.object_id) {
return Err(io::Error::other("duplicate catalog object id"));
}
self.object_high_water = self.object_high_water.max(schema.object_id);
}
for schema in self.tables.values_mut() {
if schema.object_id == 0 {
self.object_high_water = self.object_high_water.checked_add(1)
.ok_or_else(|| io::Error::other("catalog object id exhausted"))?;
schema.object_id = self.object_high_water;
}
}
Ok(())
}
/// 分配不复用的对象编号
pub fn reserve_object_id(&mut self) -> io::Result<u64> {
self.object_high_water = self.object_high_water.checked_add(1)
.ok_or_else(|| io::Error::other("catalog object id exhausted"))?;
Ok(self.object_high_water)
}
/// 返回所有表的 (表名, schema)。
@@ -106,15 +141,28 @@ impl Catalog {
/// 建表；已存在报错
pub fn create_table(&mut self, table: &str, columns: Vec<SchemaColumn>) -> io::Result<()> {
self.create_table_with_id(table, columns, None)
}
/// 按日志中的编号建立表
pub fn create_table_with_id(&mut self, table: &str, columns: Vec<SchemaColumn>, object_id: Option<u64>) -> io::Result<()> {
if self.tables.contains_key(table) {
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
format!("table already exists: {}", table),
));
}
let object_id = match object_id {
Some(id) if id == 0 || self.tables.values().any(|schema| schema.object_id == id) => {
return Err(io::Error::other("invalid catalog object id"));
}
Some(id) => { self.object_high_water = self.object_high_water.max(id); id }
None => self.reserve_object_id()?,
};
self.tables.insert(
table.to_string(),
TableSchema {
object_id,
system: false,
columns,
dropped_columns: BTreeSet::new(),
@@ -188,6 +236,18 @@ impl Catalog {
let entry = self.tables.entry(table.to_string()).or_default();
entry.stats.clear();
entry.row_count = 0;
for row in rows {
for (name, value) in row {
if let Some(x) = value.as_f64() {
let stat = entry.stats.entry(name.clone()).or_default();
stat.lo = Some(stat.lo.map_or(x, |lo| lo.min(x)));
stat.hi = Some(stat.hi.map_or(x, |hi| hi.max(x)));
if stat.hist.is_empty() {
stat.hist = vec![0; HIST_BUCKETS];
}
}
}
}
}
let mut changed = false;
for row in rows {
chusql-core/storage/src/protocol.rs
+8 -1
View File
@@ -25,6 +25,8 @@ pub struct Account {
pub is_superuser: bool,
#[serde(default)]
pub system_catalog_manager: bool,
#[serde(default)]
pub allow_sudo_auth: bool,
#[serde(default = "default_true")]
pub enabled: bool,
#[serde(default)]
@@ -96,7 +98,7 @@ pub enum Request {
AccountsList,
IdentityInitialize { administrator: String },
RoleCreate { user: String },
IdentityAlter { user: String, can_login: Option<bool>, is_superuser: Option<bool>, enabled: Option<bool>, system_catalog_manager: Option<bool> },
IdentityAlter { user: String, can_login: Option<bool>, is_superuser: Option<bool>, enabled: Option<bool>, system_catalog_manager: Option<bool>, allow_sudo_auth: Option<bool> },
AccountCreate { user: String, password_hash: String },
AccountReset { user: String, password_hash: String },
AccountLogin {
@@ -305,11 +307,16 @@ pub enum Response {
},
Error { message: String },
Catalog { schemas: Vec<TableSchemaWire> },
Object { database_id: u64, object_id: Option<u64> },
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSchemaWire {
pub table: String,
#[serde(default)]
pub database_id: u64,
#[serde(default)]
pub object_id: u64,
pub columns: Vec<SchemaColumn>,
#[serde(default)]
pub row_count: u64,
chusql-core/storage/src/runtime.rs
+157 -15
View File
@@ -30,10 +30,48 @@ struct Server {
type TableDescription = (Vec<SchemaColumn>, u64, Vec<IndexWire>, Vec<ColumnStatWire>);
impl Server {
/// 读取受保护的对象权限快照
fn object_acl_read(&self) -> std::io::Result<Vec<Row>> {
if !self.table_exists("__system_object_acl")? { return Ok(Vec::new()); }
let rows = self.with_existing_table("__system_object_acl", |table| table.scan())?;
let mut payload = String::new();
for (index, row) in rows.iter().enumerate() {
if row.get("id").and_then(|value| value.as_u64()) != Some(index as u64 + 1) {
return Err(std::io::Error::other("invalid object ACL chunk sequence"));
}
payload.push_str(row.get("payload").and_then(|value| value.as_str())
.ok_or_else(|| std::io::Error::other("invalid object ACL chunk"))?);
}
if rows.is_empty() { return Err(std::io::Error::other("empty object ACL snapshot")); }
Ok(vec![serde_json::from_value(serde_json::json!({"id":1,"payload":payload})).map_err(std::io::Error::other)?])
}
/// 比较并原子替换对象权限快照
fn object_acl_replace(&self, expected: Option<&str>, payload: &str) -> std::io::Result<()> {
let rows = self.object_acl_read()?;
let current = rows.first().and_then(|row| row.get("payload")).and_then(|value| value.as_str());
if current != expected { return Err(std::io::Error::other("object ACL changed concurrently")); }
let _: serde_json::Value = serde_json::from_str(payload).map_err(std::io::Error::other)?;
let mut ops = Vec::new();
if !self.table_exists("__system_object_acl")? {
ops.push(WalOp::CreateTable { table: "__system_object_acl".into(),
columns: vec![SchemaColumn { name: "id".into(), ty: "int".into(), ..Default::default() },
SchemaColumn { name: "payload".into(), ty: "str".into(), ..Default::default() }], object_id: None });
}
let chars: Vec<char> = payload.chars().collect();
let rows = chars.chunks(256).enumerate().map(|(index, chunk)| {
serde_json::from_value(serde_json::json!({"id":index + 1,"payload":chunk.iter().collect::<String>()}))
.map_err(std::io::Error::other)
}).collect::<std::io::Result<Vec<Row>>>()?;
ops.push(WalOp::ReplaceAll { table: "__system_object_acl".into(), rows });
self.group_via_wal(&ops)
}
/// 新建服务器状态
fn new(cfg: Config) -> std::io::Result<Self> {
std::fs::create_dir_all(&cfg.data_dir)?;
let catalog = Catalog::load(cfg.data_dir.join("catalog.json"))?;
catalog.save(cfg.data_dir.join("catalog.json"))?;
let wal = Wal::new(cfg.data_dir.join("wal.log"));
Ok(Server {
cfg,
@@ -306,11 +344,11 @@ impl Server {
let _ = std::fs::remove_file(self.index_path(table, column));
Ok(())
}
WalOp::CreateTable { table, columns } => {
WalOp::CreateTable { table, columns, object_id } => {
self.with_table(table, |_t| Ok(()))?;
let mut c = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
c.create_table(table, columns.clone())?;
if table == TYPES_TABLE { c.mark_system(table)?; }
c.create_table_with_id(table, columns.clone(), *object_id)?;
if table == TYPES_TABLE || table == "__system_object_acl" { c.mark_system(table)?; }
c.save(self.catalog_path())
}
WalOp::DropTable { table } => {
@@ -506,12 +544,22 @@ impl Server {
fn group_via_wal(&self, ops: &[WalOp]) -> std::io::Result<()> {
std::fs::create_dir_all(&self.cfg.data_dir)?;
let wal = &self.wal;
let commit_lsn = wal.append_group(ops)?;
let mut ops = std::borrow::Cow::Borrowed(ops);
if ops.iter().any(|op| matches!(op, WalOp::CreateTable { object_id: None, .. })) {
let mut catalog = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
for op in ops.to_mut() {
if let WalOp::CreateTable { object_id, .. } = op
&& object_id.is_none() {
*object_id = Some(catalog.reserve_object_id()?);
}
}
}
let commit_lsn = wal.append_group(&ops)?;
self.recovery_required.store(true, Ordering::Release);
for op in ops {
for op in ops.iter() {
self.apply_op(op)?;
}
for op in ops {
for op in ops.iter() {
self.flush_op_tables(op)?;
}
self.flush_all_tables()?;
@@ -639,7 +687,16 @@ impl Server {
let c = self.catalog.lock().unwrap();
Ok(c.describe(table).is_some_and(|s| !s.columns.iter().any(|col| &col.name == column)))
}
WalOp::CreateTable { table, .. } => self.table_exists(table),
WalOp::CreateTable { table, object_id, .. } => {
let catalog = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
match catalog.describe(table) {
Some(schema) if object_id.is_some_and(|id| id != schema.object_id) => {
Err(std::io::Error::other("WAL catalog object id mismatch"))
}
Some(_) => Ok(true),
None => Ok(false),
}
}
WalOp::DropTable { table } => Ok(!self.table_exists(table)?
&& !self.table_path(table).exists()
&& !self.index_path(table, "id").exists()),
@@ -1063,11 +1120,11 @@ impl Server {
Request::CatalogManage { command } => {
let target = match command.as_ref() {
Request::AccountCreate { user, .. } | Request::AccountReset { user, .. } | Request::AccountDrop { user }
| Request::IdentityAlter { user, is_superuser: None, system_catalog_manager: None, .. } => Some(account_name(user)?),
| Request::IdentityAlter { user, is_superuser: None, system_catalog_manager: None, allow_sudo_auth: None, .. } => Some(account_name(user)?),
Request::AccountsList => None,
_ => return Err(Error::other("superuser required for this catalog operation")),
};
if accounts.iter().any(|a| target.as_ref() == Some(&a.user) && (a.is_superuser || a.system_catalog_manager)) {
if accounts.iter().any(|a| target.as_ref() == Some(&a.user) && (a.is_superuser || a.system_catalog_manager || a.allow_sudo_auth)) {
return Err(Error::other("superuser required to manage privileged identities"));
}
*command
@@ -1116,10 +1173,13 @@ impl Server {
}
let unified = self.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?.describe(IDENTITIES_TABLE).is_some();
if !accounts.is_empty() { require_superuser(&accounts)?; }
if unified && previous == accounts { return Ok(accounts); }
let sudo_column = self.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?.describe(IDENTITIES_TABLE)
.is_some_and(|schema| schema.columns.iter().any(|column| column.name == "allow_sudo_auth"));
if unified && sudo_column && previous == accounts { return Ok(accounts); }
}
Request::AccountCreate { user, password_hash } => {
let name = account_name(&user)?;
if name == "public" { return Err(Error::other("public is reserved for default privileges")); }
if accounts.iter().any(|a| a.user == name) {
return Err(Error::new(ErrorKind::AlreadyExists, "account already exists"));
}
@@ -1129,6 +1189,7 @@ impl Server {
}
Request::RoleCreate { user } => {
let name = account_name(&user)?;
if name == "public" { return Err(Error::other("public is reserved for default privileges")); }
if accounts.iter().any(|a| a.user == name) { return Err(Error::new(ErrorKind::AlreadyExists, "identity already exists")); }
let id = accounts.iter().map(|a| a.id).max().unwrap_or(0).max(self.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?.identity_high_water)
.checked_add(1).ok_or_else(|| Error::other("account id exhausted"))?;
@@ -1136,13 +1197,14 @@ impl Server {
role.can_login = false;
accounts.push(role);
}
Request::IdentityAlter { user, can_login, is_superuser, enabled, system_catalog_manager } => {
Request::IdentityAlter { user, can_login, is_superuser, enabled, system_catalog_manager, allow_sudo_auth } => {
let name = account_name(&user)?;
let account = accounts.iter_mut().find(|a| a.user == name).ok_or_else(|| Error::new(ErrorKind::NotFound, "unknown account"))?;
account.can_login = can_login.unwrap_or(account.can_login);
account.is_superuser = is_superuser.unwrap_or(account.is_superuser);
account.enabled = enabled.unwrap_or(account.enabled);
account.system_catalog_manager = system_catalog_manager.unwrap_or(account.system_catalog_manager);
account.allow_sudo_auth = allow_sudo_auth.unwrap_or(account.allow_sudo_auth);
account.revision = account.revision.checked_add(1).ok_or_else(|| Error::other("account revision exhausted"))?;
if had_superuser { require_superuser(&accounts)?; }
}
@@ -1226,7 +1288,7 @@ impl Server {
}
if existing.is_some() && original == rows { return Ok(rows); }
let mut ops = Vec::new();
if existing.is_none() { ops.push(WalOp::CreateTable { table: TYPES_TABLE.into(), columns }); }
if existing.is_none() { ops.push(WalOp::CreateTable { table: TYPES_TABLE.into(), columns, object_id: None }); }
ops.push(WalOp::ReplaceAll { table: TYPES_TABLE.into(), rows: rows.clone() });
self.recovery_required.store(true, Ordering::Release);
self.group_via_wal(&ops)?;
@@ -1499,6 +1561,8 @@ impl Server {
.filter(|(name, _)| !blocked_table(name))
.map(|(name, ts)| TableSchemaWire {
table: name.to_string(),
database_id: c.database_id,
object_id: ts.object_id,
columns: ts.columns.clone(),
row_count: ts.row_count,
indexes: index_wires(ts),
@@ -1543,7 +1607,7 @@ impl Server {
};
}
let tname = table.clone();
let op = WalOp::CreateTable { table, columns };
let op = WalOp::CreateTable { table, columns, object_id: None };
match self.write_via_wal(&op) {
Ok(()) => Response::Ok,
Err(e) => {
@@ -1733,6 +1797,7 @@ fn identity_rows(accounts: &[Account]) -> std::io::Result<Vec<Row>> {
("can_login".into(), serde_json::json!(a.can_login)),
("is_superuser".into(), serde_json::json!(a.is_superuser)),
("system_catalog_manager".into(), serde_json::json!(a.system_catalog_manager)),
("allow_sudo_auth".into(), serde_json::json!(a.allow_sudo_auth)),
("enabled".into(), serde_json::json!(a.enabled)),
("identity_version".into(), serde_json::json!(a.identity_version)),
("registered_at".into(), serde_json::Value::String(a.registered_at.clone())),
@@ -1764,6 +1829,7 @@ fn identity_columns() -> Vec<SchemaColumn> {
SchemaColumn { name: "can_login".into(), ty: "bool".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "is_superuser".into(), ty: "bool".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "system_catalog_manager".into(), ty: "bool".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "allow_sudo_auth".into(), ty: "bool".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "enabled".into(), ty: "bool".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "identity_version".into(), ty: "int".into(), nullable: false, ..Default::default() },
SchemaColumn { name: "registered_at".into(), ty: "timestamp".into(),
@@ -1805,6 +1871,10 @@ fn decode_account_row(row: &Row) -> std::io::Result<Account> {
can_login: row.get("can_login").and_then(serde_json::Value::as_bool).unwrap_or(true),
is_superuser: row.get("is_superuser").and_then(serde_json::Value::as_bool).unwrap_or(false),
system_catalog_manager: row.get("system_catalog_manager").and_then(serde_json::Value::as_bool).unwrap_or(false),
allow_sudo_auth: match row.get("allow_sudo_auth") {
None => false,
Some(value) => value.as_bool().ok_or_else(|| std::io::Error::other("invalid sudo authentication attribute"))?,
},
enabled: row.get("enabled").and_then(serde_json::Value::as_bool).unwrap_or(true),
identity_version: row.get("identity_version").and_then(serde_json::Value::as_u64).unwrap_or(0) as u8,
})
@@ -1880,6 +1950,7 @@ fn new_account(id: i64, user: &str, password_hash: String) -> std::io::Result<Ac
can_login: true,
is_superuser: false,
system_catalog_manager: false,
allow_sudo_auth: false,
enabled: true,
identity_version: 1,
})
@@ -2019,7 +2090,37 @@ impl Databases {
named.insert(name, Arc::new(server));
}
}
Ok(Self { root: cfg, system, named: Mutex::new(named) })
{
let mut catalog = system.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
let mut ids = HashSet::new();
if catalog.database_id != 0 { ids.insert(catalog.database_id); }
for server in named.values() {
let id = server.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?.database_id;
if id != 0 && !ids.insert(id) { return Err(std::io::Error::other("duplicate database object id")); }
catalog.object_high_water = catalog.object_high_water.max(id);
}
if catalog.database_id == 0 { catalog.database_id = catalog.reserve_object_id()?; }
catalog.save(system.catalog_path())?;
}
let databases = Self { root: cfg, system, named: Mutex::new(named) };
{
let named = databases.named.lock().map_err(|_| std::io::Error::other("database lock poisoned"))?;
for server in named.values() { databases.assign_database_id(server)?; }
}
Ok(databases)
}
/// 为数据库持久化不复用的编号
fn assign_database_id(&self, server: &Server) -> std::io::Result<()> {
let mut catalog = server.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
if catalog.database_id == 0 {
let mut system = self.system.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
let id = system.reserve_object_id()?;
system.save(self.system.catalog_path())?;
catalog.database_id = id;
catalog.save(server.catalog_path())?;
}
Ok(())
}
/// 按名字打开或新建一个库
@@ -2049,6 +2150,42 @@ impl Databases {
if !name.is_empty() && !valid_database(&name) { return Err(Error::other("invalid database name")); }
let mut named = self.named.lock().map_err(|_| Error::other("database lock poisoned"))?;
match method.as_str() {
"object_acl_read" => {
drop(named);
let _gate = self.system.gate.read().map_err(|_| Error::other("database gate poisoned"))?;
if self.system.recovery_required.load(Ordering::Acquire) { return Err(Error::other("recovery required")); }
return Ok(Response::Rows { rows: self.system.object_acl_read()? });
}
"object_acl_replace" => {
drop(named);
let _gate = self.system.gate.write().map_err(|_| Error::other("database gate poisoned"))?;
if self.system.recovery_required.load(Ordering::Acquire) { return Err(Error::other("recovery required")); }
let expected = match value.get("expected") {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::String(value)) => Some(value.as_str()),
_ => return Err(Error::other("invalid expected ACL payload")),
};
let payload = value.get("payload").and_then(|value| value.as_str()).ok_or_else(|| Error::other("missing ACL payload"))?;
self.system.object_acl_replace(expected, payload)?;
return Ok(Response::Ok);
}
"resolve_object" => {
let server = if name == SYSTEM_DATABASE { &self.system }
else { named.get(&name).ok_or_else(|| Error::other("unknown database"))? };
let _gate = server.gate.read().map_err(|_| Error::other("database gate poisoned"))?;
if server.recovery_required.load(Ordering::Acquire) { return Err(Error::other("recovery required")); }
let catalog = server.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?;
let object_id = match value.get("table") {
None => None,
Some(serde_json::Value::String(table)) if !blocked_table(table) => {
let schema = catalog.describe(table).ok_or_else(|| Error::other("unknown table"))?;
if schema.object_id == 0 { return Err(Error::other("missing catalog object id")); }
Some(schema.object_id)
}
_ => return Err(Error::other("invalid object table")),
};
return Ok(Response::Object { database_id: catalog.database_id, object_id });
}
"all_catalogs" => {
let servers: Vec<(String, Arc<Server>)> =
std::iter::once((SYSTEM_DATABASE.to_string(), Arc::clone(&self.system)))
@@ -2082,6 +2219,7 @@ impl Databases {
cfg.data_dir = cfg.data_dir.join("databases").join(&name);
std::fs::create_dir(&cfg.data_dir)?;
let server = Server::new(cfg)?;
self.assign_database_id(&server)?;
server.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?.save(server.catalog_path())?;
named.insert(name, Arc::new(server));
return Ok(Response::Ok);
@@ -2215,7 +2353,11 @@ fn maintenance_candidate(cfg: Config, mode: &str, user: &str, hash: &str, types:
}
let repair = mode == "repair";
let catalog_path = cfg.data_dir.join("catalog.json");
let schemas = maintenance_schemas();
let mut schemas = maintenance_schemas();
if cfg.data_dir.join("__system_object_acl.db").is_file() {
schemas.push(("__system_object_acl", vec![SchemaColumn { name: "id".into(), ty: "int".into(), ..Default::default() },
SchemaColumn { name: "payload".into(), ty: "str".into(), ..Default::default() }]));
}
let mut missing_data = Vec::new();
let mut catalog = match Catalog::load(&catalog_path) {
Ok(catalog) => catalog,
chusql-core/storage/src/wal.rs
+6 -2
View File
@@ -61,6 +61,7 @@ pub enum WalOp {
CreateTable {
table: String,
columns: Vec<SchemaColumn>,
object_id: Option<u64>,
},
DropTable {
table: String,
@@ -92,6 +93,8 @@ struct ReplaceSchemaPayload {
#[derive(Debug, Serialize, Deserialize)]
struct CreateTablePayload {
columns: Vec<SchemaColumn>,
#[serde(default)]
object_id: Option<u64>,
}
pub struct Wal {
@@ -416,10 +419,10 @@ fn encode(op: &WalOp, lsn: u64) -> io::Result<Vec<u8>> {
.map_err(io::Error::other)?;
(OP_REPLACE_SCHEMA, table.clone(), p)
}
WalOp::CreateTable { table, columns } => (
WalOp::CreateTable { table, columns, object_id } => (
OP_CREATE_TABLE,
table.clone(),
serde_json::to_vec(&CreateTablePayload { columns: columns.clone() })
serde_json::to_vec(&CreateTablePayload { columns: columns.clone(), object_id: *object_id })
.map_err(io::Error::other)?,
),
WalOp::DropTable { table } => (OP_DROP_TABLE, table.clone(), Vec::new()),
@@ -536,6 +539,7 @@ fn decode_body(body: &[u8]) -> Result<(u64, WalOp), String> {
WalOp::CreateTable {
table,
columns: p.columns,
object_id: p.object_id,
}
}
OP_DROP_TABLE => WalOp::DropTable { table },
chusql-core/storage/tests/catalog_test.rs
+71
View File
@@ -4,6 +4,55 @@ use serde_json::json;
// 数据字典测试：补列、去重、未知表。
/// 表删除重建和重载均保留编号规则
#[test]
fn object_ids_survive_reload_and_are_not_reused() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("catalog.json");
let mut catalog = Catalog::default();
catalog.create_table("t", vec![]).unwrap();
let old = catalog.describe("t").unwrap().object_id;
assert_ne!(old, 0);
catalog.save(&path).unwrap();
let mut catalog = Catalog::load(&path).unwrap();
assert_eq!(catalog.describe("t").unwrap().object_id, old);
catalog.drop_table("t").unwrap();
catalog.save(&path).unwrap();
let mut catalog = Catalog::load(&path).unwrap();
catalog.create_table("t", vec![]).unwrap();
assert!(catalog.describe("t").unwrap().object_id > old);
}
/// 旧目录迁移保留已有编号并拒绝重复
#[test]
fn legacy_object_ids_upgrade_without_collisions() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("catalog.json");
std::fs::write(&path, json!({"tables": {
"a": {"columns": [], "object_id": 42},
"b": {"columns": []}
}}).to_string()).unwrap();
let catalog = Catalog::load(&path).unwrap();
assert_eq!(catalog.describe("a").unwrap().object_id, 42);
assert_eq!(catalog.describe("b").unwrap().object_id, 43);
catalog.save(&path).unwrap();
assert_eq!(Catalog::load(&path).unwrap().describe("b").unwrap().object_id, 43);
std::fs::write(&path, json!({"tables": {
"a": {"columns": [], "object_id": 42},
"b": {"columns": [], "object_id": 42}
}}).to_string()).unwrap();
assert!(Catalog::load(&path).unwrap_err().to_string().contains("duplicate"));
}
/// 编号耗尽明确失败且不建立对象
#[test]
fn exhausted_object_ids_fail_without_creating_table() {
let mut catalog = Catalog::default();
catalog.object_high_water = u64::MAX;
assert!(catalog.create_table("t", vec![]).is_err());
assert!(catalog.describe("t").is_none());
}
/// 用键值对构造一行
@@ -154,6 +203,28 @@ fn capped_distinct_keeps_histogram_and_other_columns_accurate() {
assert_eq!(catalog.describe("t").unwrap().stats["id"].distinct, 4000);
}
/// 重建直方图不受行顺序影响
#[test]
fn rebuilt_histogram_has_exact_buckets_in_any_order() {
let rows: Vec<Row> = (0..160).map(|n| row(&[("v", json!(n))])).collect();
let mut forward = Catalog::default();
forward.rebuild_stats("t", &rows);
let mut reverse = Catalog::default();
let reversed: Vec<Row> = rows.iter().rev().cloned().collect();
reverse.rebuild_stats("t", &reversed);
let stat = &forward.describe("t").unwrap().stats["v"];
assert_eq!(stat.hist, vec![10; 16]);
assert_eq!(stat.hist, reverse.describe("t").unwrap().stats["v"].hist);
forward.record_insert("t", &row(&[("v", json!(10000))]));
forward.record_delete("t", &[row(&[("v", json!(10000))])]);
forward.rebuild_stats("t", &rows[80..]);
let repaired = &forward.describe("t").unwrap().stats["v"];
assert_eq!(repaired.lo, Some(80.0));
assert_eq!(repaired.hi, Some(159.0));
assert_eq!(repaired.hist, vec![5; 16]);
assert_eq!(forward.describe("t").unwrap().row_count, 80);
}
/// 老类型名并到新写法，参数保留，未知类型原样
#[test]
fn normalize_type_maps_legacy_names() {
chusql-core/storage/tests/runtime_test.rs
+248 -2
View File
@@ -5,11 +5,165 @@ use chusql_core_storage::runtime::Storage;
// 进程内存储集成测试：协议请求、索引、账号与配置。
/// 权限快照支持跨页文本并拒绝旧版本覆盖
#[test]
fn object_acl_snapshots_are_private_and_compare_versions() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
let mut c = connect()?;
let payload = serde_json::json!({"grants": "权限\"\\".repeat(4000)}).to_string();
request_ok(&mut c, serde_json::json!({"method":"object_acl_replace","expected":null,"payload":payload}))?;
let current: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"object_acl_read"}"#))?;
assert_eq!(current["rows"][0]["payload"], payload);
let conflict = send(&mut c, r#"{"method":"object_acl_replace","expected":null,"payload":"{}"}"#);
assert!(conflict.contains("changed concurrently"), "{conflict}");
assert!(send(&mut c, r#"{"method":"scan","database":"system","table":"__system_object_acl"}"#).contains("reserved system table"));
assert!(!send(&mut c, r#"{"method":"all_catalogs"}"#).contains("__system_object_acl"));
drop(c);
drop(server);
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
let reloaded: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"object_acl_read"}"#))?;
assert_eq!(reloaded, current);
Ok(())
}
/// 权限并发替换只能有一个版本提交
#[test]
fn concurrent_object_acl_replacement_preserves_one_winner() -> Result<(), Box<dyn std::error::Error>> {
let (_server, _data) = start_server();
let mut c = connect()?;
let workers: Vec<_> = (0..8).map(|index| {
let storage = Arc::clone(&c.0);
std::thread::spawn(move || storage.request_line(&serde_json::json!({
"method":"object_acl_replace","expected":null,"payload":serde_json::json!({"winner":index}).to_string()
}).to_string()))
}).collect();
let replies: Vec<serde_json::Value> = workers.into_iter().map(|worker| serde_json::from_str(&worker.join().unwrap()).unwrap()).collect();
assert_eq!(replies.iter().filter(|reply| reply["status"] == "ok").count(), 1);
assert_eq!(replies.iter().filter(|reply| reply["message"].as_str().is_some_and(|message| message.contains("changed concurrently"))).count(), 7);
let loaded: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"object_acl_read"}"#))?;
assert!(serde_json::from_str::<serde_json::Value>(loaded["rows"][0]["payload"].as_str().unwrap())?["winner"].is_number());
Ok(())
}
/// 权限快照落盘失败后重放完整替换
#[test]
fn object_acl_commit_failure_replays_the_complete_snapshot() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
let mut c = connect()?;
let pending = data.path().join("system/catalog.json.pending");
std::fs::create_dir(&pending)?;
let failed = send(&mut c, r#"{"method":"object_acl_replace","expected":null,"payload":"{\"version\":1}"}"#);
assert!(failed.contains("error"), "{failed}");
assert!(send(&mut c, r#"{"method":"object_acl_read"}"#).contains("recovery required"));
drop(c);
drop(server);
std::fs::remove_dir(pending)?;
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
let loaded: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"object_acl_read"}"#))?;
assert_eq!(loaded["rows"][0]["payload"], "{\"version\":1}");
Ok(())
}
/// 对象编号区分跨库同名及删除重建
#[test]
fn object_identity_tracks_database_and_table_incarnations() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
let mut c = connect()?;
request_ok(&mut c, serde_json::json!({"method":"create_database","database":"other"}))?;
for database in ["main", "other"] {
request_ok(&mut c, serde_json::json!({"method":"create_table","database":database,"table":"items","columns":[{"name":"id","ty":"int"}]}))?;
}
let old: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"main","table":"items"}"#))?;
let other: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"other","table":"items"}"#))?;
assert_eq!(old["status"], "object");
assert_ne!(old["database_id"], other["database_id"]);
assert!(old["object_id"].as_u64().unwrap() > 0);
let catalog: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"list_catalog","database":"main"}"#))?;
let info = catalog["schemas"].as_array().unwrap().iter().find(|entry| entry["table"] == "items").unwrap();
assert_eq!(info["database_id"], old["database_id"]);
assert_eq!(info["object_id"], old["object_id"]);
drop(c);
drop(server);
let restarted = restart_at(data.path())?;
let mut c = connect()?;
let stable: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"main","table":"items"}"#))?;
assert_eq!(stable, old);
request_ok(&mut c, serde_json::json!({"method":"drop_table","database":"main","table":"items"}))?;
assert!(send(&mut c, r#"{"method":"resolve_object","database":"main","table":"items"}"#).contains("unknown table"));
request_ok(&mut c, serde_json::json!({"method":"create_table","database":"main","table":"items","columns":[{"name":"id","ty":"int"}]}))?;
let new: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"main","table":"items"}"#))?;
assert_eq!(new["database_id"], old["database_id"]);
assert_ne!(new["object_id"], old["object_id"]);
request_ok(&mut c, serde_json::json!({"method":"drop_database","database":"other"}))?;
request_ok(&mut c, serde_json::json!({"method":"create_database","database":"other"}))?;
let new_database: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"other"}"#))?;
assert_ne!(new_database["database_id"], other["database_id"]);
assert!(new_database["object_id"].is_null());
drop(c);
drop(restarted);
Ok(())
}
thread_local! {
/// 当前线程的存储实例：一个用例一个线程，用例内部按顺序起/停
static CURRENT: RefCell<Option<Arc<Storage>>> = const { RefCell::new(None) };
}
/// 建表应用失败后按日志恢复原编号
#[test]
fn object_id_survives_committed_create_recovery() -> Result<(), Box<dyn std::error::Error>> {
use chusql_core_storage::wal::{Wal, WalOp};
let (server, data) = start_server();
let mut c = connect()?;
let directory = data.path().join("databases/main");
let pending = directory.join("catalog.json.pending");
std::fs::create_dir(&pending)?;
assert!(send(&mut c, r#"{"method":"create_table","table":"recovered","columns":[{"name":"id","ty":"int"}]}"#).contains("error"));
assert!(send(&mut c, r#"{"method":"resolve_object","database":"main","table":"recovered"}"#).contains("recovery required"));
let wal = Wal::new(directory.join("wal.log"));
let scan = wal.read_all()?;
let id = scan.records.iter().find_map(|(_, op)| match op {
WalOp::CreateTable { table, object_id, .. } if table == "recovered" => *object_id,
_ => None,
}).unwrap();
drop(wal);
drop(c);
drop(server);
std::fs::remove_dir(pending)?;
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
let resolved: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"resolve_object","database":"main","table":"recovered"}"#))?;
assert_eq!(resolved["object_id"], id);
Ok(())
}
/// 并发建表分配互不重复的编号
#[test]
fn concurrent_table_creation_allocates_distinct_object_ids() -> Result<(), Box<dyn std::error::Error>> {
let (_server, _data) = start_server();
let mut c = connect()?;
let workers: Vec<_> = (0..16).map(|index| {
let storage = Arc::clone(&c.0);
std::thread::spawn(move || storage.request_line(&serde_json::json!({
"method":"create_table", "database":"main", "table":format!("concurrent_{index}"),
"columns":[{"name":"id","ty":"int"}]
}).to_string()))
}).collect();
for worker in workers {
let result: serde_json::Value = serde_json::from_str(&worker.join().unwrap())?;
assert_eq!(result["status"], "ok");
}
let response: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"list_catalog","database":"main"}"#))?;
let ids: std::collections::HashSet<_> = response["schemas"].as_array().unwrap().iter()
.filter(|schema| schema["table"].as_str().unwrap().starts_with("concurrent_"))
.map(|schema| schema["object_id"].as_u64().unwrap()).collect();
assert_eq!(ids.len(), 16);
assert!(!ids.contains(&0));
Ok(())
}
/// 定点写入检查舍入和整批原子拒绝
#[test]
fn decimal_writes_enforce_precision_before_wal() -> Result<(), Box<dyn std::error::Error>> {
@@ -246,6 +400,66 @@ fn legacy_account_wal_migrates_without_promoting_other_names() -> Result<(), Box
Ok(())
}
/// 读取并校验身份命令响应
fn account_response(c: &mut Conn, request: serde_json::Value) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
let response: serde_json::Value = serde_json::from_str(&send(c, &request.to_string()))?;
assert_eq!(response["status"], "accounts", "{response}");
Ok(response)
}
/// 本机认证开关独立持久化并推进会话版本
#[test]
fn sudo_auth_is_independent_revocable_and_persistent() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
let mut c = connect()?;
let original = account_response(&mut c, serde_json::json!({"method":"account_create","user":"local_user","password_hash":"hash"}))?;
let created = original["accounts"].as_array().unwrap().iter().find(|a| a["user"] == "local_user").unwrap();
assert_eq!(created["allow_sudo_auth"], false);
let changed = account_response(&mut c, serde_json::json!({"method":"identity_alter","user":"local_user","allow_sudo_auth":true}))?;
let local = changed["accounts"].as_array().unwrap().iter().find(|a| a["user"] == "local_user").unwrap();
assert_eq!(local["allow_sudo_auth"], true);
assert_eq!(local["is_superuser"], false);
assert_eq!(local["system_catalog_manager"], false);
assert_eq!(local["revision"], 2);
for command in [
serde_json::json!({"method":"identity_alter","user":"local_user","allow_sudo_auth":false}),
serde_json::json!({"method":"account_reset","user":"local_user","password_hash":"other"}),
serde_json::json!({"method":"identity_alter","user":"root","allow_sudo_auth":true}),
] {
let denied = send(&mut c, &serde_json::json!({"method":"catalog_manage","command":command}).to_string());
assert!(denied.contains("superuser required"), "{denied}");
}
drop(c);
drop(server);
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
assert_eq!(account_response(&mut c, serde_json::json!({"method":"accounts_list"}))?, changed);
let revoked = account_response(&mut c, serde_json::json!({"method":"identity_alter","user":"local_user","allow_sudo_auth":false}))?;
let local = revoked["accounts"].as_array().unwrap().iter().find(|a| a["user"] == "local_user").unwrap();
assert_eq!(local["allow_sudo_auth"], false);
assert_eq!(local["revision"], 3);
Ok(())
}
/// 升级缺失本机认证列的旧目录
#[test]
fn sudo_auth_upgrades_old_identity_schema() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
drop(server);
CURRENT.with(|slot| *slot.borrow_mut() = None);
let path = data.path().join("system/catalog.json");
let mut catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(&path)?)?;
catalog["tables"]["__system_identities"]["columns"].as_array_mut().unwrap().retain(|column| column["name"] != "allow_sudo_auth");
std::fs::write(&path, serde_json::to_vec(&catalog)?)?;
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
let upgraded = account_response(&mut c, serde_json::json!({"method":"identity_initialize","administrator":"root"}))?;
assert!(upgraded["accounts"].as_array().unwrap().iter().all(|a| a["allow_sudo_auth"] == false));
let catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(&path)?)?;
assert!(catalog["tables"]["__system_identities"]["columns"].as_array().unwrap().iter().any(|column| column["name"] == "allow_sudo_auth"));
Ok(())
}
/// 身份属性重启保留且禁止删除最后管理员
#[test]
fn identity_attributes_survive_restart_and_protect_superuser() -> Result<(), Box<dyn std::error::Error>> {
@@ -1691,6 +1905,7 @@ fn create_table_replays_after_restart() -> Result<(), Box<dyn std::error::Error>
drop(server);
let group = vec![WalOp::CreateTable {
object_id: None,
table: "m4_create".into(),
columns: vec![
SchemaColumn {
@@ -1835,6 +2050,7 @@ fn already_applied_create_table_is_skipped_on_restart() -> Result<(), Box<dyn st
let wal = Wal::new(data.path().join("databases/main/wal.log"));
wal.append_group(&[WalOp::CreateTable {
object_id: None,
table: "m4_done".into(),
columns: vec![SchemaColumn {
name: "id".into(),
@@ -1995,6 +2211,36 @@ fn create_index_accepts_duplicate_values() {
assert!(r.contains(r#""status":"error""#), "the id index stays unique: {}", r);
}
/// 重启修复统计并保持点查范围结果
#[test]
fn restart_rebuilds_stale_stats_without_changing_index_results() -> Result<(), Box<dyn std::error::Error>> {
let (server, data) = start_server();
let mut c = connect()?;
let rows: Vec<serde_json::Value> = (0..160).map(|id| serde_json::json!({"id":id,"v":id})).collect();
request_ok(&mut c, serde_json::json!({"method":"insert_batch","table":"rebuilt","rows":rows}))?;
request_ok(&mut c, serde_json::json!({"method":"create_index","table":"rebuilt","column":"v"}))?;
let point: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"lookup_by_index","table":"rebuilt","column":"v","key":80}"#))?;
let range: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"range_by_index","table":"rebuilt","column":"v","lo":80,"hi":89}"#))?;
let scan: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"scan","table":"rebuilt"}"#))?;
drop(c);
drop(server);
let path = data.path().join("databases/main/catalog.json");
let mut catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(&path)?)?;
catalog["tables"]["rebuilt"]["row_count"] = serde_json::json!(9999);
catalog["tables"]["rebuilt"]["stats"] = serde_json::json!({});
std::fs::write(&path, serde_json::to_vec(&catalog)?)?;
let _restarted = restart_at(data.path())?;
let mut c = connect()?;
assert_eq!(serde_json::from_str::<serde_json::Value>(&send(&mut c, r#"{"method":"lookup_by_index","table":"rebuilt","column":"v","key":80}"#))?, point);
assert_eq!(serde_json::from_str::<serde_json::Value>(&send(&mut c, r#"{"method":"range_by_index","table":"rebuilt","column":"v","lo":80,"hi":89}"#))?, range);
assert_eq!(serde_json::from_str::<serde_json::Value>(&send(&mut c, r#"{"method":"scan","table":"rebuilt"}"#))?, scan);
let schema: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"describe_table","table":"rebuilt"}"#))?;
assert_eq!(schema["row_count"], 160);
let stat = schema["stats"].as_array().unwrap().iter().find(|stat| stat["name"] == "v").unwrap();
assert_eq!(stat["hist"], serde_json::json!([10,10,10,10,10,10,10,10,10,10,10,10,10,10,10,10]));
Ok(())
}
/// 数据字典里的统计：不同值个数
#[test]
fn stats_report_distinct_values() {
@@ -2264,9 +2510,9 @@ fn account_table_uses_typed_columns_and_stamps_login() -> Result<(), Box<dyn std
let entry = &catalog["tables"]["__system_identities"];
let columns = entry["columns"].as_array().ok_or("missing columns")?;
let names: Vec<&str> = columns.iter().map(|col| col["name"].as_str().unwrap_or("")).collect();
assert_eq!(names, vec!["id", "user", "can_login", "is_superuser", "system_catalog_manager", "enabled", "identity_version", "registered_at", "last_login_at", "revision"]);
assert_eq!(names, vec!["id", "user", "can_login", "is_superuser", "system_catalog_manager", "allow_sudo_auth", "enabled", "identity_version", "registered_at", "last_login_at", "revision"]);
let types: Vec<&str> = columns.iter().map(|col| col["ty"].as_str().unwrap_or("")).collect();
assert_eq!(types, vec!["int", "varchar(64)", "bool", "bool", "bool", "bool", "int", "timestamp", "timestamp", "int"]);
assert_eq!(types, vec!["int", "varchar(64)", "bool", "bool", "bool", "bool", "bool", "int", "timestamp", "timestamp", "int"]);
assert_eq!(columns[0]["primary_key"], true, "id 是主键");
assert_eq!(columns[0]["nullable"], false);
assert_eq!(columns[1]["unique"], true, "用户名唯一");
chusql-core/storage/tests/wal_test.rs
+23
View File
@@ -4,6 +4,29 @@ use serde_json::json;
// WAL 测试：追加、读取、清空、截断尾巴、LSN 与提交标记。
/// 建表编号完整往返且兼容旧载荷
#[test]
fn create_table_ids_round_trip_and_legacy_payloads_decode() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("wal.log");
let wal = Wal::new(&path);
let op = WalOp::CreateTable { table: "t".into(), columns: vec![], object_id: Some(42) };
wal.append(&op).unwrap();
assert_eq!(wal.read_all().unwrap().records[0].1, op);
drop(wal);
let mut body = 1_u64.to_le_bytes().to_vec();
body.push(11);
body.extend_from_slice(&1_u16.to_le_bytes());
body.extend_from_slice(b"t");
body.extend_from_slice(br#"{"columns":[]}"#);
let mut frame = (body.len() as u32).to_le_bytes().to_vec();
frame.extend_from_slice(&body);
std::fs::write(&path, frame).unwrap();
let wal = Wal::new(&path);
assert_eq!(wal.read_all().unwrap().records[0].1,
WalOp::CreateTable { table: "t".into(), columns: vec![], object_id: None });
}
/// 用键值对构造一行
fn row(pairs: &[(&str, serde_json::Value)]) -> Row {