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

db.rs

Latest commit

 

History

History
3507 lines (3176 loc) · 139 KB

db.rs

File metadata and controls

3507 lines (3176 loc) · 139 KB
Raw
///
/// 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, MinigrafError> {
self.execute_inner(input).map_err(MinigrafError::from)
}
fn execute_inner(&self, input: &str) -> Result<QueryResult> {
// Detect same-thread reentrant write (would deadlock on the Mutex).
if is_write_tx_active() {
bail_coded!(ErrorCode::Int001);
}
if let Some(kind) = statement::kind_of(input) {
return self.execute_statement(input, kind);
}
let cmd = parse_datalog_command(input)?;
match cmd {
DatalogCommand::Query(ref q) => {
check_overrides(q, &self.inner.options.allow_overrides)?;
reject_unbound_slots(q)?;
// Read-only: no lock needed
let snapshot = self.inner.fact_storage.snapshot()?;
self.read_executor(snapshot).execute(cmd)
}
DatalogCommand::Rule(_) => {
// Rule registration is a write: it mutates the shared
// RuleRegistry, so it runs under write_lock (no WAL needed).
let _ctx = self
.inner
.write_lock
.lock()
.map_err(|_| err_coded!(ErrorCode::Api001))?;
let executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.execute(cmd)
}
DatalogCommand::Transact(_) | DatalogCommand::Retract(_) => {
Err(err_coded!(ErrorCode::Api002))
}
}
}
/// Execute a `transact` or `retract` with correct WAL-first ordering:
/// 1. Read the statement (pass 1, #550) and materialize its 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
fn execute_statement(&self, input: &str, kind: Kind) -> Result<QueryResult> {
// Before the lock, as parsing is: a malformed statement never waits.
// Its materialisation and window errors come with its facts, below.
let statement = statement::read_statement(
input,
kind,
crate::graph::types::tx_id_now(),
self.chunk_facts(),
)?;
self.reject_if_read_only(kind.name())?;
{
let mut ctx = self
.inner
.write_lock
.lock()
.map_err(|_| err_coded!(ErrorCode::Api001))?;
#[cfg(not(target_arch = "wasm32"))]
if statement.is_large() && matches!(*ctx, WriteContext::File { .. }) {
return self.execute_large(&mut ctx, statement, kind);
}
let stamped = statement.facts()?;
let is_retract = kind == Kind::Retract;
// Before allocating: a rejected transaction takes no tx_count.
let tx_id = crate::graph::types::tx_id_now();
crate::graph::storage::check_valid_windows(&stamped, tx_id)?;
crate::graph::storage::check_one_window_per_triple(&stamped)?;
refuse_if_failed(&ctx)?;
let tx_count = self.inner.fact_storage.allocate_tx_count();
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, in one
// batch so no query sees part of the transaction (#554).
let _ = self.inner.fact_storage.load_facts(stamped)?;
// 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))
}
}
}
/// A statement with more facts than one chunk, on a file database (#550
/// spec §4): written a chunk at a time into copy-on-write working trees
/// and committed as a checkpoint, with no WAL entry. Its `tx_id` is the
/// time it became large.
#[cfg(not(target_arch = "wasm32"))]
fn execute_large(
&self,
ctx: &mut WriteContext,
statement: statement::Statement<'_>,
kind: Kind,
) -> Result<QueryResult> {
use crate::storage::persistent_facts::ChunkError;
let chunks = statement.chunks()?;
refuse_if_failed(ctx)?;
let WriteContext::File {
pfs,
wal,
db_path,
wal_entry_count,
} = ctx
else {
bail_coded!(ErrorCode::Int019, "a large statement without a file");
};
// The WAL guard: a crash before the commit point finds a WAL whose
// base is this generation, never an orphan `g + 1` page (§4.5).
Self::open_wal(
wal,
db_path,
self.inner.options.synchronous,
pfs.generation(),
)?;
let mut tx = pfs.begin_large(crate::graph::types::tx_id_now())?;
for chunk in chunks {
let written = match chunk {
Ok(mut chunk) => pfs.write_large_chunk(&mut tx, &mut chunk),
Err(e) => Err(ChunkError::Statement(e)),
};
if let Err(ChunkError::Statement(e) | ChunkError::Failure(e)) = written {
pfs.rollback_large(tx);
return Err(e);
}
}
let tx_id = tx.tx_id();
pfs.commit_large(tx)?;
Self::delete_wal(wal, db_path, wal_entry_count)?;
Ok(match kind {
Kind::Retract => QueryResult::Retracted(tx_id),
Kind::Transact => QueryResult::Transacted(tx_id),
})
}
/// The WAL writer, opening the file (or creating it with the header only,
/// base generation `generation`) if it is not open.
#[cfg(not(target_arch = "wasm32"))]
fn open_wal<'w>(
wal: &'w mut Option<WalWriter>,
db_path: &Path,
sync: SyncMode,
generation: u64,
) -> Result<&'w mut WalWriter> {
if wal.is_none() {
let wal_path = Self::wal_path_for(db_path);
*wal = Some(WalWriter::open_or_create_at(&wal_path, sync, generation)?);
}
wal.as_mut().ok_or_else(|| err_coded!(ErrorCode::Api009))
}
/// The statement chunk size (`OpenOptions::transaction_chunk_facts`).
fn chunk_facts(&self) -> usize {
statement::tx_chunk_facts(self.inner.options.transaction_chunk_facts)
}
View remainder of file in raw view
0