8000 minigraf/src/db.rs at v1.2.2 · project-minigraf/minigraf · GitHub
[go: up one dir, main page]
Skip to content

db.rs

Latest commit

 

History

History
1931 lines (1692 loc) · 73.1 KB

db.rs

File metadata and controls

1931 lines (1692 loc) · 73.1 KB
Raw
//! Public `Minigraf` facade with `WriteTransaction` and `OpenOptions`.
//!
//! This module provides the primary user-facing API for Minigraf:
//! - `Minigraf::open()` / `Minigraf::in_memory()` for database creation
//! - `Minigraf::execute()` for implicit (self-contained) transactions
//! - `Minigraf::begin_write()` / `WriteTransaction` for explicit transactions
//! - `Minigraf::checkpoint()` for manual WAL compaction
use crate::graph::types::{Fact, TxId, VALID_TIME_FOREVER};
/// Sentinel value used in `materialize_transaction` to signal "no explicit `valid_from`
/// was provided; use the transaction timestamp at commit time."
///
/// `i64::MIN` is chosen because it is not a representable Unix millisecond timestamp
/// in any practical context, avoiding the collision that `0` would have with the Unix
/// epoch (1970-01-01T00:00:00Z), which is a legitimate `valid_from` value.
pub(crate) const VALID_FROM_USE_TX_TIME: i64 = i64::MIN;
use crate::graph::FactStorage;
use crate::graph::types::Value;
use crate::query::datalog::evaluator::DEFAULT_MAX_DERIVED_FACTS;
use crate::query::datalog::evaluator::DEFAULT_MAX_RESULTS;
use crate::query::datalog::executor::DatalogExecutor;
use crate::query::datalog::executor::QueryResult;
use crate::query::datalog::functions::{
AggImpl, AggregateDesc, FunctionRegistry, PredicateDesc, UdfFinaliseFn, UdfOps, UdfStepFn,
};
use crate::query::datalog::parser::parse_datalog_command;
use crate::query::datalog::rules::RuleRegistry;
use crate::query::datalog::types::{AttributeSpec, DatalogCommand, Transaction};
use crate::storage::backend::MemoryBackend;
#[cfg(not(target_arch = "wasm32"))]
use crate::storage::backend::file::FileBackend;
use crate::storage::persistent_facts::PersistentFactStorage;
#[cfg(not(target_arch = "wasm32"))]
use crate::wal::WalWriter;
use anyhow::{Result, bail};
use std::any::Any;
#[cfg(not(target_arch = "wasm32"))]
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, RwLock};
// ─── Thread-local reentrant-write detection ─────────────────────────────────
thread_local! {
/// Set to `true` while a `WriteTransaction` is active on this thread.
/// Prevents same-thread deadlock when `db.execute()` is called while
/// a `WriteTransaction` is in progress.
static WRITE_TX_ACTIVE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
fn set_write_tx_active(val: bool) {
WRITE_TX_ACTIVE.with(|f| f.set(val));
}
fn is_write_tx_active() -> bool {
WRITE_TX_ACTIVE.with(|f| f.get())
}
// ─── OpenOptions ─────────────────────────────────────────────────────────────
/// Configuration options for opening a `Minigraf` database.
#[derive(Debug, Clone)]
pub struct OpenOptions {
/// Number of WAL entries committed before an automatic checkpoint is triggered.
///
/// Defaults to 1000. Lower values mean more frequent checkpoints (smaller WAL,
/// more I/O). Higher values mean less frequent checkpoints (larger WAL, less I/O).
pub wal_checkpoint_threshold: usize,
/// Number of pages to hold in the LRU page cache. Default: 256 (= 1MB at 4KB pages).
pub page_cache_size: usize,
/// Maximum facts that can be derived per recursive rule iteration.
/// Defaults to 1_000_000. Use to prevent runaway recursive rules.
pub max_derived_facts: usize,
/// Maximum total query results. Defaults to 1_000_000.
pub max_results: usize,
}
impl Default for OpenOptions {
fn default() -> Self {
OpenOptions {
wal_checkpoint_threshold: 1000,
page_cache_size: 256,
max_derived_facts: DEFAULT_MAX_DERIVED_FACTS,
max_results: DEFAULT_MAX_RESULTS,
}
}
}
impl OpenOptions {
/// Create a new `OpenOptions` with default settings.
pub fn new() -> Self {
Self::default()
}
/// Set the number of pages to hold in the LRU page cache.
///
/// Each page is 4KB, so the default of 256 pages uses ~1MB of memory.
pub fn page_cache_size(mut self, size: usize) -> Self {
self.page_cache_size = size;
self
}
/// Set the maximum facts that can be derived per recursive rule iteration.
///
/// Defaults to 1_000_000. Use lower values to prevent runaway recursive rules
/// from consuming excessive memory. Can be overridden per-query using
/// `:max-derived-facts N` in the query vector.
pub fn max_derived_facts(mut self, n: usize) -> Self {
self.max_derived_facts = n;
self
}
/// Set the maximum total query results.
///
/// Defaults to 1_000_000. Use lower values to limit result set size.
pub fn max_results(mut self, n: usize) -> Self {
self.max_results = n;
self
}
/// Set the path for a file-backed database.
#[cfg(not(target_arch = "wasm32"))]
pub fn path(self, path: impl AsRef<Path>) -> OpenOptionsWithPath {
OpenOptionsWithPath {
opts: self,
path: path.as_ref().to_path_buf(),
}
}
/// Open an in-memory (non-persistent) database.
///
/// Uses the options set on the builder. WAL-related options are ignored.
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn open_memory(self) -> Result<Minigraf> {
Minigraf::in_memory_with_options(self)
}
}
/// `OpenOptions` combined with a file path, ready to open.
#[cfg(not(target_arch = "wasm32"))]
pub struct OpenOptionsWithPath {
opts: OpenOptions,
path: PathBuf,
}
#[cfg(not(target_arch = "wasm32"))]
impl OpenOptionsWithPath {
/// Open or create the file-backed database.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
pub fn open(self) -> Result<Minigraf> {
Minigraf::open_with_options(self.path, self.opts)
}
}
// ─── WriteContext ─────────────────────────────────────────────────────────────
/// Internal write context: distinguishes in-memory from file-backed databases.
enum WriteContext {
/// In-memory database: no WAL, no persistence.
Memory,
/// File-backed database: has a WAL sidecar and a persistent storage layer.
#[cfg(not(target_arch = "wasm32"))]
File {
pfs: PersistentFactStorage<FileBackend>,
/// WAL writer. `None` after a checkpoint until the next write.
wal: Option<WalWriter>,
db_path: PathBuf,
/// Count of WAL entries written since the last checkpoint (or since open).
wal_entry_count: usize,
},
}
// ─── Inner ────────────────────────────────────────────────────────────────────
struct Inner {
/// The shared in-memory fact store. Cloning is cheap (Arc-based).
fact_storage: FactStorage,
/// Shared rule registry, persists across all `execute()` calls.
rules: Arc<RwLock<RuleRegistry>>,
/// Function registry for aggregates and window functions.
/// `RwLock` is used in anticipation of the 7.7b `register_aggregate`/`register_predicate` mutation API.
functions: Arc<RwLock<FunctionRegistry>>,
/// Serialises all writes. Holds `WriteContext` which contains the PFS/WAL
/// for file-backed databases.
write_lock: Mutex<WriteContext>,
/// Configuration options.
options: OpenOptions,
}
impl Drop for Inner {
fn drop(&mut self) {
// On clean close, perform a best-effort checkpoint to reduce WAL size.
// Errors are silently ignored (can't propagate from Drop).
// Skip if wal_checkpoint_threshold is usize::MAX — that sentinel suppresses
// all checkpointing (used by benchmarks to keep WAL entries pending).
if self.options.wal_checkpoint_threshold == usize::MAX {
return;
}
if let Ok(mut ctx) = self.write_lo 8 ck.lock( 10BC8 ) {
let _ = Minigraf::do_checkpoint(&self.fact_storage, &mut ctx);
}
}
}
// ─── Fact size validation (moved to WAL serialization) ────────────────────────
// ─── Minigraf ─────────────────────────────────────────────────────────────────
/// The primary embedded graph database handle.
///
/// `Minigraf` is cheap to clone — all clones share the same underlying database.
///
/// # File-backed usage
/// ```no_run
/// # #[cfg(not(target_arch = "wasm32"))] {
/// use minigraf::db::Minigraf;
///
/// let db = Minigraf::open("mydb.graph").unwrap();
/// db.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// # }
/// ```
///
/// # In-memory usage
/// ```
/// use minigraf::db::Minigraf;
///
/// let db = Minigraf::in_memory().unwrap();
/// db.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// ```
///
/// # Fact Size Limit (file-backed databases only)
///
/// Each fact persisted to a `.graph` file must serialise to at most
/// 4 080 bytes (the per-page capacity limit).
///
/// In practice, `Value::String` content is limited to roughly **3 900–4 000 bytes**
/// depending on entity and attribute name lengths.
///
/// Facts that exceed this limit are rejected at insertion time with a descriptive
/// error. This check does **not** apply to `Minigraf::in_memory()`.
///
/// ## Workarounds for large payloads
///
/// - **External blob reference** — store the payload in a file or object store
/// and record its path, URL, or content-addressed hash as a `Value::String`:
/// ```text
/// (transact [[:doc123 :blob/sha256 "a3f5c9..."]])
/// ```
/// - **Entity decomposition** — split large values across multiple facts using
/// a continuation-entity pattern.
/// - **In-memory database** — `Minigraf::in_memory()` has no fact size limit
/// and is suitable for workloads that do not require persistence.
#[derive(Clone)]
pub struct Minigraf {
inner: Arc<Inner>,
}
impl Minigraf {
// ── Constructors ─────────────────────────────────────────────────────────
/// Open or create a file-backed database with default options.
///
/// A sidecar WAL file (`<path>.wal`) is created alongside the main file.
/// Any existing WAL from a previous crash is replayed automatically.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
#[cfg(not(target_arch = "wasm32"))]
pub fn open(path: impl AsRef<Path>) -> Result<Self> {
Self::open_with_options(path, OpenOptions::default())
}
/// Open or create a file-backed database with custom options.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
#[cfg(not(target_arch = "wasm32"))]
pub fn open_with_options(path: impl AsRef<Path>, opts: OpenOptions) -> Result<Self> {
let db_path = path.as_ref().to_path_buf();
// Open the main .graph file
let backend = FileBackend::open(&db_path)?;
let pfs = PersistentFactStorage::new(backend, opts.page_cache_size)?;
// Share the fact storage
let fact_storage = pfs.storage().clone();
// Derive WAL path: "<db_path>.wal"
let wal_path = Self::wal_path_for(&db_path);
// Replay any existing WAL entries before opening the writer
let wal_entry_count = Self::replay_wal(&wal_path, &fact_storage, &pfs)?;
// Open the WAL writer only if the WAL file already exists from a previous session.
// Otherwise, create it lazily on the first write.
let wal = if wal_path.exists() {
Some(WalWriter::open_or_create(&wal_path)?)
} else {
None
};
let ctx = WriteContext::File {
pfs,
wal,
db_path,
wal_entry_count,
};
Ok(Minigraf {
inner: Arc::new(Inner {
fact_storage,
rules: Arc::new(RwLock::new(RuleRegistry::new())),
functions: Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
write_lock: Mutex::new(ctx),
options: opts,
}),
})
}
/// Create an in-memory database (no WAL, no persistence). Suitable for tests and REPL.
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn in_memory() -> Result<Self> {
Self::in_memory_with_options(OpenOptions::default())
}
/// Create an in-memory database with custom options.
///
/// Note: WAL-related options are ignored for in-memory databases.
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn in_memory_with_options(opts: OpenOptions) -> Result<Self> {
let backend = MemoryBackend::new();
let pfs = PersistentFactStorage::new(backend, opts.page_cache_size)?;
let fact_storage = pfs.storage().clone();
// For in-memory databases we don't need the PFS beyond initialisation;
// we just use the shared FactStorage directly.
drop(pfs);
Ok(Minigraf {
inner: Arc::new(Inner {
fact_storage,
rules: Arc::new(RwLock::new(RuleRegistry::new())),
functions: Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
write_lock: Mutex::new(WriteContext::Memory),
options: opts,
}),
})
}
// ── WAL replay helper ────────────────────────────────────────────────────
/// Replay any WAL entries that are newer than the main file's checkpoint.
///
/// Returns the number of entries replayed (used to seed `wal_entry_count`).
#[cfg(not(target_arch = "wasm32"))]
fn replay_wal(
wal_path: &Path,
fact_storage: &FactStorage,
pfs: &PersistentFactStorage<FileBackend>,
) -> Result<usize> {
if !wal_path.exists() {
return Ok(0);
}
let mut reader = crate::wal::WalReader::open(wal_path)?;
let entries = reader.read_entries()?;
let last_checkpointed = pfs.last_checkpointed_tx_count();
let mut replayed = 0;
for entry in &entries {
if entry.tx_count <= last_checkpointed {
// Already present in the main file; skip.
continue;
}
for fact in &entry.facts {
let _ = fact_storage.load_fact(fact.clone())?;
}
#[allow(clippy::arithmetic_side_effects)]
{
replayed += 1;
}
}
// Re-synchronise tx_counter to the maximum tx_count across all facts
fact_storage.restore_tx_counter()?;
Ok(replayed)
}
// ── Execute ──────────────────────────────────────────────────────────────
/// Execute a Datalog command as a self-contained implicit transaction.
///
/// For file-backed databases, the WAL entry is written **before** facts are
/// applied to the in-memory store. A successful return means the facts are in
/// both the WAL and the in-memory store; a crash after this call returns will
/// replay the facts on next open.
///
/// If the WAL write fails, an error is returned and the in-memory store is
/// left unchanged. The database remains consistent for subsequent in-process
/// operations.
///
/// Returns `Err` if called from the same thread that holds an active
/// `WriteTransaction` (use `tx.execute()` instead).
///
/// # Errors
///
/// Returns an error if:
/// - A `WriteTransaction` is already active on **this thread** (deadlock prevention).
/// - Parsing fails.
/// - Execution fails.
/// - WAL write fails (file-backed databases).
pub fn execute(&self, input: &str) -> Result<QueryResult> {
// Detect same-thread reentrant write (would deadlock on the Mutex).
if is_write_tx_active() {
bail!(
"a WriteTransaction is already in progress on this thread; use tx.execute() instead"
);
}
let cmd = parse_datalog_command(input).map_err(|e| anyhow::anyhow!("{}", e))?;
// Determine if this is a read-only command (query only).
// Rule registration is treated as a write because it mutates the shared RuleRegistry.
let is_write = matches!(
cmd,
DatalogCommand::Transact(_) | DatalogCommand::Retract(_) | DatalogCommand::Rule(_)
);
if is_write {
let mut ctx = self.inner.write_lock.lock().map_err(|_| {
anyhow::anyhow!("write lock is poisoned; database may be in an inconsistent state")
})?;
// Handle write commands with correct WAL-first ordering for Transact/Retract:
// 1. Materialize facts (no storage mutation yet)
// 2. Allocate tx_count + tx_id and stamp facts
// 3. Write WAL entry FIRST — if this fails, FactStorage is unchanged
// 4. Apply facts to shared FactStorage
// For Rule registration: acquire lock to serialize rule changes, no WAL needed
let (stamped, is_retract) = match &cmd {
DatalogCommand::Transact(tx) => (Minigraf::materialize_transaction(tx)?, false),
DatalogCommand::Retract(tx) => (Minigraf::materialize_retraction(tx)?, true),
DatalogCommand::Rule(_) => {
// Rule registration: execute while holding write_lock to serialize
// rule registry changes (no WAL needed for rules)
let executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
return executor.execute(cmd);
}
_ => return Err(anyhow::anyhow!("unexpected command variant in write path")),
};
let tx_count = self.inner.fact_storage.allocate_tx_count();
let tx_id = crate::graph::types::tx_id_now();
let stamped: Vec<Fact> = stamped
.into_iter()
.map(|mut f| {
f.tx_id = tx_id;
f.tx_count = tx_count;
// Fix valid_from if it was left as the sentinel
if f.asserted && f.valid_from == VALID_FROM_USE_TX_TIME {
f.valid_from = tx_id.cast_signed();
}
f
})
.collect();
// WAL write includes size validation — no separate check needed.
// Write WAL BEFORE applying to shared FactStorage.
// If this fails, FactStorage is still unchanged — clean rollback.
let should_checkpoint = WriteTransaction::wal_write_stamped_batch(
&mut ctx,
&self.inner.options,
tx_count,
&stamped,
)?;
// WAL succeeded — now apply facts to shared FactStorage.
for fact in &stamped {
let _ = self.inner.fact_storage.load_fact(fact.clone())?;
}
// Trigger auto-checkpoint AFTER facts are in FactStorage so the
// checkpoint captures the newly written facts.
if should_checkpoint {
Minigraf::do_checkpoint(&self.inner.fact_storage, &mut ctx)?;
}
// Return the same QueryResult the executor would have returned.
if is_retract {
Ok(QueryResult::Retracted(tx_id))
} else {
Ok(QueryResult::Transacted(tx_id))
}
} else {
// Read-only: no lock needed
let mut executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.set_limits(
self.inner.options.max_derived_facts,
self.inner.options.max_results,
);
executor.execute(cmd)
}
}
// ── Explicit transaction ──────────────────────────────────────────────────
/// Begin an explicit write transaction.
///
/// Acquires the write lock; held until `commit()`, `rollback()`, or drop.
///
/// # Errors
///
/// Returns an error if a `WriteTransaction` is already active on **this thread**.
pub fn begin_write(&self) -> Result<WriteTransaction<'_>> {
if is_write_tx_active() {
bail!(
"a WriteTransaction is already in progress on this thread; use tx.execute() instead"
);
}
let guard = self.inner.write_lock.lock().map_err(|_| {
anyhow::anyhow!("write lock is poisoned; database may be in an inconsistent state")
})?;
// Set flag only after successfully acquiring the lock
set_write_tx_active(true);
Ok(WriteTransaction {
guard,
inner: &self.inner,
pending_facts: Vec::new(),
next_pending_tx_count: self.inner.fact_storage.current_tx_count().saturating_add(1),
next_pending_tx_id: crate::graph::types::tx_id_now(),
committed: false,
})
}
// ── Checkpoint ───────────────────────────────────────────────────────────
/// Manually trigger a checkpoint: flush all in-memory facts to the main file
/// and delete the WAL sidecar.
///
/// No-op for in-memory databases.
///
/// # Errors
///
/// Returns an error if the write lock is poisoned or the checkpoint I/O fails.
pub fn checkpoint(&self) -> Result<()> {
let mut ctx = self.inner.write_lock.lock().map_err(|_| {
anyhow::anyhow!("write lock is poisoned; database may be in an inconsistent state")
})?;
Self::do_checkpoint(&self.inner.fact_storage, &mut ctx)
}
/// Returns the current monotonic transaction counter.
///
/// This is the value that `:as-of N` compares against. After a successful
/// [`execute`](Self::execute) that returns [`QueryResult::Transacted`] or
/// [`QueryResult::Retracted`], this reflects the count of that transaction.
///
/// Starts at 0 for a new database and increments once per `transact`/`retract` call,
/// regardless of how many facts the batch contains.
pub fn current_tx_count(&self) -> u64 {
self.inner.fact_storage.current_tx_count()
}
/// Internal checkpoint logic (operates on an already-held write-lock guard).
fn do_checkpoint(_fact_storage: &FactStorage, ctx: &mut WriteContext) -> Result<()> {
match ctx {
WriteContext::Memory => {
// No-op for in-memory databases.
}
#[cfg(not(target_arch = "wasm32"))]
WriteContext::File {
pfs,
wal,
db_path,
wal_entry_count,
} => {
// Skip checkpoint if nothing to flush.
//
// `wal_entry_count` is non-zero when this handle has made writes *or*
// replayed WAL entries on open (crash-recovery path). `pfs.is_dirty()`
// catches any facts marked dirty via the normal write path.
//
// File locking (`.graph.lock` sidecar, acquired by FileBackend::open)
// already prevents a second *process* from opening the file while this
// handle holds the lock, which covers the main multi-process exposure
// described in issue #226. This guard closes the remaining edge cases:
// same-process double-opens (same PID bypasses the stale-lock check)
// and environments where the advisory lock can be bypassed (e.g.
// network filesystems, manual lock deletion).
if *wal_entry_count == 0 && !pfs.is_dirty() {
return Ok(());
}
// `force_dirty` is needed for the WAL-replay case: facts were loaded
// into memory during `replay_wal` but `pfs.dirty` was not set because
// no write path was exercised. Without it `save()` would no-op and
// the replayed facts would never reach the main file.
pfs.force_dirty();
pfs.save()?;
// Derive WAL path and delete the sidecar.
let wal_path = Self::wal_path_for(db_path);
// Close the WAL writer (drop it) before deleting the file.
*wal = None;
if wal_path.exists() {
WalWriter::delete_file(&wal_path)?;
}
// WAL writer will be recreated lazily on the next write.
*wal_entry_count = 0;
}
}
Ok(())
}
/// Parse and plan a query once; bind slots (`$name`) are left unresolved.
///
/// Returns a [`crate::query::datalog::prepared::PreparedQuery`] that can be executed
/// many times with different bind values via
/// [`crate::query::datalog::prepared::PreparedQuery::execute`].
///
/// # Errors
/// - Parse failure.
/// - A bind slot appears in an attribute position (rejected at prepare time).
/// - The command is not a `(query ...)` — `transact`, `retract`, and `rule`
/// are not preparable.
pub fn prepare(
&self,
query_str: &str,
) -> Result<crate::query::datalog::prepared::PreparedQuery> {
use crate::query::datalog::prepared::prepare_query;
let cmd = parse_datalog_command(query_str).map_err(|e| anyhow::anyhow!("{}", e))?;
let query = match cmd {
DatalogCommand::Query(q) => q,
DatalogCommand::Transact(_) => {
anyhow::bail!("only (query ...) commands can be prepared; got transact")
}
DatalogCommand::Retract(_) => {
anyhow::bail!("only (query ...) commands can be prepared; got retract")
}
DatalogCommand::Rule(_) => {
anyhow::bail!("only (query ...) commands can be prepared; got rule")
}
};
prepare_query(
query,
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
)
}
/// Return an interactive REPL that reads commands from stdin.
///
/// The REPL borrows the database for the duration of the session.
/// Call [`crate::repl::Repl::run`] to start the interactive loop.
///
/// # Example
/// ```no_run
/// # use minigraf::Minigraf;
/// let db = Minigraf::in_memory().unwrap();
/// db.repl().run();
/// ```
pub fn repl(&self) -> crate::repl::Repl<'_> {
crate::repl::Repl::new(self)
}
/// Compute the WAL sidecar path for a given database path.
#[cfg(not(target_arch = "wasm32"))]
fn wal_path_for(db_path: &Path) -> PathBuf {
let mut p = db_path.to_path_buf();
let name = p
.file_name()
.map(|n| {
let mut s = n.to_os_string();
s.push(".wal");
s
})
.unwrap_or_else(|| std::ffi::OsString::from("db.graph.wal"));
p.set_file_name(name);
p
}
// ── Materialize helpers ───────────────────────────────────────────────────
/// Convert a `Transaction` into a list of assertion `Fact`s (tx_id and tx_count
/// are set to 0 as placeholders; they are assigned at commit time).
pub(crate) fn materialize_transaction(tx: &Transaction) -> Result<Vec<Fact>> {
use crate::query::datalog::matcher::{edn_to_entity_id, edn_to_value};
use crate::query::datalog::types::EdnValue;
let tx_valid_from = tx.valid_from;
let tx_valid_to = tx.valid_to;
let mut facts = V 10BC8 ec::new();
for pattern in &tx.facts {
let entity = edn_to_entity_id(&pattern.entity)
.map_err(|e| anyhow::anyhow!("invalid entity: {}", e))?;
let attr = match &pattern.attribute {
AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
AttributeSpec::Real(_) => anyhow::bail!("attribute must be a keyword"),
AttributeSpec::Pseudo(_) => anyhow::bail!("cannot transact a pseudo-attribute"),
};
let value = edn_to_value(&pattern.value)
.map_err(|e| anyhow::anyhow!("invalid value: {}", e))?;
let valid_from = pattern
.valid_from
.or(tx_valid_from)
.unwrap_or(VALID_FROM_USE_TX_TIME);
let valid_to = pattern
.valid_to
.or(tx_valid_to)
.unwrap_or(VALID_TIME_FOREVER);
facts.push(Fact::with_valid_time(
entity, attr, value, 0, 0, valid_from, valid_to,
));
}
Ok(facts)
}
/// Convert a `Transaction` into a list of retraction `Fact`s.
pub(crate) fn materialize_retraction(tx: &Transaction) -> Result<Vec<Fact>> {
use crate::query::datalog::matcher::{edn_to_entity_id, edn_to_value};
use crate::query::datalog::types::EdnValue;
let mut facts = Vec::new();
for pattern in &tx.facts {
let entity = edn_to_entity_id(&pattern.entity)
.map_err(|e| anyhow::anyhow!("invalid entity: {}", e))?;
let attr = match &pattern.attribute {
AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
AttributeSpec::Real(_) => anyhow::bail!("attribute must be a keyword"),
AttributeSpec::Pseudo(_) => anyhow::bail!("cannot transact a pseudo-attribute"),
};
let value = edn_to_value(&pattern.value)
.map_err(|e| anyhow::anyhow!("invalid value: {}", e))?;
let mut f = Fact::retract(entity, attr, value, 0);
f.tx_count = 0;
facts.push(f);
}
Ok(facts)
}
// ── UDF registration ─────────────────────────────────────────────────────
/// Register a custom aggregate function.
///
/// `Acc` is any `Send + 'static` type that serves as the accumulator.
/// It is type-erased internally. The function is usable in both `:find`
/// grouping position and `:over` (window) position.
///
/// # Errors
///
/// Returns an error if `name` is already registered (built-in or UDF)
/// or the function registry lock is poisoned.
///
/// # Example
/// ```
/// # use minigraf::db::Minigraf;
/// # use minigraf::Value;
/// let db = Minigraf::in_memory().unwrap();
/// db.register_aggregate(
/// "mysum",
/// || 0i64,
/// |acc: &mut i64, v: &Value| { if let Value::Integer(i) = v { *acc += i; } },
/// |acc: &i64, _n: usize| Value::Integer(*acc),
/// ).unwrap();
/// ```
pub fn register_aggregate<Acc>(
&self,
name: &str,
init: impl Fn() -> Acc + Send + Sync + 'static,
step: impl Fn(&mut Acc, &Value) + Send + Sync + 'static,
finalise: impl Fn(&Acc, usize) -> Value + Send + Sync + 'static,
) -> Result<()>
where
Acc: Any + Send + 'static,
{
let init_boxed: Arc<dyn Fn() -> Box<dyn Any + Send> + Send + Sync> =
Arc::new(move || Box::new(init()) as Box<dyn Any + Send>);
let step_boxed: UdfStepFn = Arc::new(move |acc, v| {
// `init_boxed` always creates `Box<Acc>`, so downcast is infallible.
if let Some(typed_acc) = acc.downcast_mut::<Acc>() {
step(typed_acc, v);
}
});
let finalise_boxed: UdfFinaliseFn = Arc::new(move |acc, n| {
acc.downcast_ref::<Acc>()
.map(|typed_acc| finalise(typed_acc, n))
.unwrap_or(Value::Null)
});
let desc = AggregateDesc {
impl_: AggImpl::Udf(UdfOps {
init: init_boxed,
step: step_boxed,
finalise: finalise_boxed,
}),
is_builtin: false,
};
self.inner
.functions
.write()
.map_err(|e| anyhow::anyhow!("function registry lock poisoned: {}", e))?
.register_aggregate_desc(name.to_string(), desc)
}
/// Register a custom single-argument filter predicate.
///
/// The predicate is usable in `[(name? ?var)]` `:where` clauses.
///
/// # Errors
///
/// Returns an error if `name` is already registered (built-in or UDF)
/// or the function registry lock is poisoned.
///
/// # Example
/// ```
/// # use minigraf::db::Minigraf;
/// # use minigraf::Value;
/// let db = Minigraf::in_memory().unwrap();
/// db.register_predicate(
/// "email?",
/// |v: &Value| matches!(v, Value::String(s) if s.contains('@')),
/// ).unwrap();
/// ```
pub fn register_predicate(
&self,
name: &str,
f: impl Fn(&Value) -> bool + Send + Sync + 'static,
) -> Result<()> {
let desc = PredicateDesc {
f: Arc::new(f),
is_builtin: false,
};
self.inner
.functions
.write()
.map_err(|e| anyhow::anyhow!("function registry lock poisoned: {}", e))?
.register_predicate_desc(name.to_string(), desc)
}
}
// ─── WriteTransaction ─────────────────────────────────────────────────────────
/// An explicit write transaction. Holds the write lock for its lifetime.
///
/// # Usage
/// ```no_run
/// use minigraf::db::Minigraf;
///
/// let db = Minigraf::in_memory().unwrap();
/// let mut tx = db.begin_write().unwrap();
/// tx.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// tx.execute(r#"(transact [[:alice :person/age 30]])"#).unwrap();
/// tx.commit().unwrap();
/// ```
///
/// Dropping without committing performs an implicit rollback.
pub struct WriteTransaction<'a> {
guard: MutexGuard<'a, WriteContext>,
inner: &'a Inner,
/// Facts buffered in this transaction (not yet committed to FactStorage).
pending_facts: Vec<Fact>,
/// Synthetic tx_count assigned to the next staged write batch.
next_pending_tx_count: u64,
/// Synthetic tx_id seed used to keep staged writes monotonic.
next_pending_tx_id: TxId,
/// Set to `true` after a successful `commit()` to suppress rollback in `Drop`.
committed: bool,
}
impl<'a> WriteTransaction<'a> {
/// Execute a Datalog command within this transaction.
///
/// - **Writes** (transact / retract): buffered in-memory; not durable until `commit()`.
/// Returns `Ok(QueryResult::Ok)` immediately (not `Transacted`/`Retracted`).
/// - **Reads** (query): see committed facts in `FactStorage` **plus** all facts
/// buffered in this transaction (read-your-own-writes).
/// - **Rules**: registered immediately into the shared rule registry.
///
/// # Errors
///
/// Returns an error if parsing or execution fails.
pub fn execute(&mut self, input: &str) -> Result<QueryResult> {
let cmd = parse_datalog_command(input).map_err(|e| anyhow::anyhow!("{}", e))?;
match cmd {
DatalogCommand::Transact(tx) => {
self.stage_pending_facts(Minigraf::materialize_transaction(&tx)?);
Ok(QueryResult::Ok)
}
DatalogCommand::Retract(tx) => {
self.stage_pending_facts(Minigraf::materialize_retraction(&tx)?);
Ok(QueryResult::Ok)
}
DatalogCommand::Query(_) => self.execute_read_command(cmd),
DatalogCommand::Rule(rule) => self.execute_rule_command(rule),
}
}
fn execute_read_command(&self, cmd: DatalogCommand) -> Result<QueryResult> {
if self.pending_facts.is_empty() {
let mut executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.set_limits(
self.inner.options.max_derived_facts,
self.inner.options.max_results,
);
return executor.execute(cmd);
}
let merged_facts = self.merged_query_facts()?;
let mut executor = DatalogExecutor::new_from_facts_with_rules_and_functions(
merged_facts,
self.pending_read_now_floor(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.set_limits(
self.inner.options.max_derived_facts,
self.inner.options.max_results,
);
executor.execute(cmd)
}
fn execute_rule_command(
&self,
rule: crate::query::datalog::types::Rule,
) -> Result<QueryResult> {
let mut executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.set_limits(
self.inner.options.max_derived_facts,
self.inner.options.max_results,
);
executor.execute(DatalogCommand::Rule(rule))
}
/// Stage buffered facts with stable synthetic read metadata.
fn stage_pending_facts(&mut self, facts: Vec<Fact>) {
let staged_tx_id = std::cmp::max(crate::graph::types::tx_id_now(), self.next_pending_tx_id);
let staged_tx_count = self.next_pending_tx_count;
self.pending_facts.extend(facts.into_iter().map(|mut fact| {
fact.tx_id = staged_tx_id;
fact.tx_count = staged_tx_count;
fact
}));
self.next_pending_tx_id = staged_tx_id.saturating_add(1);
self.next_pending_tx_count = self.next_pending_tx_count.saturating_add(1);
}
View remainder of file in raw view
0