use std::{
collections::HashSet,
hash::{Hash, Hasher},
path::{Path, PathBuf},
sync::{Mutex, RwLock},
};
use rustc_hash::FxHashMap;
pub(crate) struct CustomTableState {
pub columns: Vec<String>,
pub rows: Vec<Vec<String>>,
pub first_row_page: u32,
pub last_row_page: u32,
}
use mq_markdown::Markdown;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DatabaseAlias(String);
impl DatabaseAlias {
const RESERVED: [&'static str; 3] = ["main", "blocks", "documents"];
pub fn parse(raw: &str) -> Result<Self, MqdbError> {
if raw.is_empty() {
return Err(MqdbError::SqlExec("database alias cannot be empty".into()));
}
let lower = raw.to_ascii_lowercase();
if Self::RESERVED.contains(&lower.as_str()) {
return Err(MqdbError::SqlExec(format!(
"'{raw}' is a reserved name and cannot be used as a database alias"
)));
}
Ok(Self(lower))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for DatabaseAlias {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::borrow::Borrow<str> for DatabaseAlias {
fn borrow(&self) -> &str {
&self.0
}
}
use crate::{
block::{BlockType, DocumentId},
document::Document,
error::MqdbError,
index,
indexes::DocumentIndex,
query::Query,
storage::{
Storage,
catalog::{CatalogEntry, CustomTableEntry, ViewEntry},
codec::{decode_zone_map, encode_zone_map},
page::{FILE_VERSION, PAGE_SIZE},
},
};
fn persist_unsaved_table_rows(
storage: &mut Storage,
custom_tables: &RwLock<FxHashMap<String, CustomTableState>>,
) -> Result<Vec<CustomTableEntry>, MqdbError> {
let mut guard = custom_tables.write().unwrap();
for state in guard.values_mut() {
if state.first_row_page == 0 && !state.rows.is_empty() {
let (first, last) = storage.write_table_rows(&state.rows)?;
state.first_row_page = first;
state.last_row_page = last;
}
}
Ok(guard
.iter()
.map(|(name, state)| CustomTableEntry {
name: name.clone(),
columns: state.columns.clone(),
first_row_page: state.first_row_page,
last_row_page: state.last_row_page,
num_rows: state.rows.len() as u32,
})
.collect())
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ReindexReport {
pub added: Vec<PathBuf>,
pub updated: Vec<PathBuf>,
pub unchanged: usize,
pub removed: Vec<PathBuf>,
pub failed: Vec<(PathBuf, String)>,
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct StoreStats {
pub documents: usize,
pub blocks: usize,
pub block_type_counts: Vec<(BlockType, usize)>,
pub code_lang_counts: Vec<(String, usize)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VacuumReport {
pub pages_before: u32,
pub pages_after: u32,
}
impl VacuumReport {
pub fn bytes_reclaimed(&self) -> u64 {
u64::from(self.pages_before.saturating_sub(self.pages_after)) * PAGE_SIZE as u64
}
}
pub struct DocumentStore {
documents: Vec<Document>,
next_doc_id: DocumentId,
store_spans: bool,
pub(crate) storage: Mutex<Option<Storage>>,
pub(crate) doc_indexes: Vec<Option<DocumentIndex>>,
pub(crate) custom_tables: RwLock<FxHashMap<String, CustomTableState>>,
pub(crate) views: RwLock<FxHashMap<String, String>>,
pub(crate) attached: RwLock<FxHashMap<DatabaseAlias, DocumentStore>>,
content_hashes: FxHashMap<DocumentId, u64>,
}
impl Default for DocumentStore {
fn default() -> Self {
Self {
documents: Vec::new(),
next_doc_id: 0,
store_spans: true,
storage: Mutex::new(None),
doc_indexes: Vec::new(),
custom_tables: RwLock::new(FxHashMap::default()),
views: RwLock::new(FxHashMap::default()),
attached: RwLock::new(FxHashMap::default()),
content_hashes: FxHashMap::default(),
}
}
}
fn hash_bytes(bytes: &[u8]) -> u64 {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
bytes.hash(&mut hasher);
hasher.finish()
}
fn read_files_parallel(files: &[PathBuf]) -> Vec<Result<String, MqdbError>> {
let worker_count = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
.min(files.len().max(1));
if worker_count <= 1 {
return files
.iter()
.map(|p| std::fs::read_to_string(p).map_err(MqdbError::from))
.collect();
}
let chunk_size = files.len().div_ceil(worker_count);
std::thread::scope(|scope| {
files
.chunks(chunk_size)
.map(|chunk| {
scope.spawn(move || {
chunk
.iter()
.map(|p| std::fs::read_to_string(p).map_err(MqdbError::from))
.collect::<Vec<_>>()
})
})
.collect::<Vec<_>>()
.into_iter()
.flat_map(|handle| handle.join().expect("file-read worker thread panicked"))
.collect()
})
}
impl DocumentStore {
pub fn new() -> Self {
Self::default()
}
pub fn set_store_spans(&mut self, val: bool) {
self.store_spans = val;
}
pub fn register_table(
&mut self,
name: impl Into<String>,
columns: Vec<String>,
rows: Vec<Vec<String>>,
) {
self.custom_tables.write().unwrap().insert(
name.into(),
CustomTableState {
columns,
rows,
first_row_page: 0,
last_row_page: 0,
},
);
}
pub fn unregister_table(&mut self, name: &str) -> bool {
self.custom_tables.write().unwrap().remove(name).is_some()
}
pub fn attach(&self, alias: DatabaseAlias, path: &Path) -> Result<(), MqdbError> {
if self.attached.read().unwrap().contains_key(&alias) {
return Err(MqdbError::SqlExec(format!(
"database alias '{alias}' is already attached — DETACH it first"
)));
}
let mut other = DocumentStore::open(path)?;
other.load_all_blocks()?;
other.load_all_indexes()?;
self.attached.write().unwrap().insert(alias, other);
Ok(())
}
pub fn detach(&self, alias: &str) -> bool {
self.attached
.write()
.unwrap()
.remove(alias.to_ascii_lowercase().as_str())
.is_some()
}
pub fn add_file(&mut self, path: impl AsRef<Path>) -> Result<DocumentId, MqdbError> {
let path = path.as_ref();
let content = std::fs::read_to_string(path)?;
self.add_str_with_path(&content, Some(path.to_path_buf()))
}
pub fn add_str(&mut self, content: &str) -> Result<DocumentId, MqdbError> {
self.add_str_with_path(content, None)
}
pub fn add_str_with_path(
&mut self,
content: &str,
path: Option<std::path::PathBuf>,
) -> Result<DocumentId, MqdbError> {
let md =
Markdown::from_markdown_str(content).map_err(|e| MqdbError::Parse(e.to_string()))?;
let doc_id = self.next_doc_id;
self.next_doc_id += 1;
let mut blocks = index::build_blocks(doc_id, &md.nodes);
if !self.store_spans {
for block in &mut blocks {
block.span = None;
}
}
let doc = Document::new(doc_id, path, blocks);
self.documents.push(doc);
self.doc_indexes.push(None);
Ok(doc_id)
}
pub fn append_str(&mut self, content: &str) -> Result<DocumentId, MqdbError> {
self.do_append(content, None, true)
}
pub fn append_file(&mut self, path: impl AsRef<Path>) -> Result<DocumentId, MqdbError> {
let path = path.as_ref();
let content = std::fs::read_to_string(path)?;
self.do_append(&content, Some(path.to_path_buf()), true)
}
fn do_append(
&mut self,
content: &str,
md_path: Option<PathBuf>,
flush: bool,
) -> Result<DocumentId, MqdbError> {
let md =
Markdown::from_markdown_str(content).map_err(|e| MqdbError::Parse(e.to_string()))?;
let doc_id = self.next_doc_id;
self.next_doc_id += 1;
let mut blocks = index::build_blocks(doc_id, &md.nodes);
if !self.store_spans {
for block in &mut blocks {
block.span = None;
}
}
let mut doc = Document::new(doc_id, md_path, blocks);
let idx_opt = {
let mut storage_guard = self.storage.lock().unwrap();
if let Some(storage) = storage_guard.as_mut() {
let first_block_page = storage.write_document(&doc)?;
doc.first_block_page = first_block_page;
let idx = DocumentIndex::build(&doc.blocks);
let index_start_page = storage.write_index(&idx.to_bytes())?;
doc.index_start_page = index_start_page;
Some(idx)
} else {
None
}
};
self.doc_indexes.push(idx_opt);
self.documents.push(doc);
if flush {
self.try_flush_catalog_to_storage();
}
Ok(doc_id)
}
pub fn replace_document(
&mut self,
doc_id: DocumentId,
content: &str,
path: Option<PathBuf>,
) -> Result<(), MqdbError> {
self.do_replace(doc_id, content, path, true)
}
fn do_replace(
&mut self,
doc_id: DocumentId,
content: &str,
path: Option<PathBuf>,
flush: bool,
) -> Result<(), MqdbError> {
let pos = self
.documents
.iter()
.position(|d| d.id == doc_id)
.ok_or_else(|| MqdbError::Storage(format!("no such document: {doc_id}")))?;
self.do_replace_at(pos, doc_id, content, path, flush)
}
fn do_replace_at(
&mut self,
pos: usize,
doc_id: DocumentId,
content: &str,
path: Option<PathBuf>,
flush: bool,
) -> Result<(), MqdbError> {
let md =
Markdown::from_markdown_str(content).map_err(|e| MqdbError::Parse(e.to_string()))?;
let mut blocks = index::build_blocks(doc_id, &md.nodes);
if !self.store_spans {
for block in &mut blocks {
block.span = None;
}
}
let mut doc = Document::new(doc_id, path, blocks);
let idx_opt = {
let mut storage_guard = self.storage.lock().unwrap();
if let Some(storage) = storage_guard.as_mut() {
let first_block_page = storage.write_document(&doc)?;
doc.first_block_page = first_block_page;
let idx = DocumentIndex::build(&doc.blocks);
let index_start_page = storage.write_index(&idx.to_bytes())?;
doc.index_start_page = index_start_page;
Some(idx)
} else {
None
}
};
self.documents[pos] = doc;
self.doc_indexes[pos] = idx_opt;
if flush {
self.try_flush_catalog_to_storage();
}
Ok(())
}
pub fn reindex_paths(
&mut self,
files: &[PathBuf],
prune: bool,
) -> Result<ReindexReport, MqdbError> {
let mut report = ReindexReport::default();
let mut seen: HashSet<PathBuf> = HashSet::with_capacity(files.len());
let contents = read_files_parallel(files);
let mut by_path: FxHashMap<PathBuf, (DocumentId, usize)> = self
.documents
.iter()
.enumerate()
.filter_map(|(i, d)| d.path.clone().map(|p| (p, (d.id, i))))
.collect();
for (path, content) in files.iter().zip(contents) {
seen.insert(path.clone());
let result = (|| -> Result<(), MqdbError> {
let content = content?;
let hash = hash_bytes(content.as_bytes());
let existing = by_path.get(path).copied();
match existing {
Some((doc_id, _)) if self.content_hashes.get(&doc_id) == Some(&hash) => {
report.unchanged += 1;
}
Some((doc_id, pos)) => {
self.do_replace_at(pos, doc_id, &content, Some(path.clone()), false)?;
self.content_hashes.insert(doc_id, hash);
report.updated.push(path.clone());
}
None => {
let doc_id = if self.storage.lock().unwrap().is_some() {
self.do_append(&content, Some(path.clone()), false)?
} else {
self.add_str_with_path(&content, Some(path.clone()))?
};
self.content_hashes.insert(doc_id, hash);
by_path.insert(path.clone(), (doc_id, self.documents.len() - 1));
report.added.push(path.clone());
}
}
Ok(())
})();
if let Err(e) = result {
report.failed.push((path.clone(), e.to_string()));
}
}
if prune {
let to_remove: Vec<(usize, DocumentId)> = self
.documents
.iter()
.enumerate()
.filter(|(_, d)| d.path.as_ref().is_some_and(|p| !seen.contains(p)))
.map(|(i, d)| (i, d.id))
.collect();
for (i, doc_id) in to_remove.into_iter().rev() {
let removed_doc = self.documents.remove(i);
self.doc_indexes.remove(i);
self.content_hashes.remove(&doc_id);
if let Some(p) = removed_doc.path {
report.removed.push(p);
}
}
}
self.try_flush_catalog_to_storage();
Ok(report)
}
pub fn documents(&self) -> &[Document] {
&self.documents
}
pub fn get_document(&self, id: DocumentId) -> Option<&Document> {
self.documents.iter().find(|d| d.id == id)
}
pub fn len(&self) -> usize {
self.documents.len()
}
pub fn is_empty(&self) -> bool {
self.documents.is_empty()
}
pub fn query(&self) -> Query<'_> {
Query::new(self)
}
pub fn stats(&self) -> StoreStats {
let mut type_counts: FxHashMap<BlockType, usize> = FxHashMap::default();
let mut lang_counts: FxHashMap<String, usize> = FxHashMap::default();
let mut total_blocks = 0usize;
for doc in &self.documents {
total_blocks += doc.blocks.len();
for block in &doc.blocks {
*type_counts.entry(block.block_type.clone()).or_insert(0) += 1;
if block.block_type == BlockType::Code
&& let Some(lang) = block.code_lang()
{
*lang_counts.entry(lang.to_string()).or_insert(0) += 1;
}
}
}
let mut block_type_counts: Vec<(BlockType, usize)> = type_counts.into_iter().collect();
block_type_counts.sort_by_key(|(_, v)| std::cmp::Reverse(*v));
let mut code_lang_counts: Vec<(String, usize)> = lang_counts.into_iter().collect();
code_lang_counts.sort_by_key(|(_, v)| std::cmp::Reverse(*v));
StoreStats {
documents: self.documents.len(),
blocks: total_blocks,
block_type_counts,
code_lang_counts,
}
}
pub fn load_all_blocks(&mut self) -> Result<(), MqdbError> {
let mut guard = self.storage.lock().unwrap();
let storage = match guard.as_mut() {
Some(s) => s,
None => return Ok(()),
};
for doc in &mut self.documents {
if doc.blocks.is_empty() && doc.block_count > 0 {
doc.blocks = storage.read_blocks(doc.first_block_page, doc.block_count)?;
}
}
Ok(())
}
pub fn load_all_indexes(&mut self) -> Result<(), MqdbError> {
for i in 0..self.documents.len() {
if self.doc_indexes[i].is_some() {
continue;
}
let idx = self.build_or_load_index_at(i)?;
self.doc_indexes[i] = Some(idx);
}
Ok(())
}
fn build_or_load_index_at(&mut self, i: usize) -> Result<DocumentIndex, MqdbError> {
let index_start_page = self.documents[i].index_start_page;
if index_start_page > 0 {
let mut guard = self.storage.lock().unwrap();
if let Some(storage) = guard.as_mut() {
let bytes = storage.read_index_bytes(index_start_page)?;
return DocumentIndex::from_bytes(&bytes);
}
}
Ok(DocumentIndex::build(&self.documents[i].blocks))
}
pub(crate) fn get_doc_index(&self, i: usize) -> Option<&DocumentIndex> {
self.doc_indexes.get(i).and_then(|o| o.as_ref())
}
fn catalog_entries(&self) -> Vec<CatalogEntry> {
self.documents
.iter()
.map(|d| CatalogEntry {
document_id: d.id,
path: d.path.as_ref().map(|p| p.to_string_lossy().into_owned()),
first_block_page: d.first_block_page,
num_blocks: d.block_count,
zone_map_bytes: encode_zone_map(&d.zone_maps),
index_start_page: d.index_start_page,
})
.collect()
}
fn content_hash_pairs(&self) -> Vec<(u32, u64)> {
self.content_hashes.iter().map(|(k, v)| (*k, *v)).collect()
}
fn views_entries(&self) -> Vec<ViewEntry> {
self.views
.read()
.unwrap()
.iter()
.map(|(name, sql)| ViewEntry {
name: name.clone(),
sql: sql.clone(),
})
.collect()
}
pub(crate) fn try_flush_catalog_to_storage(&self) {
let mut guard = self.storage.lock().unwrap();
if let Some(storage) = guard.as_mut() {
let entries = self.catalog_entries();
if let Ok(custom) = persist_unsaved_table_rows(storage, &self.custom_tables) {
let _ = storage.flush_catalog(
&entries,
&custom,
&self.content_hash_pairs(),
&self.views_entries(),
);
}
}
}
pub(crate) fn try_append_table_rows_to_storage(
&self,
table_name: &str,
new_rows: &[Vec<String>],
) {
let mut guard = self.storage.lock().unwrap();
let storage = match guard.as_mut() {
Some(s) => s,
None => return,
};
{
let mut ct_guard = self.custom_tables.write().unwrap();
if let Some(state) = ct_guard.get_mut(table_name) {
let persisted = if state.first_row_page == 0 {
storage.write_table_rows(&state.rows)
} else {
storage
.append_table_rows(state.last_row_page, new_rows)
.map(|last| (state.first_row_page, last))
};
if let Ok((first, last)) = persisted {
state.first_row_page = first;
state.last_row_page = last;
}
}
}
let entries = self.catalog_entries();
let ct_guard = self.custom_tables.read().unwrap();
let custom: Vec<CustomTableEntry> = ct_guard
.iter()
.map(|(name, state)| CustomTableEntry {
name: name.clone(),
columns: state.columns.clone(),
first_row_page: state.first_row_page,
last_row_page: state.last_row_page,
num_rows: state.rows.len() as u32,
})
.collect();
drop(ct_guard);
let _ = storage.flush_catalog(
&entries,
&custom,
&self.content_hash_pairs(),
&self.views_entries(),
);
}
pub fn save(&self, path: impl AsRef<Path>) -> Result<(), MqdbError> {
let path = path.as_ref();
let tmp_path = PathBuf::from(format!("{}.tmp", path.to_string_lossy()));
if tmp_path.exists() {
std::fs::remove_file(&tmp_path)?;
}
let write_result = (|| -> Result<(), MqdbError> {
let mut storage = Storage::create(&tmp_path)?;
let mut entries = Vec::with_capacity(self.documents.len());
for doc in &self.documents {
let first_block_page = storage.write_document(doc)?;
entries.push(CatalogEntry {
document_id: doc.id,
path: doc.path.as_ref().map(|p| p.to_string_lossy().into_owned()),
first_block_page,
num_blocks: doc.block_count,
zone_map_bytes: encode_zone_map(&doc.zone_maps),
index_start_page: 0,
});
}
for (i, doc) in self.documents.iter().enumerate() {
let idx = if let Some(cached) = self.doc_indexes.get(i).and_then(|o| o.as_ref()) {
std::borrow::Cow::Borrowed(cached)
} else {
std::borrow::Cow::Owned(DocumentIndex::build(&doc.blocks))
};
let bytes = idx.to_bytes();
entries[i].index_start_page = storage.write_index(&bytes)?;
}
let ct_guard = self.custom_tables.read().unwrap();
let mut custom = Vec::with_capacity(ct_guard.len());
for (name, state) in ct_guard.iter() {
let (first_row_page, last_row_page) = storage.write_table_rows(&state.rows)?;
custom.push(CustomTableEntry {
name: name.clone(),
columns: state.columns.clone(),
first_row_page,
last_row_page,
num_rows: state.rows.len() as u32,
});
}
drop(ct_guard);
storage.flush_catalog(
&entries,
&custom,
&self.content_hash_pairs(),
&self.views_entries(),
)?;
Ok(())
})();
if let Err(err) = write_result {
let _ = std::fs::remove_file(&tmp_path);
return Err(err);
}
std::fs::rename(&tmp_path, path)?;
Ok(())
}
pub fn vacuum(&mut self, path: impl AsRef<Path>) -> Result<VacuumReport, MqdbError> {
let path = path.as_ref();
let pages_before = {
let guard = self.storage.lock().unwrap();
let Some(storage) = guard.as_ref() else {
return Err(MqdbError::Storage(
"vacuum requires a store opened from a file (DocumentStore::open) — \
this store has no backing file"
.into(),
));
};
storage.num_pages()
};
self.save(path)?;
let reopened = Storage::open(path)?;
let pages_after = reopened.num_pages();
*self.storage.lock().unwrap() = Some(reopened);
Ok(VacuumReport {
pages_before,
pages_after,
})
}
pub fn open(path: impl AsRef<Path>) -> Result<Self, MqdbError> {
let mut storage = Storage::open(path.as_ref())?;
if storage.file_version() != FILE_VERSION {
return Err(MqdbError::Storage(format!(
"store file is version {} (expected {FILE_VERSION}); run `DocumentStore::migrate` \
(or open it once via an interactive `mq-db` command, which offers to migrate) before opening it for writes",
storage.file_version()
)));
}
let (entries, custom_table_entries, content_hashes, view_entries) =
storage.load_catalog()?;
let cap = entries.len();
let mut documents = Vec::with_capacity(cap);
let mut max_doc_id = None;
for entry in entries {
let zone_maps = decode_zone_map(&entry.zone_map_bytes)?;
let document_id = entry.document_id;
let path = entry.path.map(PathBuf::from);
documents.push(Document::from_catalog_lazy(
document_id,
path,
entry.num_blocks,
zone_maps,
entry.first_block_page,
entry.index_start_page,
));
max_doc_id =
Some(max_doc_id.map_or(document_id, |cur: DocumentId| cur.max(document_id)));
}
let mut custom_tables = FxHashMap::default();
for ct in custom_table_entries {
let rows = storage.read_table_rows(ct.first_row_page, ct.num_rows, ct.columns.len())?;
custom_tables.insert(
ct.name,
CustomTableState {
columns: ct.columns,
rows,
first_row_page: ct.first_row_page,
last_row_page: ct.last_row_page,
},
);
}
let views: FxHashMap<String, String> =
view_entries.into_iter().map(|v| (v.name, v.sql)).collect();
Ok(Self {
documents,
next_doc_id: max_doc_id.map_or(0, |id| id.saturating_add(1)),
store_spans: true,
storage: Mutex::new(Some(storage)),
doc_indexes: vec![None; cap],
custom_tables: RwLock::new(custom_tables),
views: RwLock::new(views),
attached: RwLock::new(FxHashMap::default()),
content_hashes: content_hashes.into_iter().collect(),
})
}
pub fn load(path: impl AsRef<Path>) -> Result<Self, MqdbError> {
let mut storage = Storage::open(path.as_ref())?;
let (entries, custom_table_entries, content_hashes, view_entries) =
storage.load_catalog()?;
let cap = entries.len();
let mut documents = Vec::with_capacity(cap);
let mut max_doc_id = None;
for entry in entries {
let blocks = storage.read_blocks(entry.first_block_page, entry.num_blocks)?;
let zone_maps = decode_zone_map(&entry.zone_map_bytes)?;
let document_id = entry.document_id;
let path = entry.path.map(PathBuf::from);
let mut doc = Document::from_parts(document_id, path, blocks, zone_maps);
doc.index_start_page = entry.index_start_page;
documents.push(doc);
max_doc_id =
Some(max_doc_id.map_or(document_id, |cur: DocumentId| cur.max(document_id)));
}
let mut custom_tables = FxHashMap::default();
for ct in custom_table_entries {
let rows = storage.read_table_rows(ct.first_row_page, ct.num_rows, ct.columns.len())?;
custom_tables.insert(
ct.name,
CustomTableState {
columns: ct.columns,
rows,
first_row_page: ct.first_row_page,
last_row_page: ct.last_row_page,
},
);
}
let views: FxHashMap<String, String> =
view_entries.into_iter().map(|v| (v.name, v.sql)).collect();
Ok(Self {
documents,
next_doc_id: max_doc_id.map_or(0, |id| id.saturating_add(1)),
store_spans: true,
storage: Mutex::new(None),
doc_indexes: vec![None; cap],
custom_tables: RwLock::new(custom_tables),
views: RwLock::new(views),
attached: RwLock::new(FxHashMap::default()),
content_hashes: content_hashes.into_iter().collect(),
})
}
pub fn load_catalog_only(path: impl AsRef<Path>) -> Result<Self, MqdbError> {
let mut storage = Storage::open(path.as_ref())?;
let (entries, _custom_table_entries, content_hashes, _view_entries) =
storage.load_catalog()?;
let cap = entries.len();
let mut documents = Vec::with_capacity(cap);
let mut max_doc_id = None;
for entry in entries {
let zone_maps = decode_zone_map(&entry.zone_map_bytes)?;
let document_id = entry.document_id;
let path = entry.path.map(PathBuf::from);
documents.push(Document::from_catalog(
document_id,
path,
entry.num_blocks,
zone_maps,
));
max_doc_id =
Some(max_doc_id.map_or(document_id, |cur: DocumentId| cur.max(document_id)));
}
Ok(Self {
documents,
next_doc_id: max_doc_id.map_or(0, |id| id.saturating_add(1)),
store_spans: true,
storage: Mutex::new(None),
doc_indexes: vec![None; cap],
custom_tables: RwLock::new(FxHashMap::default()),
views: RwLock::new(FxHashMap::default()),
attached: RwLock::new(FxHashMap::default()),
content_hashes: content_hashes.into_iter().collect(),
})
}
pub fn file_version(path: impl AsRef<Path>) -> Result<u32, MqdbError> {
Ok(Storage::open(path.as_ref())?.file_version())
}
pub fn migrate(path: impl AsRef<Path>) -> Result<u32, MqdbError> {
let path = path.as_ref();
let old_version = Self::file_version(path)?;
if old_version == FILE_VERSION {
return Ok(old_version);
}
let store = Self::load(path)?;
store.save(path)?;
Ok(old_version)
}
}
#[cfg(test)]
mod alias_tests {
use super::*;
#[test]
fn parse_lowercases() {
assert_eq!(DatabaseAlias::parse("Other").unwrap().as_str(), "other");
}
#[test]
fn parse_rejects_empty() {
assert!(DatabaseAlias::parse("").is_err());
}
#[test]
fn parse_rejects_reserved_names_case_insensitively() {
for reserved in ["main", "BLOCKS", "Documents"] {
let err = DatabaseAlias::parse(reserved).unwrap_err();
assert!(err.to_string().contains("reserved"));
}
}
}
#[cfg(test)]
mod reindex_tests {
use super::*;
fn write_md(dir: &tempfile::TempDir, name: &str, content: &str) -> PathBuf {
let path = dir.path().join(name);
std::fs::write(&path, content).unwrap();
path
}
#[test]
fn reindex_in_memory_store_adds_new_files() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n\nHello\n");
let b = write_md(&dir, "b.md", "# B\n\nWorld\n");
let mut store = DocumentStore::new();
let report = store.reindex_paths(&[a.clone(), b.clone()], false).unwrap();
assert_eq!(report.added, vec![a, b]);
assert!(report.updated.is_empty());
assert_eq!(report.unchanged, 0);
assert!(report.removed.is_empty());
assert!(report.failed.is_empty());
assert_eq!(store.documents().len(), 2);
}
#[test]
fn reindex_skips_unchanged_file_on_second_run() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n\nHello\n");
let mut store = DocumentStore::new();
store
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
let doc_id_before = store.documents()[0].id;
let report = store
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
assert!(report.added.is_empty());
assert!(report.updated.is_empty());
assert_eq!(report.unchanged, 1);
assert_eq!(store.documents()[0].id, doc_id_before);
}
#[test]
fn reindex_replaces_changed_file_keeping_document_id() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n\nHello\n");
let mut store = DocumentStore::new();
store
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
let doc_id_before = store.documents()[0].id;
std::fs::write(&a, "# A Changed\n\nNew body\n").unwrap();
let report = store
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
assert!(report.added.is_empty());
assert_eq!(report.updated, vec![a]);
assert_eq!(report.unchanged, 0);
assert_eq!(store.documents()[0].id, doc_id_before);
assert!(
store.documents()[0]
.blocks
.iter()
.any(|b| b.content == "A Changed")
);
}
#[test]
fn reindex_prune_removes_missing_paths() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n");
let b = write_md(&dir, "b.md", "# B\n");
let mut store = DocumentStore::new();
store.reindex_paths(&[a.clone(), b.clone()], false).unwrap();
assert_eq!(store.documents().len(), 2);
let report = store.reindex_paths(std::slice::from_ref(&a), true).unwrap();
assert_eq!(report.removed, vec![b]);
assert_eq!(report.unchanged, 1);
assert_eq!(store.documents().len(), 1);
assert_eq!(store.documents()[0].path.as_deref(), Some(a.as_path()));
}
#[test]
fn reindex_on_backing_store_persists_hash_across_reload() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n\nHello\n");
let db_path = dir.path().join("store.mq-db");
let mut store = DocumentStore::new();
store
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
store.save(&db_path).unwrap();
let mut reopened = DocumentStore::open(&db_path).unwrap();
let report = reopened
.reindex_paths(std::slice::from_ref(&a), false)
.unwrap();
assert_eq!(report.unchanged, 1);
assert!(report.added.is_empty());
assert!(report.updated.is_empty());
}
#[test]
fn reindex_reports_failure_for_unreadable_path_without_aborting_others() {
let dir = tempfile::tempdir().unwrap();
let a = write_md(&dir, "a.md", "# A\n");
let missing = dir.path().join("does-not-exist.md");
let mut store = DocumentStore::new();
let report = store
.reindex_paths(&[a.clone(), missing.clone()], false)
.unwrap();
assert_eq!(report.added, vec![a]);
assert_eq!(report.failed.len(), 1);
assert_eq!(report.failed[0].0, missing);
}
}
#[cfg(test)]
mod vacuum_tests {
use super::*;
fn write_md(dir: &tempfile::TempDir, name: &str, content: &str) -> PathBuf {
let path = dir.path().join(name);
std::fs::write(&path, content).unwrap();
path
}
fn open_for_writes(path: &Path) -> DocumentStore {
let mut store = DocumentStore::open(path).unwrap();
store.load_all_blocks().unwrap();
store.load_all_indexes().unwrap();
store
}
#[test]
fn vacuum_reclaims_space_after_document_replace() {
let dir = tempfile::tempdir().unwrap();
let md_path = write_md(&dir, "a.md", "# A\n\nHello\n");
let db_path = dir.path().join("store.mq-db");
let mut store = DocumentStore::new();
let doc_id = store.add_file(&md_path).unwrap();
store.save(&db_path).unwrap();
let mut opened = open_for_writes(&db_path);
for i in 0..5 {
opened
.replace_document(
doc_id,
&format!("# A\n\nHello {i}\n"),
Some(md_path.clone()),
)
.unwrap();
}
let report = opened.vacuum(&db_path).unwrap();
assert!(
report.pages_before > report.pages_after,
"expected reclaim, got before={} after={}",
report.pages_before,
report.pages_after
);
let reloaded = DocumentStore::load(&db_path).unwrap();
assert_eq!(reloaded.documents().len(), 1);
assert!(
reloaded.documents()[0]
.blocks
.iter()
.any(|b| b.content == "Hello 4")
);
}
#[test]
fn vacuum_reclaims_space_after_drop_table() {
let dir = tempfile::tempdir().unwrap();
let md_path = write_md(&dir, "a.md", "# A\n\nHello\n");
let db_path = dir.path().join("store.mq-db");
let mut store = DocumentStore::new();
store.add_file(&md_path).unwrap();
store.save(&db_path).unwrap();
let mut opened = open_for_writes(&db_path);
opened.execute_sql_mut("CREATE TABLE t (x TEXT)").unwrap();
for i in 0..200 {
opened
.execute_sql_mut(&format!("INSERT INTO t VALUES ('row {i}')"))
.unwrap();
}
opened.execute_sql_mut("DROP TABLE t").unwrap();
let report = opened.vacuum(&db_path).unwrap();
assert!(
report.pages_before > report.pages_after,
"expected reclaim, got before={} after={}",
report.pages_before,
report.pages_after
);
}
#[test]
fn vacuum_is_a_noop_when_nothing_to_reclaim() {
let dir = tempfile::tempdir().unwrap();
let md_path = write_md(&dir, "a.md", "# A\n\nHello\n");
let db_path = dir.path().join("store.mq-db");
let mut store = DocumentStore::new();
store.add_file(&md_path).unwrap();
store.save(&db_path).unwrap();
let mut opened = open_for_writes(&db_path);
let report = opened.vacuum(&db_path).unwrap();
assert_eq!(report.pages_before, report.pages_after);
assert_eq!(report.bytes_reclaimed(), 0);
}
#[test]
fn vacuum_rejects_in_memory_only_store() {
let mut store = DocumentStore::new();
store.add_str("# A\n\nHello\n").unwrap();
let err = store.vacuum("/tmp/does-not-matter.mq-db").unwrap_err();
assert!(err.to_string().contains("backing file"));
}
#[test]
fn vacuum_preserves_views_and_custom_tables() {
let dir = tempfile::tempdir().unwrap();
let md_path = write_md(&dir, "a.md", "# A\n\nHello\n");
let db_path = dir.path().join("store.mq-db");
let mut store = DocumentStore::new();
store.add_file(&md_path).unwrap();
store.save(&db_path).unwrap();
let mut opened = open_for_writes(&db_path);
opened
.execute_sql_mut(
"CREATE VIEW v AS SELECT content FROM blocks WHERE block_type = 'heading'",
)
.unwrap();
opened.execute_sql_mut("CREATE TABLE t (x TEXT)").unwrap();
opened
.execute_sql_mut("INSERT INTO t VALUES ('hello')")
.unwrap();
opened.vacuum(&db_path).unwrap();
let out = opened.execute_sql_mut("SELECT content FROM v").unwrap();
assert_eq!(out.rows, vec![vec!["A".to_string()]]);
let out = opened.execute_sql_mut("SELECT x FROM t").unwrap();
assert_eq!(out.rows, vec![vec!["hello".to_string()]]);
let reloaded = DocumentStore::load(&db_path).unwrap();
assert_eq!(reloaded.documents().len(), 1);
}
}