ChuNa7a/ChuSQL
1
0 Fork 0
Code Issues Pull Requests Actions 7 Packages Projects Releases Wiki Activity

修复bug

Browse Source
This commit is contained in:
LIm1tZz7 committed 2026-10-11 17:32:12 +08:00
1 parent fc9bbc20e4
commit e87c4ee6a8
90 files changed
+3828 -452
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

chusql-core/storage/src/catalog.rs
+62 -28
View File
@@ -37,6 +37,9 @@ pub struct TableSchema {
#[serde(default)]
pub object_id: u64,
#[serde(default)]
// 0 表示旧目录尚无可验证的结构版本。
pub schema_version: u64,
#[serde(default)]
pub system: bool,
pub columns: Vec<SchemaColumn>,
#[serde(default)]
@@ -51,7 +54,7 @@ pub struct TableSchema {
pub stats: BTreeMap<String, ColumnStat>,
}
#[derive(Debug, Default, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Catalog {
#[serde(default)]
pub database_id: u64,
@@ -134,9 +137,10 @@ impl Catalog {
}
/// 按一行数据补列
pub fn ensure_columns(&mut self, table: &str, row: &Row) {
pub fn ensure_columns(&mut self, table: &str, row: &Row) -> io::Result<()> {
let entry = self.tables.entry(table.to_string()).or_default();
add_missing_columns(entry, row);
add_missing_columns(entry, row)?;
Ok(())
}
/// 建表；已存在报错
@@ -163,6 +167,7 @@ impl Catalog {
table.to_string(),
TableSchema {
object_id,
schema_version: 1,
system: false,
columns,
dropped_columns: BTreeSet::new(),
@@ -184,15 +189,15 @@ impl Catalog {
}
/// 记一次插入，返回列有没有变化
pub fn record_insert(&mut self, table: &str, row: &Row) -> bool {
pub fn record_insert(&mut self, table: &str, row: &Row) -> io::Result<bool> {
let changed = {
let entry = self.tables.entry(table.to_string()).or_default();
let changed = add_missing_columns(entry, row);
let changed = add_missing_columns(entry, row)?;
entry.row_count += 1;
changed
};
self.count_values(table, row);
changed
Ok(changed)
}
/// 记一次删除：行数减少 + 更新列统计
@@ -230,7 +235,7 @@ impl Catalog {
}
/// 重数一遍统计与行数
pub fn rebuild_stats(&mut self, table: &str, rows: &[Row]) -> bool {
pub fn rebuild_stats(&mut self, table: &str, rows: &[Row]) -> io::Result<bool> {
self.seen.remove(table);
{
let entry = self.tables.entry(table.to_string()).or_default();
@@ -251,9 +256,9 @@ impl Catalog {
}
let mut changed = false;
for row in rows {
changed = self.record_insert(table, row) || changed;
changed = self.record_insert(table, row)? || changed;
}
changed
Ok(changed)
}
/// 这一列第一次见到这个值就加一
@@ -291,16 +296,18 @@ impl Catalog {
/// 表改名到指定名字
pub fn rename_table(&mut self, from: &str, to: &str) -> io::Result<bool> {
let Some(entry) = self.tables.remove(from) else {
let Some(entry) = self.tables.get(from) else {
return Ok(false);
};
if self.tables.contains_key(to) {
self.tables.insert(from.to_string(), entry);
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
format!("table already exists: {}", to),
));
}
let version = next_schema_version(entry)?;
let mut entry = self.tables.remove(from).ok_or_else(|| io::Error::other("missing rename source"))?;
entry.schema_version = version;
self.seen.remove(from);
self.tables.insert(to.to_string(), entry);
Ok(true)
@@ -308,10 +315,21 @@ impl Catalog {
/// ALTER：整列定义换成新的，并清掉二级索引与统计
pub fn set_columns(&mut self, table: &str, columns: Vec<SchemaColumn>) -> io::Result<()> {
self.set_columns_version(table, columns, None)
}
/// 用日志中的目标版本替换列定义
pub fn set_columns_version(&mut self, table: &str, columns: Vec<SchemaColumn>, version: Option<u64>) -> io::Result<()> {
let entry = self
.tables
.get_mut(table)
.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, format!("unknown table: {}", table)))?;
if let Some(version) = version {
if version == 0 { return Err(io::Error::other("invalid WAL schema version")); }
entry.schema_version = version;
} else if entry.columns != columns {
entry.schema_version = next_schema_version(entry)?;
}
entry.columns = columns;
entry.dropped_columns.clear();
entry.indexes.clear();
@@ -321,9 +339,12 @@ impl Catalog {
}
/// 老数据的列类型名规范化；返回有没有改动
pub fn normalize_types(&mut self) -> bool {
pub fn normalize_types(&mut self) -> io::Result<bool> {
let mut changed = false;
for entry in self.tables.values_mut() {
if entry.columns.iter().any(|column| normalize_type(&column.ty) != column.ty) {
entry.schema_version = next_schema_version(entry)?;
}
for column in entry.columns.iter_mut() {
let fixed = normalize_type(&column.ty);
if fixed != column.ty {
@@ -332,15 +353,26 @@ impl Catalog {
}
}
}
changed
Ok(changed)
}
/// 删一列；之前没有也返回成功（重放幂等）
pub fn remove_column(&mut self, table: &str, column: &str) -> io::Result<bool> {
self.remove_column_version(table, column, None)
}
/// 用日志目标版本删除列定义
pub fn remove_column_version(&mut self, table: &str, column: &str, version: Option<u64>) -> io::Result<bool> {
let entry = self
.tables
.get_mut(table)
.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, format!("unknown table: {}", table)))?;
if let Some(version) = version {
if version == 0 { return Err(io::Error::other("invalid WAL schema version")); }
entry.schema_version = version;
} else if entry.columns.iter().any(|c| c.name == column) {
entry.schema_version = next_schema_version(entry)?;
}
let had_column = {
let before = entry.columns.len();
entry.columns.retain(|c| c.name != column);
@@ -417,22 +449,24 @@ impl Catalog {
}
/// 把行里"字典还不知道的列"补进来；返回有没有变化
fn add_missing_columns(entry: &mut TableSchema, row: &Row) -> bool {
let mut changed = false;
for (k, v) in row {
if entry.dropped_columns.contains(k) || entry.columns.iter().any(|c| &c.name == k) {
continue;
}
if let Some(ty) = infer_type(v) {
entry.columns.push(SchemaColumn {
name: k.clone(),
ty,
..Default::default()
});
changed = true;
}
fn add_missing_columns(entry: &mut TableSchema, row: &Row) -> io::Result<bool> {
let additions: Vec<_> = row.iter().filter(|(name, _)|
!entry.dropped_columns.contains(*name) && !entry.columns.iter().any(|c| &c.name == *name))
.filter_map(|(name, value)| infer_type(value).map(|ty| SchemaColumn {
name: name.clone(), ty, ..Default::default()
})).collect();
if additions.is_empty() {
return Ok(false);
}
changed
entry.schema_version = next_schema_version(entry)?;
entry.columns.extend(additions);
Ok(true)
}
/// 递增结构版本，耗尽时拒绝变更
fn next_schema_version(entry: &TableSchema) -> io::Result<u64> {
entry.schema_version.checked_add(1)
.ok_or_else(|| io::Error::other("catalog schema version exhausted"))
}
/// 由 JSON 值猜列类型
chusql-core/storage/src/ffi.rs
+18 -16
View File
@@ -74,14 +74,19 @@ pub unsafe extern "C" fn chusql_storage_close(handle: *mut Storage) {
/// 离线维护系统目录并返回 JSON
///
/// # Safety
/// 输入为有效字符串；输出指针可写，响应使用 free 释放。
/// `config_path` 为 NULL 或有效 NUL 结尾字符串；请求字节在调用期间有效。
/// 输出指针可写且不与输入重叠；响应使用 free 释放。
#[unsafe(no_mangle)]
pub unsafe extern "C" fn chusql_storage_maintenance(
config_path: *const c_char, request: *const u8, request_len: usize,
out: *mut *mut u8, out_len: *mut usize,
) -> i32 {
if request.is_null() || out.is_null() || out_len.is_null() { set_error("invalid maintenance argument"); return ERR; }
unsafe { *out = ptr::null_mut(); *out_len = 0; }
if !out.is_null() { unsafe { *out = ptr::null_mut(); } }
if !out_len.is_null() { unsafe { *out_len = 0; } }
if request.is_null() || out.is_null() || out_len.is_null() || request_len > isize::MAX as usize {
set_error("invalid maintenance argument");
return ERR;
}
let result = catch_unwind(AssertUnwindSafe(|| {
let path = if config_path.is_null() { None } else { Some(unsafe { CStr::from_ptr(config_path) }.to_str().map_err(|e| e.to_string())?) };
let text = std::str::from_utf8(unsafe { slice::from_raw_parts(request, request_len) }).map_err(|e| e.to_string())?;
@@ -92,16 +97,16 @@ pub unsafe extern "C" fn chusql_storage_maintenance(
Ok(Err(err)) => { set_error(err); return ERR; }
Err(_) => { set_error("panic in system maintenance"); return ERR; }
};
let mut buffer = response.into_bytes().into_boxed_slice();
unsafe { *out = buffer.as_mut_ptr(); *out_len = buffer.len(); }
std::mem::forget(buffer);
let buffer = response.into_bytes().into_boxed_slice();
unsafe { *out_len = buffer.len(); *out = Box::into_raw(buffer) as *mut u8; }
OK
}
/// 处理一条请求，响应缓冲须归还
///
/// # Safety
/// `handle` 必须是有效句柄；`request` 指向 `request_len` 个字节；`out`/`out_len` 可写。
/// `handle` 在调用期间有效且不能并发关闭；请求字节在调用期间有效。
/// `out`/`out_len` 可写且不与输入重叠；响应使用 free 释放。
#[unsafe(no_mangle)]
pub unsafe extern "C" fn chusql_storage_request(
handle: *const Storage,
@@ -116,7 +121,7 @@ pub unsafe extern "C" fn chusql_storage_request(
if !out_len.is_null() {
unsafe { *out_len = 0 };
}
if handle.is_null() || request.is_null() || out.is_null() || out_len.is_null() {
if handle.is_null() || request.is_null() || out.is_null() || out_len.is_null() || request_len > isize::MAX as usize {
set_error("invalid argument");
return ERR;
}
@@ -139,13 +144,10 @@ pub unsafe extern "C" fn chusql_storage_request(
}
};
let mut buffer = response.into_bytes().into_boxed_slice();
let ptr = buffer.as_mut_ptr();
let len = buffer.len();
std::mem::forget(buffer);
let buffer = response.into_bytes().into_boxed_slice();
unsafe {
*out = ptr;
*out_len = len;
*out_len = buffer.len();
*out = Box::into_raw(buffer) as *mut u8;
}
OK
}
@@ -153,7 +155,7 @@ pub unsafe extern "C" fn chusql_storage_request(
/// 归还响应缓冲
///
/// # Safety
/// `data`/`len` 必须是 chusql_storage_request 原样返回且未被释放过的一对值。
/// `data`/`len` 为 request 或 maintenance 原样返回且未释放的一对值。
#[unsafe(no_mangle)]
pub unsafe extern "C" fn chusql_storage_free(data: *mut u8, len: usize) {
if data.is_null() {
@@ -168,7 +170,7 @@ pub unsafe extern "C" fn chusql_storage_free(data: *mut u8, len: usize) {
/// 取本线程最近的失败信息
///
/// # Safety
/// `out_len` 可写，或为 NULL。
/// `out_len` 可写，或为 NULL；返回字节仅在本线程下次设置错误前有效，不能释放。
#[unsafe(no_mangle)]
pub unsafe extern "C" fn chusql_storage_last_error(out_len: *mut usize) -> *const u8 {
LAST_ERROR.with(|slot| {
chusql-core/storage/src/protocol.rs
+30
View File
@@ -283,12 +283,15 @@ fn default_id_column() -> String {
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum Response {
Guarded { response: Box<Response> },
Accounts { accounts: Vec<Account> },
Pong,
Rows { rows: Vec<Row> },
Tables { tables: Vec<String> },
Schema {
table: String,
#[serde(default, serialize_with = "serialize_version", deserialize_with = "deserialize_version")]
schema_version: u64,
columns: Vec<SchemaColumn>,
#[serde(default)]
row_count: u64,
@@ -310,6 +313,15 @@ pub enum Response {
Object { database_id: u64, object_id: Option<u64> },
}
#[derive(Debug, Deserialize)]
pub struct BoundSchema {
pub table: String,
pub database_id: Option<String>,
pub object_id: Option<String>,
pub schema_version: Option<String>,
pub columns: Vec<SchemaColumn>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSchemaWire {
pub table: String,
@@ -317,6 +329,8 @@ pub struct TableSchemaWire {
pub database_id: u64,
#[serde(default)]
pub object_id: u64,
#[serde(default, serialize_with = "serialize_version", deserialize_with = "deserialize_version")]
pub schema_version: u64,
pub columns: Vec<SchemaColumn>,
#[serde(default)]
pub row_count: u64,
@@ -325,3 +339,19 @@ pub struct TableSchemaWire {
#[serde(default)]
pub stats: Vec<ColumnStatWire>,
}
/// 将结构版本编码成精确十进制字符串
fn serialize_version<S: serde::Serializer>(version: &u64, serializer: S) -> Result<S::Ok, S::Error> {
serializer.serialize_str(&version.to_string())
}
/// 读取精确版本并兼容旧数值载荷
fn deserialize_version<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<u64, D::Error> {
let value = serde_json::Value::deserialize(deserializer)?;
match value {
serde_json::Value::String(s) if !s.is_empty() && s.len() <= 20 && s.bytes().all(|c| c.is_ascii_digit()) =>
s.parse().map_err(serde::de::Error::custom),
serde_json::Value::Number(n) => n.as_u64().ok_or_else(|| serde::de::Error::custom("invalid schema version")),
_ => Err(serde::de::Error::custom("invalid schema version")),
}
}
chusql-core/storage/src/runtime.rs
+115 -24
View File
@@ -236,9 +236,9 @@ impl Server {
fn apply_op(&self, op: &WalOp) -> std::io::Result<()> {
match op {
WalOp::Commit => Ok(()),
WalOp::HideColumn { table, column } => {
WalOp::HideColumn { table, column, schema_version } => {
let mut c = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
c.remove_column(table, column)?;
c.remove_column_version(table, column, *schema_version)?;
c.save(self.catalog_path())?;
drop(c);
if let Some(handle) = self.tables.lock().map_err(|_| std::io::Error::other("table lock poisoned"))?.get(table) {
@@ -257,7 +257,7 @@ impl Server {
})?;
let mut c = self.catalog.lock().unwrap();
c.mark_compacted(table)?;
c.rebuild_stats(table, rows);
c.rebuild_stats(table, rows)?;
c.save(self.catalog_path())
}
WalOp::Accounts { accounts, removed, migrate_roles } => {
@@ -287,20 +287,20 @@ impl Server {
}
let mut c = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
declare_system_schema(&mut c, &rows)?;
c.rebuild_stats(IDENTITIES_TABLE, &identities);
c.rebuild_stats(IDENTITIES_TABLE, &identities)?;
c.set_columns(IDENTITIES_TABLE, identity_columns())?;
c.rebuild_stats(IDENTITIES_TABLE, &identities);
c.rebuild_stats(IDENTITIES_TABLE, &identities)?;
c.mark_system(IDENTITIES_TABLE)?;
c.add_index(IDENTITIES_TABLE, "user")?;
c.identity_high_water = c.identity_high_water.max(accounts.iter().map(|a| a.id).max().unwrap_or(0));
if has_roles && *migrate_roles { c.rebuild_stats(legacy_roles, &[]); }
for (table, records) in cleaned { c.rebuild_stats(table, &records); }
if has_roles && *migrate_roles { c.rebuild_stats(legacy_roles, &[])?; }
for (table, records) in cleaned { c.rebuild_stats(table, &records)?; }
c.save(self.catalog_path())
}
WalOp::Insert { table, row } => {
self.with_table(table, |t| t.insert_row(row))?;
let mut c = self.catalog.lock().unwrap();
if c.record_insert(table, row) {
if c.record_insert(table, row)? {
c.save(self.catalog_path())?;
}
Ok(())
@@ -310,7 +310,7 @@ impl Server {
let mut c = self.catalog.lock().unwrap();
let mut changed = false;
for r in rows {
if c.record_insert(table, r) {
if c.record_insert(table, r)? {
changed = true;
}
}
@@ -328,17 +328,17 @@ impl Server {
WalOp::ReplaceAll { table, rows } => {
self.with_table(table, |t| t.replace_all(rows))?;
let mut c = self.catalog.lock().unwrap();
c.rebuild_stats(table, rows);
c.rebuild_stats(table, rows)?;
c.save(self.catalog_path())
}
WalOp::DropColumn { table, column, rows } => {
WalOp::DropColumn { table, column, rows, schema_version } => {
self.with_table(table, |t| {
t.detach_index(column);
t.replace_all(rows)
})?;
let mut c = self.catalog.lock().unwrap();
c.remove_column(table, column)?;
c.rebuild_stats(table, rows);
c.remove_column_version(table, column, *schema_version)?;
c.rebuild_stats(table, rows)?;
c.save(self.catalog_path())?;
drop(c);
let _ = std::fs::remove_file(self.index_path(table, column));
@@ -374,7 +374,7 @@ impl Server {
}
WalOp::CreateIndex { table, column } => self.apply_create_index(table, column),
WalOp::DropIndex { table, column } => self.apply_drop_index(table, column),
WalOp::ReplaceSchema { table, columns, rows } => {
WalOp::ReplaceSchema { table, columns, rows, schema_version } => {
let old_indexes: Vec<String> = {
let c = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
c.describe(table)
@@ -389,8 +389,8 @@ impl Server {
t.replace_all(rows)
})?;
let mut c = self.catalog.lock().unwrap();
c.set_columns(table, columns.clone())?;
c.rebuild_stats(table, rows);
c.set_columns_version(table, columns.clone(), *schema_version)?;
c.rebuild_stats(table, rows)?;
c.save(self.catalog_path())?;
drop(c);
for col in old_indexes {
@@ -483,7 +483,7 @@ impl Server {
match op {
WalOp::Insert { table, row } => validate(table, std::slice::from_ref(row))?,
WalOp::InsertBatch { table, rows } | WalOp::ReplaceAll { table, rows } => validate(table, rows)?,
WalOp::HideColumn { table, column }
WalOp::HideColumn { table, column, .. }
if !c.describe(table).is_some_and(|s| s.columns.iter().any(|col| &col.name == column)) =>
{
return Err(std::io::Error::other(format!("unknown column: {column}")));
@@ -554,6 +554,29 @@ impl Server {
}
}
}
if ops.iter().any(|op| matches!(op, WalOp::HideColumn { .. } | WalOp::DropColumn { .. } | WalOp::ReplaceSchema { .. })) {
let mut preview = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?.clone();
for op in ops.to_mut() {
match op {
WalOp::HideColumn { table, column, schema_version }
| WalOp::DropColumn { table, column, schema_version, .. } => {
preview.remove_column(table, column)?;
*schema_version = Some(preview.describe(table).ok_or_else(|| std::io::Error::other("missing schema version"))?.schema_version.max(1));
}
WalOp::ReplaceSchema { table, columns, schema_version, .. } => {
preview.set_columns(table, columns.clone())?;
*schema_version = Some(preview.describe(table).ok_or_else(|| std::io::Error::other("missing schema version"))?.schema_version.max(1));
}
WalOp::CreateTable { table, columns, object_id } => preview.create_table_with_id(table, columns.clone(), *object_id)?,
WalOp::DropTable { table } => preview.drop_table(table)?,
WalOp::Insert { table, row } => preview.ensure_columns(table, row)?,
WalOp::InsertBatch { table, rows } | WalOp::ReplaceAll { table, rows } => {
for row in rows { preview.ensure_columns(table, row)?; }
}
_ => {}
}
}
}
let commit_lsn = wal.append_group(&ops)?;
self.recovery_required.store(true, Ordering::Release);
for op in ops.iter() {
@@ -683,9 +706,10 @@ impl Server {
});
Ok(!present)
}
WalOp::HideColumn { table, column } => {
WalOp::HideColumn { table, column, schema_version } => {
let c = self.catalog.lock().unwrap();
Ok(c.describe(table).is_some_and(|s| !s.columns.iter().any(|col| &col.name == column)))
Ok(c.describe(table).is_some_and(|s| !s.columns.iter().any(|col| &col.name == column)
&& schema_version.is_none_or(|version| s.schema_version == version)))
}
WalOp::CreateTable { table, object_id, .. } => {
let catalog = self.catalog.lock().map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
@@ -739,7 +763,7 @@ impl Server {
}
let rows = self.with_table(&table, |t| t.scan())?;
let mut c = self.catalog.lock().unwrap();
c.rebuild_stats(&table, &rows);
c.rebuild_stats(&table, &rows)?;
touched = true;
}
if touched {
@@ -791,7 +815,7 @@ impl Server {
.catalog
.lock()
.map_err(|_| std::io::Error::other("catalog lock poisoned"))?;
if catalog.normalize_types() {
if catalog.normalize_types()? {
catalog.save(self.catalog_path())?;
}
Ok(())
@@ -954,6 +978,7 @@ impl Server {
let op = WalOp::HideColumn {
table: table.to_string(),
column: column.to_string(),
schema_version: None,
};
self.write_via_wal(&op)
}
@@ -1310,6 +1335,32 @@ impl Server {
}
}
/// 持读门时复查绑定身份、版本和列定义
fn validate_bound_schema(&self, table: &str, expected: &crate::protocol::BoundSchema) -> std::io::Result<()> {
use std::io::Error;
if self.recovery_required.load(Ordering::Acquire) { return Err(Error::other("storage recovery required")); }
if blocked_table(table) { return Err(Error::other("reserved system table")); }
let catalog = self.catalog.lock().map_err(|_| Error::other("catalog lock poisoned"))?;
let actual = catalog.describe(table).ok_or_else(|| Error::other(format!("bound query schema changed: {}", expected.table)))?;
match (&expected.database_id, &expected.object_id) {
(Some(database), Some(object)) => {
if parse_bound_number(database)? != catalog.database_id || parse_bound_number(object)? != actual.object_id {
return Err(Error::other(format!("bound query object changed: {}", expected.table)));
}
}
(None, None) => {}
_ => return Err(Error::other("incomplete bound object identity")),
}
if let Some(version) = &expected.schema_version
&& parse_bound_number(version)? != actual.schema_version {
return Err(Error::other(format!("bound query schema changed: {}", expected.table)));
}
if expected.columns != actual.columns {
return Err(Error::other(format!("bound query schema changed: {}", expected.table)));
}
Ok(())
}
/// 持门后分发一条请求
fn dispatch(&self, req: Request) -> Response {
if self.recovery_required.load(Ordering::Acquire) && !matches!(req, Request::Ping) {
@@ -1558,6 +1609,7 @@ impl Server {
table: name.to_string(),
database_id: c.database_id,
object_id: ts.object_id,
schema_version: ts.schema_version,
columns: ts.columns.clone(),
row_count: ts.row_count,
indexes: index_wires(ts),
@@ -1571,6 +1623,7 @@ impl Server {
log_debug!(request, "describe_table table={}", table);
match self.describe_table(&table) {
Ok((cols, count, indexes, stats)) => Response::Schema {
schema_version: self.catalog.lock().unwrap().describe(&table).map_or(0, |s| s.schema_version),
table,
columns: cols,
row_count: count,
@@ -1713,7 +1766,7 @@ impl Server {
Request::ReplaceSchema { table, columns, rows } => {
log_debug!(request, "replace_schema table={} cols={}", table, columns.len());
let tname = table.clone();
let op = WalOp::ReplaceSchema { table, columns, rows };
let op = WalOp::ReplaceSchema { table, columns, rows, schema_version: None };
match self.write_via_wal(&op) {
Ok(()) => Response::Ok,
Err(e) => {
@@ -1838,9 +1891,9 @@ fn identity_columns() -> Vec<SchemaColumn> {
/// 把系统表写进字典
fn declare_system_schema(catalog: &mut Catalog, rows: &[Row]) -> std::io::Result<()> {
catalog.rebuild_stats(USERS_TABLE, rows);
catalog.rebuild_stats(USERS_TABLE, rows)?;
catalog.set_columns(USERS_TABLE, users_columns())?;
catalog.rebuild_stats(USERS_TABLE, rows);
catalog.rebuild_stats(USERS_TABLE, rows)?;
catalog.mark_system(USERS_TABLE)?;
Ok(())
}
@@ -2137,6 +2190,10 @@ impl Databases {
fn request(&self, mut value: serde_json::Value) -> std::io::Result<Response> {
use std::io::Error;
let method = value.get("method").and_then(|v| v.as_str()).unwrap_or("").to_string();
if value.get("schema_guard").is_some()
&& !matches!(method.as_str(), "scan" | "scan_shard" | "lookup_by_index" | "range_by_index") {
return Err(Error::other("schema guard requires a table read"));
}
let name = match value.get("database") {
None => String::new(),
Some(serde_json::Value::String(s)) => s.to_ascii_lowercase(),
@@ -2258,6 +2315,8 @@ impl Databases {
selected = database.to_ascii_lowercase();
value["table"] = serde_json::Value::String(table.to_string());
}
let guard: Option<Vec<crate::protocol::BoundSchema>> = value.get("schema_guard")
.map(|guard| serde_json::from_value(guard.clone()).map_err(Error::other)).transpose()?;
let request: Request = serde_json::from_value(value).map_err(Error::other)?;
let global = matches!(request, Request::AccountsList | Request::CatalogManage { .. } | Request::AccountCreate { .. }
| Request::IdentityInitialize { .. } | Request::RoleCreate { .. } | Request::IdentityAlter { .. }
@@ -2268,11 +2327,43 @@ impl Databases {
if !global && selected.is_empty() { return Err(Error::other("no database selected")); }
let server = if global || selected == SYSTEM_DATABASE { Arc::clone(&self.system) }
else { Arc::clone(named.get(&selected).ok_or_else(|| Error::other("unknown database"))?) };
if let Some(bindings) = guard {
if !concurrent_read(&request) { return Err(Error::other("schema guard requires a table read")); }
let mut servers = std::collections::BTreeMap::from([(selected.clone(), Arc::clone(&server))]);
let mut dependencies = Vec::new();
for binding in bindings {
let (database, table) = binding.table.split_once('.').unwrap_or((&selected, &binding.table));
let database = database.to_ascii_lowercase();
if !valid_database(&database) { return Err(Error::other("invalid bound database name")); }
let dependency = if database == SYSTEM_DATABASE { Arc::clone(&self.system) }
else { Arc::clone(named.get(&database).ok_or_else(|| Error::other("bound query database changed"))?) };
servers.insert(database.clone(), dependency);
dependencies.push((database, table.to_owned(), binding));
}
if !dependencies.iter().any(|(database, table, _)| database == &selected && Some(table.as_str()) == request.table()) {
return Err(Error::other("bound read table has no schema guard"));
}
drop(named);
let _gates = servers.values().map(|server| server.gate.read()
.map_err(|_| Error::other("storage read gate poisoned"))).collect::<std::io::Result<Vec<_>>>()?;
for (database, table, binding) in &dependencies {
servers[database].validate_bound_schema(table, binding)?;
}
return Ok(Response::Guarded { response: Box::new(server.dispatch(request)) });
}
drop(named);
Ok(server.handle_request(request))
}
}
/// 读取守卫中的精确十进制编号
fn parse_bound_number(value: &str) -> std::io::Result<u64> {
if value.is_empty() || value.len() > 20 || !value.bytes().all(|c| c.is_ascii_digit()) {
return Err(std::io::Error::other("invalid bound catalog number"));
}
value.parse().map_err(std::io::Error::other)
}
/// 只读表页的请求可与其它读并发，其余独占
fn concurrent_read(req: &Request) -> bool {
matches!(
chusql-core/storage/src/wal.rs
+31 -7
View File
@@ -29,7 +29,7 @@ const OP_DROP_INDEX: u8 = 14;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum WalOp {
HideColumn { table: String, column: String },
HideColumn { table: String, column: String, schema_version: Option<u64> },
Compact { table: String, rows: Vec<Row> },
Accounts { accounts: Vec<Account>, removed: Vec<String>, migrate_roles: bool },
Insert {
@@ -51,11 +51,13 @@ pub enum WalOp {
DropColumn {
table: String,
column: String,
schema_version: Option<u64>,
rows: Vec<Row>,
},
ReplaceSchema {
table: String,
columns: Vec<SchemaColumn>,
schema_version: Option<u64>,
rows: Vec<Row>,
},
CreateTable {
@@ -78,15 +80,26 @@ pub enum WalOp {
Commit,
}
#[derive(Debug, Serialize, Deserialize)]
struct HideColumnPayload {
column: String,
#[serde(default)]
schema_version: Option<u64>,
}
#[derive(Debug, Serialize, Deserialize)]
struct DropColumnPayload {
column: String,
#[serde(default)]
schema_version: Option<u64>,
rows: Vec<Row>,
}
#[derive(Debug, Serialize, Deserialize)]
struct ReplaceSchemaPayload {
columns: Vec<SchemaColumn>,
#[serde(default)]
schema_version: Option<u64>,
rows: Vec<Row>,
}
@@ -379,8 +392,8 @@ fn truncate(f: &mut File) -> io::Result<()> {
/// 把操作编成字节
fn encode(op: &WalOp, lsn: u64) -> io::Result<Vec<u8>> {
let (op_type, table, payload) = match op {
WalOp::HideColumn { table, column } => (OP_HIDE_COLUMN, table.clone(),
serde_json::to_vec(column).map_err(io::Error::other)?),
WalOp::HideColumn { table, column, schema_version } => (OP_HIDE_COLUMN, table.clone(),
serde_json::to_vec(&serde_json::json!({"column":column,"schema_version":schema_version})).map_err(io::Error::other)?),
WalOp::Compact { table, rows } => {
let p = serde_json::to_vec(rows).map_err(io::Error::other)?;
(OP_COMPACT, table.clone(), p)
@@ -403,17 +416,19 @@ fn encode(op: &WalOp, lsn: u64) -> io::Result<Vec<u8>> {
let p = serde_json::to_vec(rows).map_err(io::Error::other)?;
(OP_REPLACE_ALL, table.clone(), p)
}
WalOp::DropColumn { table, column, rows } => {
WalOp::DropColumn { table, column, rows, schema_version } => {
let p = serde_json::to_vec(&DropColumnPayload {
column: column.clone(),
schema_version: *schema_version,
rows: rows.clone(),
})
.map_err(io::Error::other)?;
(OP_DROP_COLUMN, table.clone(), p)
}
WalOp::ReplaceSchema { table, columns, rows } => {
WalOp::ReplaceSchema { table, columns, rows, schema_version } => {
let p = serde_json::to_vec(&ReplaceSchemaPayload {
columns: columns.clone(),
schema_version: *schema_version,
rows: rows.clone(),
})
.map_err(io::Error::other)?;
@@ -474,9 +489,16 @@ fn decode_body(body: &[u8]) -> Result<(u64, WalOp), String> {
let op = match op_type {
OP_COMMIT => WalOp::Commit,
OP_HIDE_COLUMN => {
let column = serde_json::from_slice(payload)
let value: serde_json::Value = serde_json::from_slice(payload)
.map_err(|e| format!("bad hide_column payload: {e}"))?;
WalOp::HideColumn { table, column }
let (column, schema_version) = if let Some(column) = value.as_str() {
(column.to_string(), None)
} else {
let p: HideColumnPayload = serde_json::from_value(value)
.map_err(|e| format!("bad hide_column payload: {e}"))?;
(p.column, p.schema_version)
};
WalOp::HideColumn { table, column, schema_version }
}
OP_COMPACT => {
let rows: Vec<Row> = serde_json::from_slice(payload)
@@ -521,6 +543,7 @@ fn decode_body(body: &[u8]) -> Result<(u64, WalOp), String> {
WalOp::DropColumn {
table,
column: p.column,
schema_version: p.schema_version,
rows: p.rows,
}
}
@@ -530,6 +553,7 @@ fn decode_body(body: &[u8]) -> Result<(u64, WalOp), String> {
WalOp::ReplaceSchema {
table,
columns: p.columns,
schema_version: p.schema_version,
rows: p.rows,
}
}
chusql-core/storage/tests/catalog_test.rs
+78 -16
View File
@@ -4,6 +4,68 @@ use serde_json::json;
// 数据字典测试：补列、去重、未知表。
/// 结构变化递增版本，数据及重复应用不变
#[test]
fn schema_versions_track_structure_and_survive_reload() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("catalog.json");
let mut c = Catalog::default();
c.create_table("t", vec![]).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 1);
let values = row(&[("id", json!(1)), ("name", json!("a"))]);
c.record_insert("t", &values).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 2);
c.rebuild_stats("t", &[values]).unwrap();
let columns = c.describe("t").unwrap().columns.clone();
c.set_columns("t", columns).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 2);
c.remove_column("t", "name").unwrap();
c.remove_column("t", "name").unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 3);
c.rename_table("t", "renamed").unwrap();
c.save(&path).unwrap();
let c = Catalog::load(&path).unwrap();
assert_eq!(c.describe("renamed").unwrap().schema_version, 4);
}
/// 旧目录版本未知，首个结构变更才递增
#[test]
fn legacy_schema_version_remains_unknown_until_change() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("catalog.json");
std::fs::write(&path, json!({"tables": {
"t": {"object_id": 42, "columns": []}
}}).to_string()).unwrap();
let mut c = Catalog::load(&path).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 0);
c.save(&path).unwrap();
c = Catalog::load(&path).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 0);
c.ensure_columns("t", &row(&[("id", json!(1))])).unwrap();
assert_eq!(c.describe("t").unwrap().schema_version, 1);
assert_eq!(c.describe("t").unwrap().object_id, 42);
}
/// 版本耗尽拒绝结构变更并保留原定义
#[test]
fn exhausted_schema_version_preserves_structure() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("catalog.json");
std::fs::write(&path, json!({"tables": {
"t": {"object_id": 42, "schema_version": u64::MAX,
"columns": [{"name": "id", "ty": "int"}]}
}}).to_string()).unwrap();
let mut c = Catalog::load(&path).unwrap();
assert!(c.remove_column("t", "id").is_err());
assert!(c.set_columns("t", vec![]).is_err());
assert!(c.rename_table("t", "other").is_err());
assert!(c.ensure_columns("t", &row(&[("new", json!(1))])).is_err());
assert_eq!(c.describe("t").unwrap().columns.len(), 1);
assert_eq!(c.describe("t").unwrap().schema_version, u64::MAX);
assert!(c.describe("other").is_none());
c.record_insert("t", &row(&[("id", json!(1))])).unwrap();
}
/// 表删除重建和重载均保留编号规则
#[test]
fn object_ids_survive_reload_and_are_not_reused() {
@@ -80,7 +142,7 @@ fn ensure_then_describe() {
let path = dir.path().join("catalog.json");
let mut c = Catalog::load(&path).unwrap();
c.ensure_columns("users", &row(&[("id", json!(1)), ("name", json!("Alice"))]));
c.ensure_columns("users", &row(&[("id", json!(1)), ("name", json!("Alice"))])).unwrap();
c.save(&path).unwrap();
let c2 = Catalog::load(&path).unwrap();
@@ -102,8 +164,8 @@ fn second_insert_does_not_duplicate_columns() {
let path = dir.path().join("catalog.json");
let mut c = Catalog::load(&path).unwrap();
c.ensure_columns("users", &row(&[("id", json!(1))]));
c.ensure_columns("users", &row(&[("id", json!(2))]));
c.ensure_columns("users", &row(&[("id", json!(1))])).unwrap();
c.ensure_columns("users", &row(&[("id", json!(2))])).unwrap();
assert_eq!(c.describe("users").unwrap().columns.len(), 1);
}
@@ -114,9 +176,9 @@ fn remove_column_drops_definition_stats_and_index() {
let path = dir.path().join("catalog.json");
let mut c = Catalog::load(&path).unwrap();
c.ensure_columns("t", &row(&[("id", json!(1)), ("code", json!(5))]));
c.ensure_columns("t", &row(&[("id", json!(1)), ("code", json!(5))])).unwrap();
c.add_index("t", "code").unwrap();
c.rebuild_stats("t", &[row(&[("id", json!(1)), ("code", json!(5))])]);
c.rebuild_stats("t", &[row(&[("id", json!(1)), ("code", json!(5))])]).unwrap();
let removed = c.remove_column("t", "code").unwrap();
assert!(removed, "first remove should hit the column");
@@ -142,7 +204,7 @@ fn stats_track_a_histogram_for_numeric_columns() {
let rows: Vec<Row> = (0..100)
.map(|n| row(&[("id", json!(n)), ("name", json!(format!("u{}", n)))]))
.collect();
c.rebuild_stats("t", &rows);
c.rebuild_stats("t", &rows).unwrap();
let s = c.describe("t").unwrap();
let id = s.stats.get("id").unwrap();
@@ -155,7 +217,7 @@ fn stats_track_a_histogram_for_numeric_columns() {
assert!(name.hist.is_empty(), "字符串列不进直方图");
assert_eq!(name.lo, None);
c.record_insert("t", &row(&[("id", json!(1000)), ("name", json!("x"))]));
c.record_insert("t", &row(&[("id", json!(1000)), ("name", json!("x"))])).unwrap();
let s = c.describe("t").unwrap();
let id = s.stats.get("id").unwrap();
assert_eq!(id.lo, Some(0.0));
@@ -178,7 +240,7 @@ fn stats_track_a_histogram_for_numeric_columns() {
fn duplicate_deletion_keeps_distinct_until_last_occurrence() {
let mut catalog = Catalog::default();
let duplicate = row(&[("id", json!(1)), ("v", json!(7))]);
catalog.rebuild_stats("t", &[duplicate.clone(), duplicate.clone()]);
catalog.rebuild_stats("t", &[duplicate.clone(), duplicate.clone()]).unwrap();
catalog.record_delete("t", std::slice::from_ref(&duplicate));
assert_eq!(catalog.describe("t").unwrap().stats["v"].distinct, 1);
catalog.record_delete("t", &[duplicate]);
@@ -190,7 +252,7 @@ fn duplicate_deletion_keeps_distinct_until_last_occurrence() {
fn capped_distinct_keeps_histogram_and_other_columns_accurate() {
let mut catalog = Catalog::default();
let rows: Vec<Row> = (0..5000).map(|n| row(&[("id", json!(n)), ("v", json!(n % 2))])).collect();
catalog.rebuild_stats("t", &rows);
catalog.rebuild_stats("t", &rows).unwrap();
let schema = catalog.describe("t").unwrap();
assert!(schema.stats["id"].capped);
assert_eq!(schema.stats["id"].hist.iter().sum::<u64>(), 5000);
@@ -199,7 +261,7 @@ fn capped_distinct_keeps_histogram_and_other_columns_accurate() {
let schema = catalog.describe("t").unwrap();
assert_eq!(schema.stats["id"].hist.iter().sum::<u64>(), 4000);
assert_eq!(schema.stats["v"].distinct, 2);
catalog.rebuild_stats("t", &rows[1000..]);
catalog.rebuild_stats("t", &rows[1000..]).unwrap();
assert_eq!(catalog.describe("t").unwrap().stats["id"].distinct, 4000);
}
@@ -208,16 +270,16 @@ fn capped_distinct_keeps_histogram_and_other_columns_accurate() {
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);
forward.rebuild_stats("t", &rows).unwrap();
let mut reverse = Catalog::default();
let reversed: Vec<Row> = rows.iter().rev().cloned().collect();
reverse.rebuild_stats("t", &reversed);
reverse.rebuild_stats("t", &reversed).unwrap();
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_insert("t", &row(&[("v", json!(10000))])).unwrap();
forward.record_delete("t", &[row(&[("v", json!(10000))])]);
forward.rebuild_stats("t", &rows[80..]);
forward.rebuild_stats("t", &rows[80..]).unwrap();
let repaired = &forward.describe("t").unwrap().stats["v"];
assert_eq!(repaired.lo, Some(80.0));
assert_eq!(repaired.hi, Some(159.0));
@@ -253,8 +315,8 @@ fn legacy_catalog_loads_and_normalizes() {
let s = c.describe("legacy").unwrap();
assert_eq!(s.columns.len(), 2);
assert!(s.columns[0].nullable, "老列默认可空");
assert!(c.normalize_types());
assert!(c.normalize_types().unwrap());
let types: Vec<&str> = c.describe("legacy").unwrap().columns.iter().map(|col| col.ty.as_str()).collect();
assert_eq!(types, vec!["int", "str"]);
assert!(!c.normalize_types(), "第二次没有可改的");
assert!(!c.normalize_types().unwrap(), "第二次没有可改的");
}
chusql-core/storage/tests/ffi_test.rs
+91
View File
@@ -0,0 +1,91 @@
use std::ffi::CString;
use std::ptr;
use std::slice;
use chusql_core_storage::ffi::*;
use chusql_core_storage::runtime::Storage;
// 验证 FFI 错误输出、长度边界与响应所有权。
/// 读取当前线程的 FFI 错误
fn last_error() -> String {
let mut len = 0;
// SAFETY: 长度可写，返回缓冲在复制前没有被修改。
unsafe {
let data = chusql_storage_last_error(&mut len);
assert!(!data.is_null());
String::from_utf8(slice::from_raw_parts(data, len).to_vec()).unwrap()
}
}
/// 维护失败清空输出并拒绝超长请求
#[test]
fn maintenance_rejects_invalid_arguments_and_resets_outputs() {
let input = b"{}";
let mut out = input.as_ptr() as *mut u8;
let mut len = 99;
// SAFETY: 输出均可写；无效输入应在任何读取之前被拒绝。
unsafe {
assert_eq!(chusql_storage_maintenance(ptr::null(), ptr::null(), 0, &mut out, &mut len), -1);
assert!(out.is_null());
assert_eq!(len, 0);
len = 99;
assert_eq!(chusql_storage_maintenance(ptr::null(), input.as_ptr(), input.len(), ptr::null_mut(), &mut len), -1);
assert_eq!(len, 0);
out = input.as_ptr() as *mut u8;
assert_eq!(chusql_storage_maintenance(ptr::null(), input.as_ptr(), input.len(), &mut out, ptr::null_mut()), -1);
assert!(out.is_null());
assert_eq!(chusql_storage_maintenance(ptr::null(), input.as_ptr(), usize::MAX, &mut out, &mut len), -1);
}
assert!(out.is_null());
assert_eq!(len, 0);
assert_eq!(last_error(), "invalid maintenance argument");
}
/// 请求验证长度和编码并归还响应
#[test]
fn request_validates_input_and_returns_owned_response() -> Result<(), Box<dyn std::error::Error>> {
let temp = tempfile::tempdir()?;
let config = temp.path().join("settings.toml");
std::fs::write(&config, format!("[storage]\ndata_dir = '{}'\nlog_files = '{}'\n[log]\nlevel = 'off'\n", temp.path().join("data").display(), temp.path().join("logs").display()))?;
let storage = Storage::open(config.to_str())?;
let input = br#"{"method":"ping"}"#;
let mut out = ptr::null_mut();
let mut len = 0;
// SAFETY: 句柄和实际请求有效；超长长度必须在构造切片前被拒绝。
unsafe {
assert_eq!(chusql_storage_request(&storage, input.as_ptr(), usize::MAX, &mut out, &mut len), -1);
assert!(out.is_null());
assert_eq!(len, 0);
assert_eq!(last_error(), "invalid argument");
assert_eq!(chusql_storage_request(&storage, [0xff].as_ptr(), 1, &mut out, &mut len), -1);
assert!(last_error().starts_with("request is not valid UTF-8:"));
assert_eq!(chusql_storage_request(&storage, input.as_ptr(), input.len(), &mut out, &mut len), 0);
assert!(!out.is_null());
let response = slice::from_raw_parts(out, len).to_vec();
chusql_storage_free(out, len);
let response: serde_json::Value = serde_json::from_slice(&response)?;
assert_eq!(response["status"], "pong");
}
drop(storage);
let config = CString::new(config.to_str().unwrap())?;
// SAFETY: 路径与输出有效；失败不产生需要释放的缓冲。
unsafe {
assert_eq!(chusql_storage_maintenance(config.as_ptr(), [0xff].as_ptr(), 1, &mut out, &mut len), -1);
}
assert!(out.is_null());
assert_eq!(len, 0);
assert!(!last_error().is_empty());
let maintenance = br#"{"mode":"reset","user":"root","password_hash":"hash","types":[{"name":"int","base_type":"int","parameterized":false}]}"#;
// SAFETY: 维护仅使用本测试临时目录；成功响应复制后由配套函数释放。
unsafe {
assert_eq!(chusql_storage_maintenance(config.as_ptr(), maintenance.as_ptr(), maintenance.len(), &mut out, &mut len), 0, "{}", last_error());
assert!(!out.is_null());
let response = slice::from_raw_parts(out, len).to_vec();
chusql_storage_free(out, len);
let response: serde_json::Value = serde_json::from_slice(&response)?;
assert_eq!(response["status"], "maintenance");
assert_eq!(response["mode"], "reset");
}
Ok(())
}
chusql-core/storage/tests/runtime_test.rs
+126 -1
View File
@@ -171,6 +171,131 @@ fn object_id_survives_committed_create_recovery() -> Result<(), Box<dyn std::err
Ok(())
}
/// 读取复查跨库依赖而非只复查扫描表
#[test]
fn bound_reads_recheck_dependencies_in_other_databases() -> Result<(), Box<dyn std::error::Error>> {
let (_server, _data) = start_server();
let mut c = connect()?;
request_ok(&mut c, serde_json::json!({"method":"create_table","table":"source","columns":[]}))?;
request_ok(&mut c, serde_json::json!({"method":"create_database","database":"dependency"}))?;
request_ok(&mut c, serde_json::json!({"method":"create_table","database":"dependency","table":"related","columns":[]}))?;
let catalog: serde_json::Value = serde_json::from_str(&send_raw(&mut c, r#"{"method":"all_catalogs"}"#))?;
let bindings: Vec<_> = catalog["schemas"].as_array().unwrap().iter().filter(|entry|
entry["table"] == "main.source" || entry["table"] == "dependency.related").map(|entry| {
let mut binding = entry.clone();
binding["database_id"] = binding["database_id"].as_u64().unwrap().to_string().into();
binding["object_id"] = binding["object_id"].as_u64().unwrap().to_string().into();
binding
}).collect();
assert_eq!(bindings.len(), 2);
let request = serde_json::json!({"method":"scan","table":"source","schema_guard":bindings});
let valid: serde_json::Value = serde_json::from_str(&send(&mut c, &request.to_string()))?;
assert_eq!(valid["status"], "guarded");
request_ok(&mut c, serde_json::json!({"method":"replace_schema","database":"dependency","table":"related","columns":[{"name":"new","ty":"int"}],"rows":[]}))?;
let invalid: serde_json::Value = serde_json::from_str(&send(&mut c, &request.to_string()))?;
assert_eq!(invalid["status"], "error");
assert!(invalid["message"].as_str().unwrap().contains("bound query schema changed: dependency.related"));
Ok(())
}
/// 每种读取都复查守卫，拒绝旧版本和写入
#[test]
fn bound_reads_validate_dependencies_and_acknowledge_guards() -> Result<(), Box<dyn std::error::Error>> {
let (_server, _data) = start_server();
let mut c = connect()?;
request_ok(&mut c, serde_json::json!({"method":"create_table","table":"guarded","columns":[{"name":"id","ty":"int"}]}))?;
request_ok(&mut c, serde_json::json!({"method":"insert","table":"guarded","row":{"id":7}}))?;
let catalog: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"list_catalog"}"#))?;
let mut binding = catalog["schemas"][0].clone();
binding["database_id"] = binding["database_id"].as_u64().unwrap().to_string().into();
binding["object_id"] = binding["object_id"].as_u64().unwrap().to_string().into();
for mut request in [
serde_json::json!({"method":"scan","table":"guarded"}),
serde_json::json!({"method":"scan_shard","table":"guarded","shard":0,"shards":1}),
serde_json::json!({"method":"lookup_by_index","table":"guarded","column":"id","key":7}),
serde_json::json!({"method":"range_by_index","table":"guarded","column":"id","lo":null,"hi":null}),
] {
request["schema_guard"] = serde_json::json!([binding]);
let response: serde_json::Value = serde_json::from_str(&send(&mut c, &request.to_string()))?;
assert_eq!(response["status"], "guarded", "{response}");
assert!(response["response"]["status"] == "rows" || response["response"]["status"] == "no_index", "{response}");
}
request_ok(&mut c, serde_json::json!({"method":"replace_schema","table":"guarded","columns":[{"name":"id","ty":"int"},{"name":"extra","ty":"str"}],"rows":[{"id":7,"extra":"x"}]}))?;
let stale: serde_json::Value = serde_json::from_str(&send(&mut c, &serde_json::json!({"method":"scan","table":"guarded","schema_guard":[binding]}).to_string()))?;
assert_eq!(stale["status"], "error");
assert!(stale["message"].as_str().unwrap().contains("bound query schema changed"));
let write: serde_json::Value = serde_json::from_str(&send(&mut c, &serde_json::json!({"method":"insert","table":"guarded","row":{"id":9},"schema_guard":[binding]}).to_string()))?;
assert_eq!(write["status"], "error");
let management: serde_json::Value = serde_json::from_str(&send_raw(&mut c, &serde_json::json!({"method":"create_database","database":"blocked","schema_guard":[]}).to_string()))?;
assert_eq!(management["status"], "error");
let databases: serde_json::Value = serde_json::from_str(&send_raw(&mut c, r#"{"method":"list_databases"}"#))?;
assert!(!databases["tables"].as_array().unwrap().iter().any(|database| database == "blocked"));
let rows: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"scan","table":"guarded"}"#))?;
assert_eq!(rows["rows"].as_array().unwrap().len(), 1);
Ok(())
}
/// 描述与目录均透传精确版本字符串
#[test]
fn catalog_schema_versions_are_exact_and_validated() -> Result<(), Box<dyn std::error::Error>> {
use chusql_core_storage::protocol::TableSchemaWire;
let (_server, _data) = start_server();
let mut c = connect()?;
request_ok(&mut c, serde_json::json!({"method":"create_table","table":"versioned","columns":[{"name":"id","ty":"int"}]}))?;
let describe: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"describe_table","table":"versioned"}"#))?;
assert_eq!(describe["schema_version"], "1");
let catalog: serde_json::Value = serde_json::from_str(&send(&mut c, r#"{"method":"list_catalog"}"#))?;
assert_eq!(catalog["schemas"][0]["schema_version"], "1");
let wire: TableSchemaWire = serde_json::from_value(serde_json::json!({"table":"t","columns":[],"schema_version":u64::MAX.to_string()}))?;
assert_eq!(wire.schema_version, u64::MAX);
assert_eq!(serde_json::to_value(wire)?["schema_version"], u64::MAX.to_string());
for version in [serde_json::json!(-1), serde_json::json!(1.5), serde_json::json!("18446744073709551616"), serde_json::json!("+1")] {
assert!(serde_json::from_value::<TableSchemaWire>(serde_json::json!({"table":"t","columns":[],"schema_version":version})).is_err());
}
Ok(())
}
/// 检查点失败及反复恢复保留结构版本
#[test]
fn schema_versions_survive_checkpoint_failure_and_repeated_recovery() -> Result<(), Box<dyn std::error::Error>> {
use chusql_core_storage::catalog::Catalog;
use chusql_core_storage::wal::{Wal, WalOp};
let (server, data) = start_server();
let mut c = connect()?;
request_ok(&mut c, serde_json::json!({"method":"create_table","table":"versioned","columns":[{"name":"id","ty":"int"}]}))?;
let directory = data.path().join("databases/main");
let pending = directory.join("wal.checkpoint.pending");
std::fs::create_dir(&pending)?;
let response = send(&mut c, r#"{"method":"replace_schema","table":"versioned","columns":[{"name":"id","ty":"int"},{"name":"tag","ty":"str"}],"rows":[]}"#);
assert!(response.contains("error"), "{response}");
let catalog = Catalog::load(directory.join("catalog.json"))?;
let version = catalog.describe("versioned").unwrap().schema_version;
assert_eq!(version, 2);
let wal = Wal::new(directory.join("wal.log"));
let records = wal.read_all()?.records;
assert!(records.iter().any(|(_, op)| matches!(op, WalOp::ReplaceSchema { schema_version: Some(2), .. })));
let columns = catalog.describe("versioned").unwrap().columns.clone();
drop(wal);
drop(c);
drop(server);
std::fs::remove_dir(pending)?;
for _ in 0..2 {
let wal = Wal::new(directory.join("wal.log"));
wal.append_group(&[
WalOp::ReplaceSchema { table: "versioned".into(), columns: vec![], rows: vec![], schema_version: Some(1) },
WalOp::ReplaceSchema { table: "versioned".into(), columns: columns.clone(), rows: vec![], schema_version: Some(version) },
])?;
drop(wal);
let restarted = restart_at(data.path())?;
let catalog = Catalog::load(directory.join("catalog.json"))?;
assert_eq!(catalog.describe("versioned").unwrap().schema_version, version);
assert_eq!(catalog.describe("versioned").unwrap().columns, columns);
assert_eq!(std::fs::metadata(directory.join("wal.log"))?.len(), 0);
drop(restarted);
}
Ok(())
}
/// 并发建表分配互不重复的编号
#[test]
fn concurrent_table_creation_allocates_distinct_object_ids() -> Result<(), Box<dyn std::error::Error>> {
@@ -1026,7 +1151,7 @@ fn hidden_column_wal_replays_twice_without_touching_heap() -> Result<(), Box<dyn
let mut compacted: Option<Vec<u8>> = None;
for round in 0..2 {
let wal = Wal::new(data.path().join("databases/main/wal.log"));
let op = WalOp::HideColumn { table: "items".into(), column: "secret".into() };
let op = WalOp::HideColumn { table: "items".into(), column: "secret".into(), schema_version: None };
wal.truncate()?;
wal.append(&op)?;
assert_eq!(wal.read()?, Some(op));
chusql-core/storage/tests/wal_test.rs
+28
View File
@@ -4,6 +4,33 @@ use serde_json::json;
// WAL 测试：追加、读取、清空、截断尾巴、LSN 与提交标记。
/// 结构版本往返且旧隐藏列载荷可读
#[test]
fn schema_versions_round_trip_and_legacy_hide_decodes() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("wal.log");
let wal = Wal::new(&path);
let ops = [
WalOp::HideColumn { table: "t".into(), column: "name".into(), schema_version: Some(u64::MAX) },
WalOp::DropColumn { table: "t".into(), column: "name".into(), rows: vec![], schema_version: Some(3) },
WalOp::ReplaceSchema { table: "t".into(), columns: vec![], rows: vec![], schema_version: Some(4) },
];
wal.append_group(&ops).unwrap();
assert_eq!(committed_ops(&wal.read_all().unwrap().records).into_iter().map(|(_, op)| op).collect::<Vec<_>>(), ops);
drop(wal);
let mut body = 1_u64.to_le_bytes().to_vec();
body.push(8);
body.extend_from_slice(&1_u16.to_le_bytes());
body.extend_from_slice(b"t");
body.extend_from_slice(br#""name""#);
let mut frame = (body.len() as u32).to_le_bytes().to_vec();
frame.extend_from_slice(&body);
std::fs::write(&path, frame).unwrap();
assert_eq!(Wal::new(&path).read().unwrap(), Some(WalOp::HideColumn {
table: "t".into(), column: "name".into(), schema_version: None,
}));
}
/// 建表编号完整往返且兼容旧载荷
#[test]
fn create_table_ids_round_trip_and_legacy_payloads_decode() {
@@ -58,6 +85,7 @@ fn append_and_read_round_trips() {
WalOp::DropColumn {
table: "users".into(),
column: "age".into(),
schema_version: None,
rows: vec![
row(&[("id", json!(1)), ("name", json!("Alice"))]),
row(&[("id", json!(2)), ("name", json!("Bob"))]),