use async_trait::async_trait;
use sbol_db_core::DomainError;
use sbol_db_storage::{
Activity, BlockingLock, DatabaseSize, DbStats, IncomingForeignKey, IndexStats,
OutgoingForeignKey, RelationalColumn, RelationalSchema, RelationalTable, SlowQuery,
TableColumn, TableSchema, TableStats,
};
use sqlx::Row;
use crate::pool::db_err;
use crate::SqlitePool;
#[derive(Clone)]
pub struct SqliteStats {
pool: SqlitePool,
}
fn quote_ident(name: &str) -> String {
format!("\"{}\"", name.replace('"', "\"\""))
}
impl SqliteStats {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
async fn user_tables(&self) -> Result<Vec<String>, DomainError> {
let rows = sqlx::query(
"SELECT name FROM sqlite_master \
WHERE type = 'table' \
AND name NOT LIKE 'sqlite_%' \
AND name <> '_sqlx_migrations' \
ORDER BY name",
)
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
rows.into_iter()
.map(|r| r.try_get::<String, _>("name").map_err(db_err))
.collect()
}
async fn object_bytes(&self, name: &str) -> i64 {
sqlx::query_scalar::<_, i64>("SELECT COALESCE(SUM(pgsize), 0) FROM dbstat WHERE name = ?")
.bind(name)
.fetch_one(&self.pool)
.await
.unwrap_or(0)
}
async fn table_exists(&self, name: &str) -> Result<bool, DomainError> {
let found: Option<String> =
sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?")
.bind(name)
.fetch_optional(&self.pool)
.await
.map_err(db_err)?;
Ok(found.is_some())
}
}
#[async_trait]
impl DbStats for SqliteStats {
async fn schema_overview(&self) -> Result<RelationalSchema, DomainError> {
let mut tables = Vec::new();
for name in self.user_tables().await? {
let rows = sqlx::query(&format!("PRAGMA table_info({})", quote_ident(&name)))
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
let columns = rows
.into_iter()
.map(|r| {
Ok::<_, DomainError>(RelationalColumn {
name: r.try_get("name").map_err(db_err)?,
column_type: r.try_get("type").map_err(db_err)?,
nullable: r.try_get::<i64, _>("notnull").map_err(db_err)? == 0,
})
})
.collect::<Result<_, _>>()?;
tables.push(RelationalTable { name, columns });
}
Ok(RelationalSchema { tables })
}
async fn database_size(&self) -> Result<DatabaseSize, DomainError> {
let page_count: i64 = sqlx::query_scalar("PRAGMA page_count")
.fetch_one(&self.pool)
.await
.map_err(db_err)?;
let page_size: i64 = sqlx::query_scalar("PRAGMA page_size")
.fetch_one(&self.pool)
.await
.map_err(db_err)?;
Ok(DatabaseSize {
database: "main".to_string(),
total_bytes: page_count * page_size,
})
}
async fn table_stats(&self, limit: i64, offset: i64) -> Result<Vec<TableStats>, DomainError> {
let names: Vec<String> = self
.user_tables()
.await?
.into_iter()
.skip(offset.max(0) as usize)
.take(limit.max(0) as usize)
.collect();
let mut stats = Vec::with_capacity(names.len());
for name in names {
let rows_estimate: i64 =
sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {}", quote_ident(&name)))
.fetch_one(&self.pool)
.await
.map_err(db_err)?;
let total_bytes = self.object_bytes(&name).await;
stats.push(TableStats {
name,
rows_estimate,
total_bytes,
index_bytes: 0,
n_live_tup: rows_estimate,
n_dead_tup: 0,
last_vacuum: None,
last_autovacuum: None,
last_analyze: None,
});
}
Ok(stats)
}
async fn index_stats(&self, limit: i64) -> Result<Vec<IndexStats>, DomainError> {
let rows = sqlx::query(
"SELECT name, tbl_name FROM sqlite_master \
WHERE type = 'index' AND name NOT LIKE 'sqlite_%' \
ORDER BY name LIMIT ?",
)
.bind(limit)
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
let mut stats = Vec::with_capacity(rows.len());
for r in rows {
let index: String = r.try_get("name").map_err(db_err)?;
let table: String = r.try_get("tbl_name").map_err(db_err)?;
let bytes = self.object_bytes(&index).await;
stats.push(IndexStats {
table,
index,
idx_scan: 0,
bytes,
});
}
Ok(stats)
}
async fn activity(&self, _limit: i64) -> Result<Vec<Activity>, DomainError> {
Ok(vec![])
}
async fn blocking_locks(&self) -> Result<Vec<BlockingLock>, DomainError> {
Ok(vec![])
}
async fn table_schema(&self, name: &str) -> Result<Option<TableSchema>, DomainError> {
if !self.table_exists(name).await? {
return Ok(None);
}
let column_rows = sqlx::query(&format!("PRAGMA table_info({})", quote_ident(name)))
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
let columns = column_rows
.into_iter()
.map(|r| {
Ok::<_, DomainError>(TableColumn {
name: r.try_get("name").map_err(db_err)?,
column_type: r.try_get("type").map_err(db_err)?,
nullable: r.try_get::<i64, _>("notnull").map_err(db_err)? == 0,
default_expr: r.try_get("dflt_value").map_err(db_err)?,
ordinal: r.try_get::<i64, _>("cid").map_err(db_err)? as i32 + 1,
comment: None,
is_primary_key: r.try_get::<i64, _>("pk").map_err(db_err)? != 0,
})
})
.collect::<Result<_, _>>()?;
let foreign_keys_out = self.foreign_keys_out(name).await?;
let foreign_keys_in = self.foreign_keys_in(name).await?;
Ok(Some(TableSchema {
name: name.to_string(),
comment: None,
columns,
foreign_keys_out,
foreign_keys_in,
}))
}
async fn has_slow_query_stats(&self) -> Result<bool, DomainError> {
Ok(false)
}
async fn slow_queries(&self, _limit: i64) -> Result<Vec<SlowQuery>, DomainError> {
Ok(vec![])
}
async fn optimize(&self) -> Result<(), DomainError> {
sqlx::query("VACUUM;")
.execute(&self.pool)
.await
.map_err(db_err)?;
sqlx::query("ANALYZE;")
.execute(&self.pool)
.await
.map_err(db_err)?;
Ok(())
}
}
impl SqliteStats {
async fn foreign_keys_out(&self, name: &str) -> Result<Vec<OutgoingForeignKey>, DomainError> {
let rows = sqlx::query(&format!("PRAGMA foreign_key_list({})", quote_ident(name)))
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
let mut order: Vec<i64> = Vec::new();
let mut by_id: std::collections::HashMap<i64, OutgoingForeignKey> =
std::collections::HashMap::new();
for r in rows {
let id: i64 = r.try_get("id").map_err(db_err)?;
let from: String = r.try_get("from").map_err(db_err)?;
let to: String = r.try_get("to").map_err(db_err)?;
let target_table: String = r.try_get("table").map_err(db_err)?;
let entry = by_id.entry(id).or_insert_with(|| {
order.push(id);
OutgoingForeignKey {
name: format!("fk_{id}"),
columns: Vec::new(),
target_table,
target_columns: Vec::new(),
}
});
entry.columns.push(from);
entry.target_columns.push(to);
}
Ok(order
.into_iter()
.filter_map(|id| by_id.remove(&id))
.collect())
}
async fn foreign_keys_in(&self, name: &str) -> Result<Vec<IncomingForeignKey>, DomainError> {
let mut incoming = Vec::new();
for source in self.user_tables().await? {
if source == name {
continue;
}
let rows = sqlx::query(&format!(
"PRAGMA foreign_key_list({})",
quote_ident(&source)
))
.fetch_all(&self.pool)
.await
.map_err(db_err)?;
let mut order: Vec<i64> = Vec::new();
let mut by_id: std::collections::HashMap<i64, IncomingForeignKey> =
std::collections::HashMap::new();
for r in rows {
let target_table: String = r.try_get("table").map_err(db_err)?;
if target_table != name {
continue;
}
let id: i64 = r.try_get("id").map_err(db_err)?;
let from: String = r.try_get("from").map_err(db_err)?;
let to: String = r.try_get("to").map_err(db_err)?;
let entry = by_id.entry(id).or_insert_with(|| {
order.push(id);
IncomingForeignKey {
name: format!("fk_{id}"),
source_table: source.clone(),
source_columns: Vec::new(),
target_columns: Vec::new(),
}
});
entry.source_columns.push(from);
entry.target_columns.push(to);
}
for id in order {
if let Some(fk) = by_id.remove(&id) {
incoming.push(fk);
}
}
}
Ok(incoming)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{connect, run_migrations};
async fn stats() -> SqliteStats {
let pool = connect("sqlite::memory:").await.unwrap();
run_migrations(&pool).await.unwrap();
SqliteStats::new(pool)
}
#[tokio::test]
async fn schema_overview_lists_known_tables() {
let stats = stats().await;
let schema = stats.schema_overview().await.unwrap();
let names: Vec<&str> = schema.tables.iter().map(|t| t.name.as_str()).collect();
assert!(names.contains(&"sbol_graphs"));
assert!(names.contains(&"sbol_triples"));
assert!(names.contains(&"sbol_objects"));
assert!(!names.contains(&"_sqlx_migrations"));
let graphs = schema
.tables
.iter()
.find(|t| t.name == "sbol_graphs")
.unwrap();
assert!(!graphs.columns.is_empty());
}
#[tokio::test]
async fn database_size_is_positive() {
let stats = stats().await;
let size = stats.database_size().await.unwrap();
assert_eq!(size.database, "main");
assert!(size.total_bytes > 0);
}
#[tokio::test]
async fn table_schema_returns_columns() {
let stats = stats().await;
let schema = stats.table_schema("sbol_triples").await.unwrap().unwrap();
assert_eq!(schema.name, "sbol_triples");
assert!(!schema.columns.is_empty());
let unknown = stats.table_schema("does_not_exist").await.unwrap();
assert!(unknown.is_none());
}
#[tokio::test]
async fn table_stats_and_indexes() {
let stats = stats().await;
let tables = stats.table_stats(100, 0).await.unwrap();
assert!(tables.iter().any(|t| t.name == "sbol_triples"));
let schema = stats.table_schema("sbol_triples").await.unwrap().unwrap();
assert!(!schema.foreign_keys_out.is_empty());
let _ = stats.index_stats(50).await.unwrap();
}
#[tokio::test]
async fn empty_observability_surfaces() {
let stats = stats().await;
assert!(stats.activity(10).await.unwrap().is_empty());
assert!(stats.blocking_locks().await.unwrap().is_empty());
assert!(!stats.has_slow_query_stats().await.unwrap());
assert!(stats.slow_queries(10).await.unwrap().is_empty());
}
}