// Copyright 2026 RadixDB Contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Database struct and operations
//!
//! Provides a modern, ergonomic Rust API for database operations.
//!
//! # Examples
//!
//! ```no_run
//! use radixdb_api::{Database, params};
//! # fn main() -> radixdb_core::Result<()> {
//!
//! let db = Database::open("memory://")?;
//!
//! // DDL - no params needed
//! db.execute("CREATE TABLE users (id INTEGER, name TEXT, age INTEGER)", ())?;
//!
//! // Insert with params - using tuple syntax
//! db.execute("INSERT INTO users VALUES ($1, $2, $3)", (1, "Alice", 30))?;
//!
//! // Insert with params! macro
//! db.execute("INSERT INTO users VALUES ($1, $2, $3)", params![2, "Bob", 25])?;
//!
//! // Query with iteration
//! for row in db.query("SELECT * FROM users WHERE age > $1", (20,))? {
//! let row = row?;
//! let name: String = row.get(1)?;
//! println!("{}", name);
//! }
//!
//! // Query single value
//! let count: i64 = db.query_one("SELECT COUNT(*) FROM users", ())?;
//! # assert_eq!(count, 2);
//! # Ok(())
//! # }
//! ```
use rustc_hash::FxHashMap;
use std::collections::HashSet;
use std::sync::{Arc, Condvar, Mutex, RwLock};
#[cfg(test)]
type DatabaseOpenTestHook = Arc<dyn Fn(&str) + Send + Sync>;
#[cfg(test)]
static DATABASE_OPEN_TEST_HOOK: std::sync::LazyLock<Mutex<Option<DatabaseOpenTestHook>>> =
std::sync::LazyLock::new(|| Mutex::new(None));
#[cfg(test)]
static DATABASE_OPEN_TEST_HOOK_OWNER: Mutex<()> = Mutex::new(());
#[cfg(test)]
struct DatabaseOpenTestHookGuard {
_owner: std::sync::MutexGuard<'static, ()>,
}
#[cfg(test)]
impl DatabaseOpenTestHookGuard {
fn install(hook: DatabaseOpenTestHook) -> Self {
let owner = DATABASE_OPEN_TEST_HOOK_OWNER
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
*DATABASE_OPEN_TEST_HOOK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
Self { _owner: owner }
}
}
#[cfg(test)]
impl Drop for DatabaseOpenTestHookGuard {
fn drop(&mut self) {
*DATABASE_OPEN_TEST_HOOK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
}
}
#[cfg(test)]
fn run_database_open_test_hook(dsn: &str) {
let hook = DATABASE_OPEN_TEST_HOOK
.lock()
.expect("database open test hook lock")
.clone();
if let Some(hook) = hook {
hook(dsn);
}
}
use radixdb_core::{Error, IsolationLevel, Result};
use radixdb_executor::context::{
clear_all_thread_local_caches, ExecutionContext, ExecutionContextBuilder,
};
use radixdb_executor::optimizer::FeedbackCache;
use radixdb_executor::query_cache::CachedPlanRef;
use radixdb_executor::semantic_cache::{SemanticCache, SemanticCacheStatsSnapshot};
use radixdb_executor::Executor;
use radixdb_executor::{DatabasePluginAdmission, PluginRegistry};
use radixdb_storage::mvcc::engine::MVCCEngine;
use radixdb_storage::traits::Engine;
use radixdb_storage::{Config, SyncMode};
use super::params::{NamedParams, Params};
use super::rows::{FromRow, Rows};
use super::statement::Statement;
use super::transaction::Transaction;
pub use super::value::FromValue;
/// Storage scheme constants
pub const MEMORY_SCHEME: &str = "memory";
pub const FILE_SCHEME: &str = "file";
fn composed_mvcc_engine(config: Config, plugin_registry: Arc<PluginRegistry>) -> MVCCEngine {
MVCCEngine::new_with_composition_binders(
config,
radixdb_executor::mutation::partial_index::bind_from_sql,
radixdb_executor::mutation::row_validation::bind,
radixdb_executor::mutation::view_binding::bind_from_sql,
radixdb_executor::plugin_catalog_runtime_binder(plugin_registry),
)
}
#[cfg(not(feature = "test-filedb"))]
fn composed_in_memory_engine() -> MVCCEngine {
composed_mvcc_engine(Config::default(), Arc::clone(&DEFAULT_PLUGIN_REGISTRY))
}
#[derive(Clone)]
enum DatabaseRegistryEntry {
Opening(Arc<DatabaseOpenSlot>),
Ready(Arc<DatabaseOwner>),
}
struct DatabaseOpenSlot {
outcome: Mutex<Option<Result<()>>>,
changed: Condvar,
}
impl DatabaseOpenSlot {
fn new() -> Self {
Self {
outcome: Mutex::new(None),
changed: Condvar::new(),
}
}
fn finish(&self, outcome: Result<()>) {
let mut state = self
.outcome
.lock()
.unwrap_or_else(|error| error.into_inner());
*state = Some(outcome);
self.changed.notify_all();
}
fn wait(&self) -> Result<()> {
let mut state = self
.outcome
.lock()
.map_err(|_| Error::LockAcquisitionFailed("database open slot".to_string()))?;
while state.is_none() {
state = self
.changed
.wait(state)
.map_err(|_| Error::LockAcquisitionFailed("database open slot".to_string()))?;
}
state
.as_ref()
.expect("database open slot outcome was checked")
.clone()
}
}
/// Global database registry to ensure single instance per DSN while allowing
/// independent DSNs to recover concurrently.
static DATABASE_REGISTRY: std::sync::LazyLock<RwLock<FxHashMap<String, DatabaseRegistryEntry>>> =
std::sync::LazyLock::new(|| RwLock::new(FxHashMap::default()));
static DEFAULT_PLUGIN_REGISTRY: std::sync::LazyLock<Arc<PluginRegistry>> =
std::sync::LazyLock::new(|| Arc::new(PluginRegistry::empty()));
/// Registry-owned durable engine/file-lock lifetime. It intentionally has no
/// SQL executor or connection-local transaction state.
struct DatabaseOwner {
engine: Arc<MVCCEngine>,
semantic_cache: Arc<SemanticCache>,
feedback_cache: Arc<FeedbackCache>,
plugin_registry: Arc<PluginRegistry>,
registry_key: String,
dsn: String,
requested_config: Config,
/// Temp directory for test-filedb feature. Deleted on drop.
#[cfg(feature = "test-filedb")]
_temp_dir: Option<tempfile::TempDir>,
}
/// Connection-local database state. Multiple connections may share the owner
/// and engine, but never this Executor or its hidden SQL transaction.
pub(crate) struct DatabaseInner {
engine: Arc<MVCCEngine>,
executor: Mutex<Executor>,
owner: Arc<DatabaseOwner>,
}
/// Type alias for Statement to use (avoids exposing DatabaseInner directly)
pub(crate) type DatabaseInnerHandle = DatabaseInner;
impl DatabaseInner {
/// Build a transaction-local executor while retaining the engine-owner
/// caches shared by every connection. Parsed plans and transaction state
/// stay local, but committed DML must invalidate the same semantic and
/// feedback caches that later readers consult.
pub(crate) fn transaction_executor(&self) -> Executor {
Executor::with_shared_runtime_caches_and_plugin_registry(
Arc::clone(&self.engine),
Arc::clone(&self.owner.semantic_cache),
Arc::clone(&self.owner.feedback_cache),
Arc::clone(&self.owner.plugin_registry),
)
}
}
impl Drop for DatabaseOwner {
fn drop(&mut self) {
if let Err(error) = self.engine.close_engine() {
eprintln!("automatic database close failed: {error}");
}
}
}
impl Drop for DatabaseInner {
fn drop(&mut self) {
clear_all_thread_local_caches();
// Statements and FFI objects retain this whole connection inner. Only
// its final drop releases the owner's last connection reference.
Database::try_unregister_owner(&self.owner);
}
}
/// Database represents a RadixDB database connection.
///
/// This is the main entry point for using RadixDB. It wraps the storage engine
/// and executor, providing a simple API for executing SQL queries.
///
/// # Thread Safety
///
/// Database is thread-safe and can be shared across threads via cloning.
/// Each clone shares the same underlying storage engine.
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, params};
///
/// // Open in-memory database
/// let db = Database::open("memory://")?;
///
/// // Create table
/// db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)", ())?;
///
/// // Insert with parameters
/// db.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
///
/// // Query
/// for row in db.query("SELECT * FROM users", ())? {
/// let row = row?;
/// println!("{}: {}", row.get::<i64>("id")?, row.get::<String>("name")?);
/// }
/// ```
pub struct Database {
inner: Arc<DatabaseInner>,
}
impl Database {
/// Authenticate one durable catalog Principal and return its stable ID.
#[doc(hidden)]
pub fn authenticate_principal(&self, login: &str, password: &str) -> Result<crate::ObjectId> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.authenticate_principal(login, password)
}
/// Return the bounded package-binding admission result without decoding
/// any external value or exposing storage internals.
#[doc(hidden)]
pub fn plugin_admission(&self) -> Result<DatabasePluginAdmission> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.plugin_admission()
}
#[doc(hidden)]
pub fn plugin_admission_diagnostic(&self) -> Result<Option<String>> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.plugin_admission_diagnostic()
}
pub(crate) fn with_connection_executor<T>(
&self,
operation: impl FnOnce(&Executor) -> Result<T>,
) -> Result<T> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
operation(&executor)
}
fn canonical_registry_key(dsn: &str) -> Result<String> {
let (scheme, path) = Self::parse_dsn(dsn)?;
if scheme == MEMORY_SCHEME {
return Ok(format!("{MEMORY_SCHEME}://{path}"));
}
let (clean_path, _) = Self::parse_file_config(&path)?;
let absolute = if std::path::Path::new(&clean_path).is_absolute() {
std::path::PathBuf::from(clean_path)
} else {
std::env::current_dir()
.map_err(|error| Error::internal(format!("cannot resolve database path: {error}")))?
.join(clean_path)
};
let mut normalized = std::path::PathBuf::new();
for component in absolute.components() {
match component {
std::path::Component::CurDir => {}
std::path::Component::ParentDir => {
normalized.pop();
}
other => normalized.push(other.as_os_str()),
}
}
Ok(format!("{FILE_SCHEME}://{}", normalized.display()))
}
fn requested_config(dsn: &str, registry_key: &str) -> Result<Config> {
let (scheme, path) = Self::parse_dsn(dsn)?;
if scheme == MEMORY_SCHEME {
return Ok(Config::in_memory());
}
let (_, mut config) = Self::parse_file_config(&path)?;
config.path = Some(
registry_key
.strip_prefix("file://")
.expect("file registry key must have file scheme")
.to_string(),
);
Ok(config)
}
fn ensure_matching_config(owner: &DatabaseOwner, requested: &Config, dsn: &str) -> Result<()> {
if &owner.requested_config == requested {
Ok(())
} else {
Err(Error::invalid_argument(format!(
"database `{dsn}` is already open with a different effective configuration"
)))
}
}
fn ensure_matching_plugin_registry(
owner: &DatabaseOwner,
requested: &Arc<PluginRegistry>,
dsn: &str,
) -> Result<()> {
if Arc::ptr_eq(&owner.plugin_registry, requested) {
Ok(())
} else {
Err(Error::invalid_argument(format!(
"database `{dsn}` is already open with a different plugin registry generation"
)))
}
}
pub(crate) fn ensure_open(&self) -> Result<()> {
match self.inner.engine.lifecycle_state() {
radixdb_storage::mvcc::engine::EngineLifecycleState::Ready => Ok(()),
radixdb_storage::mvcc::engine::EngineLifecycleState::CloseFailed(error)
| radixdb_storage::mvcc::engine::EngineLifecycleState::Failed(error) => Err(error),
_ => Err(Error::EngineNotOpen),
}
}
fn owner_arc(&self) -> &Arc<DatabaseOwner> {
&self.inner.owner
}
/// Build a connection-local facade around the registry-owned engine.
fn connection_from_owner(owner: Arc<DatabaseOwner>) -> Self {
let engine = Arc::clone(&owner.engine);
let executor = Executor::with_shared_runtime_caches_and_plugin_registry(
Arc::clone(&engine),
Arc::clone(&owner.semantic_cache),
Arc::clone(&owner.feedback_cache),
Arc::clone(&owner.plugin_registry),
);
Self {
inner: Arc::new(DatabaseInner {
engine,
executor: Mutex::new(executor),
owner,
}),
}
}
/// Remove an engine owner only after its final connection is released.
fn try_unregister_owner(owner: &Arc<DatabaseOwner>) {
if let Ok(mut registry) = DATABASE_REGISTRY.write() {
if let Some(DatabaseRegistryEntry::Ready(entry)) = registry.get(&owner.registry_key) {
if Arc::ptr_eq(entry, owner) && Arc::strong_count(owner) == 2 {
registry.remove(&owner.registry_key);
}
}
}
}
}
impl Clone for Database {
/// Clone with independent transaction state over the shared engine.
fn clone(&self) -> Self {
Self::connection_from_owner(Arc::clone(self.owner_arc()))
}
}
impl Drop for Database {
fn drop(&mut self) {
// DatabaseInner may be retained by a Statement/FFI object. Registry
// cleanup therefore belongs to its final Drop, not to this facade.
}
}
impl Database {
/// Open a memory or file DSN, reusing the engine owner for an identical DSN.
pub fn open(dsn: &str) -> Result<Self> {
Self::open_with_plugin_registry(dsn, Arc::clone(&DEFAULT_PLUGIN_REGISTRY))
}
/// Open against the immutable registry admitted before server bind.
#[doc(hidden)]
pub fn open_with_plugin_registry(
dsn: &str,
plugin_registry: Arc<PluginRegistry>,
) -> Result<Self> {
let registry_key = Self::canonical_registry_key(dsn)?;
let requested_config = Self::requested_config(dsn, ®istry_key)?;
let (opening_slot, owns_open) = {
let mut registry = DATABASE_REGISTRY
.write()
.map_err(|_| Error::LockAcquisitionFailed("registry write".to_string()))?;
match registry.get(®istry_key) {
Some(DatabaseRegistryEntry::Ready(inner)) => match inner.engine.lifecycle_state() {
radixdb_storage::mvcc::engine::EngineLifecycleState::Ready => {
Self::ensure_matching_config(inner, &requested_config, dsn)?;
Self::ensure_matching_plugin_registry(inner, &plugin_registry, dsn)?;
return Ok(Self::connection_from_owner(Arc::clone(inner)));
}
radixdb_storage::mvcc::engine::EngineLifecycleState::Closed => {
let slot = Arc::new(DatabaseOpenSlot::new());
registry.insert(
registry_key.clone(),
DatabaseRegistryEntry::Opening(Arc::clone(&slot)),
);
(slot, true)
}
radixdb_storage::mvcc::engine::EngineLifecycleState::CloseFailed(error)
| radixdb_storage::mvcc::engine::EngineLifecycleState::Failed(error) => {
return Err(error);
}
state => {
return Err(Error::internal(format!(
"database `{dsn}` is not ready for open: {state:?}"
)));
}
},
Some(DatabaseRegistryEntry::Opening(slot)) => (Arc::clone(slot), false),
None => {
let slot = Arc::new(DatabaseOpenSlot::new());
registry.insert(
registry_key.clone(),
DatabaseRegistryEntry::Opening(Arc::clone(&slot)),
);
(slot, true)
}
}
};
if !owns_open {
opening_slot.wait()?;
let registry = DATABASE_REGISTRY
.read()
.map_err(|_| Error::LockAcquisitionFailed("registry read".to_string()))?;
let Some(DatabaseRegistryEntry::Ready(inner)) = registry.get(®istry_key) else {
return Err(Error::internal(format!(
"database `{dsn}` open completed without publishing its owner"
)));
};
Self::ensure_matching_config(inner, &requested_config, dsn)?;
Self::ensure_matching_plugin_registry(inner, &plugin_registry, dsn)?;
return Ok(Self::connection_from_owner(Arc::clone(inner)));
}
#[cfg(test)]
run_database_open_test_hook(dsn);
let open_result = (|| -> Result<Arc<DatabaseOwner>> {
let (scheme, _path) = Self::parse_dsn(dsn)?;
#[cfg(feature = "test-filedb")]
let mut _temp_dir_holder: Option<tempfile::TempDir> = None;
let engine = match scheme.as_str() {
MEMORY_SCHEME => {
#[cfg(feature = "test-filedb")]
{
let tmp = tempfile::tempdir().map_err(|error| {
Error::internal(format!("failed to create temp dir: {error}"))
})?;
let file_dsn = format!("file://{}", tmp.path().display());
let (_clean_path, config) = Self::parse_file_config(&file_dsn[7..])?;
let engine = composed_mvcc_engine(config, Arc::clone(&plugin_registry));
engine.open_engine()?;
let engine = Arc::new(engine);
engine.start_cleanup();
_temp_dir_holder = Some(tmp);
engine
}
#[cfg(not(feature = "test-filedb"))]
{
let engine =
composed_mvcc_engine(Config::default(), Arc::clone(&plugin_registry));
engine.open_engine()?;
let engine = Arc::new(engine);
engine.start_cleanup();
engine
}
}
FILE_SCHEME => {
let engine = composed_mvcc_engine(
requested_config.clone(),
Arc::clone(&plugin_registry),
);
engine.open_engine()?;
let engine = Arc::new(engine);
engine.start_cleanup();
engine
}
_ => {
return Err(Error::parse(format!(
"Unsupported scheme '{scheme}'. Use 'memory://' or 'file://path'"
)));
}
};
Ok(Arc::new(DatabaseOwner {
engine,
semantic_cache: Arc::new(SemanticCache::new()),
feedback_cache: Arc::new(FeedbackCache::new()),
plugin_registry: Arc::clone(&plugin_registry),
registry_key: registry_key.clone(),
dsn: dsn.to_string(),
requested_config: requested_config.clone(),
#[cfg(feature = "test-filedb")]
_temp_dir: _temp_dir_holder,
}))
})();
match open_result {
Ok(inner) => {
let mut registry = match DATABASE_REGISTRY.write() {
Ok(registry) => registry,
Err(_) => {
let error = Error::LockAcquisitionFailed("registry write".to_string());
opening_slot.finish(Err(error.clone()));
return Err(error);
}
};
let owns_slot = matches!(
registry.get(®istry_key),
Some(DatabaseRegistryEntry::Opening(slot))
if Arc::ptr_eq(slot, &opening_slot)
);
if !owns_slot {
let error = Error::internal(format!(
"database `{dsn}` opening owner changed before publication"
));
opening_slot.finish(Err(error.clone()));
return Err(error);
}
registry.insert(
registry_key.clone(),
DatabaseRegistryEntry::Ready(Arc::clone(&inner)),
);
drop(registry);
opening_slot.finish(Ok(()));
Ok(Self::connection_from_owner(inner))
}
Err(error) => {
if let Ok(mut registry) = DATABASE_REGISTRY.write() {
let owns_slot = matches!(
registry.get(®istry_key),
Some(DatabaseRegistryEntry::Opening(slot))
if Arc::ptr_eq(slot, &opening_slot)
);
if owns_slot {
registry.remove(®istry_key);
}
}
opening_slot.finish(Err(error.clone()));
Err(error)
}
}
}
/// Open an in-memory database
///
/// This is a convenience method that creates a new in-memory database.
/// Each call creates a unique instance (unlike `open("memory://")` which
/// would share the same instance).
pub fn open_in_memory() -> Result<Self> {
Self::create_in_memory_engine()
}
#[cfg(feature = "test-filedb")]
fn create_in_memory_engine() -> Result<Self> {
let tmp = tempfile::tempdir()
.map_err(|e| Error::internal(format!("failed to create temp dir: {}", e)))?;
let file_dsn = format!("file://{}", tmp.path().display());
let (_clean_path, config) = Self::parse_file_config(&file_dsn[7..])?;
let engine = composed_mvcc_engine(config, Arc::clone(&DEFAULT_PLUGIN_REGISTRY));
engine.open_engine()?;
let engine = Arc::new(engine);
engine.start_cleanup();
let owner = Arc::new(DatabaseOwner {
engine,
semantic_cache: Arc::new(SemanticCache::new()),
feedback_cache: Arc::new(FeedbackCache::new()),
plugin_registry: Arc::clone(&DEFAULT_PLUGIN_REGISTRY),
registry_key: "memory://".to_string(),
dsn: "memory://".to_string(),
requested_config: Config::in_memory(),
_temp_dir: Some(tmp),
});
Ok(Self::connection_from_owner(owner))
}
#[cfg(not(feature = "test-filedb"))]
fn create_in_memory_engine() -> Result<Self> {
let engine = composed_in_memory_engine();
engine.open_engine()?;
let engine = Arc::new(engine);
engine.start_cleanup();
let owner = Arc::new(DatabaseOwner {
engine,
semantic_cache: Arc::new(SemanticCache::new()),
feedback_cache: Arc::new(FeedbackCache::new()),
plugin_registry: Arc::clone(&DEFAULT_PLUGIN_REGISTRY),
registry_key: "memory://".to_string(),
dsn: "memory://".to_string(),
requested_config: Config::in_memory(),
});
Ok(Self::connection_from_owner(owner))
}
/// Parse a DSN into scheme and path
fn parse_dsn(dsn: &str) -> Result<(String, String)> {
let idx = dsn
.find("://")
.ok_or_else(|| Error::parse("Invalid DSN format: expected scheme://path"))?;
let scheme = dsn[..idx].to_lowercase();
let path = dsn[idx + 3..].to_string();
// Validate scheme
match scheme.as_str() {
MEMORY_SCHEME | FILE_SCHEME => {}
_ => {
return Err(Error::parse(format!(
"Unsupported scheme '{}'. Use 'memory://' or 'file://path'",
scheme
)));
}
}
// Validate file path
if scheme == FILE_SCHEME {
let clean_path = if path.contains('?') {
&path[..path.find('?').unwrap()]
} else {
&path
};
if clean_path.is_empty() {
return Err(Error::parse("file:// scheme requires a non-empty path"));
}
}
Ok((scheme, path))
}
/// Parse file:// config from query parameters
fn parse_file_config(path: &str) -> Result<(String, Config)> {
fn parse_bool_option(key: &str, value: &str) -> Result<bool> {
match value.to_ascii_lowercase().as_str() {
"on" | "true" | "1" | "yes" => Ok(true),
"off" | "false" | "0" | "no" => Ok(false),
_ => Err(Error::invalid_argument(format!(
"invalid {key}: '{value}' (expected on/off)"
))),
}
}
let (clean_path, query) = if let Some(idx) = path.find('?') {
(path[..idx].to_string(), Some(&path[idx + 1..]))
} else {
(path.to_string(), None)
};
let mut config = Config::with_path(&clean_path);
// Parse query parameters
if let Some(query) = query {
let mut seen_options = HashSet::new();
for param in query.split('&') {
let mut parts = param.splitn(2, '=');
let key = parts.next().unwrap_or("");
let value = parts.next().unwrap_or("");
if !seen_options.insert(key) {
return Err(Error::invalid_argument(format!(
"duplicate file database option: '{key}'"
)));
}
match key {
// Sync mode: sync_mode=none|normal|full
"sync_mode" => {
config.persistence.sync_mode = match value.to_lowercase().as_str() {
"none" | "off" | "0" => SyncMode::None,
"normal" | "1" => SyncMode::Normal,
"full" | "2" => SyncMode::Full,
_ => {
return Err(Error::invalid_argument(format!(
"invalid sync mode: '{value}' (expected none/normal/full)"
)))
}
};
}
// Checkpoint interval in seconds: checkpoint_interval=60
"checkpoint_interval" => {
config.persistence.checkpoint_interval =
value.parse::<u32>().map_err(|_| {
Error::invalid_argument(format!(
"invalid checkpoint_interval: '{}'",
value
))
})?;
}
// Compaction threshold: compact_threshold=4
"compact_threshold" => {
config.persistence.compact_threshold =
value.parse::<u32>().map_err(|_| {
Error::invalid_argument(format!(
"invalid compact_threshold: '{}'",
value
))
})?;
}
"max_compaction_jobs" => {
let count = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid max_compaction_jobs: '{}'",
value
))
})?;
if !(1..=radixdb_storage::config::MAX_COMPACTION_JOBS).contains(&count) {
return Err(Error::invalid_argument(format!(
"max_compaction_jobs must be in 1..={}",
radixdb_storage::config::MAX_COMPACTION_JOBS
)));
}
config.persistence.max_compaction_jobs = count;
}
// Shared CPU-heavy storage worker budget. Zero selects
// host/cgroup-visible automatic parallelism.
"storage_cpu_workers" => {
config.persistence.storage_cpu_workers =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid storage_cpu_workers: '{}'",
value
))
})?;
}
"page_cache_level" => {
let level = value.parse::<u8>().map_err(|_| {
Error::invalid_argument(format!(
"invalid page_cache_level: '{}'",
value
))
})?;
if level > radixdb_storage::config::MAX_PAGE_CACHE_LEVEL {
return Err(Error::invalid_argument(format!(
"page_cache_level must be in 0..={}",
radixdb_storage::config::MAX_PAGE_CACHE_LEVEL
)));
}
config.persistence.page_cache_level = level;
}
"page_cache_max_bytes" => {
config.persistence.page_cache_max_bytes =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid page_cache_max_bytes: '{}'",
value
))
})?;
}
"page_cache_memory_reserve" => {
config.persistence.page_cache_memory_reserve =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid page_cache_memory_reserve: '{}'",
value
))
})?;
}
// Maximum immutable inputs owned by one compaction job.
"max_compaction_input_segments" => {
let count = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid max_compaction_input_segments: '{}'",
value
))
})?;
config.persistence.max_compaction_input_segments = count.max(1);
}
// Maximum physical payload + posting bytes owned by one
// compaction job.
"max_compaction_input_bytes" => {
let bytes = value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid max_compaction_input_bytes: '{}'",
value
))
})?;
config.persistence.max_compaction_input_bytes = bytes.max(1);
}
"max_compaction_output_bytes" => {
let bytes = value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid max_compaction_output_bytes: '{}'",
value
))
})?;
config.persistence.max_compaction_output_bytes = bytes.max(1);
}
"compaction_job_time_budget_ms" => {
config.persistence.compaction_job_time_budget_ms =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid compaction_job_time_budget_ms: '{}'",
value
))
})?;
}
"compaction_io_bytes_per_sec" => {
config.persistence.compaction_io_bytes_per_sec =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid compaction_io_bytes_per_sec: '{}'",
value
))
})?;
}
"compaction_disk_reserve_bytes" => {
config.persistence.compaction_disk_reserve_bytes =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid compaction_disk_reserve_bytes: '{}'",
value
))
})?;
}
"compaction_retry_cooldown_ms" => {
config.persistence.compaction_retry_cooldown_ms =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid compaction_retry_cooldown_ms: '{}'",
value
))
})?;
}
"l0_soft_limit_segments" => {
config.persistence.l0_soft_limit_segments =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid l0_soft_limit_segments: '{}'",
value
))
})?;
}
"l0_hard_limit_segments" => {
config.persistence.l0_hard_limit_segments =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid l0_hard_limit_segments: '{}'",
value
))
})?;
}
"l0_soft_limit_bytes" => {
config.persistence.l0_soft_limit_bytes =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid l0_soft_limit_bytes: '{}'",
value
))
})?;
}
"l0_hard_limit_bytes" => {
config.persistence.l0_hard_limit_bytes =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid l0_hard_limit_bytes: '{}'",
value
))
})?;
}
"l0_soft_backpressure_wait_ms" => {
config.persistence.l0_soft_backpressure_wait_ms =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid l0_soft_backpressure_wait_ms: '{}'",
value
))
})?;
}
// Number of backup snapshots to keep: keep_snapshots=3
"keep_snapshots" => {
config.persistence.keep_snapshots = value.parse::<u32>().map_err(|_| {
Error::invalid_argument(format!("invalid keep_snapshots: '{}'", value))
})?;
}
// WAL flush trigger in bytes: wal_flush_trigger=32768
"wal_flush_trigger" => {
config.persistence.wal_flush_trigger =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid wal_flush_trigger: '{}'",
value
))
})?;
}
// WAL buffer size in bytes: wal_buffer_size=65536
"wal_buffer_size" => {
config.persistence.wal_buffer_size =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid wal_buffer_size: '{}'",
value
))
})?;
}
// WAL max size in bytes: wal_max_size=67108864
"wal_max_size" => {
config.persistence.wal_max_size = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!("invalid wal_max_size: '{}'", value))
})?;
}
// Fail-closed memory envelope for one atomic COPY.
"copy_max_transaction_bytes" => {
let bytes = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid copy_max_transaction_bytes: '{}'",
value
))
})?;
if bytes == 0 {
return Err(Error::invalid_argument(
"copy_max_transaction_bytes must not be zero",
));
}
config.persistence.copy_max_transaction_bytes = bytes;
}
// Removed before v1: Normal mode durably syncs every
// terminal commit, so accepting this knob would promise a
// batching policy that does not exist.
"commit_batch_size" => {
return Err(Error::invalid_argument(format!(
"commit_batch_size is unsupported; choose sync_mode instead (got '{}')",
value
)));
}
// Sync interval in ms: sync_interval_ms=10
"sync_interval_ms" => {
config.persistence.sync_interval_ms =
value.parse::<u32>().map_err(|_| {
Error::invalid_argument(format!(
"invalid sync_interval_ms: '{}'",
value
))
})?;
}
// WAL compression: wal_compression=on|off
"wal_compression" => {
config.persistence.wal_compression =
parse_bool_option("wal_compression", value)?;
}
// Volume LZ4 compression: volume_compression=on|off
"volume_compression" => {
config.persistence.volume_compression =
parse_bool_option("volume_compression", value)?;
}
// All compressions (WAL + volume): compression=on|off
"compression" => {
let enabled = parse_bool_option("compression", value)?;
config.persistence.wal_compression = enabled;
config.persistence.volume_compression = enabled;
}
// Target rows per volume: target_volume_rows=1048576
"target_volume_rows" => {
let rows = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid target_volume_rows: '{}'",
value
))
})?;
config.persistence.target_volume_rows = rows.max(65_536);
}
// First seal hot byte threshold: seal_hot_bytes_threshold=67108864
"seal_hot_bytes_threshold" => {
let bytes = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid seal_hot_bytes_threshold: '{}'",
value
))
})?;
config.persistence.seal_hot_bytes_threshold = bytes.max(1);
}
// Incremental seal hot byte threshold:
// seal_incremental_hot_bytes_threshold=16777216
"seal_incremental_hot_bytes_threshold" => {
let bytes = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid seal_incremental_hot_bytes_threshold: '{}'",
value
))
})?;
config.persistence.seal_incremental_hot_bytes_threshold = bytes.max(1);
}
// Global resident cold-volume payload cache budget:
// volume_cache_bytes=1073741824
"volume_cache_bytes" => {
config.persistence.volume_cache_bytes =
value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid volume_cache_bytes: '{}'",
value
))
})?;
}
// Cold-volume read queue depth:
// read_queue_depth=4
"read_queue_depth" => {
let depth = value.parse::<usize>().map_err(|_| {
Error::invalid_argument(format!(
"invalid read_queue_depth: '{}'",
value
))
})?;
config.persistence.read_queue_depth = depth.max(1);
}
// Checkpoint on close: checkpoint_on_close=off
// Set to off to simulate crashes in tests (WAL not truncated)
"checkpoint_on_close" => {
config.persistence.checkpoint_on_close =
parse_bool_option("checkpoint_on_close", value)?;
}
// Cleanup interval in seconds: cleanup_interval=60
"cleanup_interval" => {
config.cleanup.interval_secs = value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid cleanup_interval: '{}'",
value
))
})?;
}
// Deleted row retention in seconds: deleted_row_retention=300
"deleted_row_retention" => {
config.cleanup.deleted_row_retention_secs =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid deleted_row_retention: '{}'",
value
))
})?;
}
// Transaction retention in seconds: transaction_retention=3600
"transaction_retention" => {
config.cleanup.transaction_retention_secs =
value.parse::<u64>().map_err(|_| {
Error::invalid_argument(format!(
"invalid transaction_retention: '{}'",
value
))
})?;
}
// Disable cleanup: cleanup=off
"cleanup" => {
config.cleanup.enabled = parse_bool_option("cleanup", value)?;
}
_ => {
return Err(Error::invalid_argument(format!(
"unknown file database option: '{key}'"
)))
}
}
}
}
if config.persistence.l0_soft_limit_segments == 0
|| config.persistence.l0_soft_limit_segments
>= config.persistence.l0_hard_limit_segments
{
return Err(Error::invalid_argument(
"l0 segment limits require 0 < soft < hard",
));
}
if config.persistence.l0_soft_limit_bytes == 0
|| config.persistence.l0_soft_limit_bytes >= config.persistence.l0_hard_limit_bytes
{
return Err(Error::invalid_argument(
"l0 byte limits require 0 < soft < hard",
));
}
Ok((clean_path, config))
}
/// Execute a SQL statement
///
/// Use this for DDL (CREATE, DROP, ALTER) and DML (INSERT, UPDATE, DELETE) statements.
///
/// # Parameters
///
/// Parameters can be passed using:
/// - Empty tuple `()` for no parameters
/// - Tuple syntax `(1, "Alice", 30)` for multiple parameters
/// - `params!` macro `params![1, "Alice", 30]`
///
/// # Returns
///
/// Returns the number of rows affected for DML statements, or 0 for DDL.
///
/// # Examples
///
/// ```ignore
/// // DDL - no parameters
/// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
///
/// // DML with tuple parameters
/// db.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
///
/// // DML with params! macro
/// db.execute("INSERT INTO users VALUES ($1, $2)", params![2, "Bob"])?;
///
/// // Update with mixed types
/// let affected = db.execute(
/// "UPDATE users SET name = $1 WHERE id = $2",
/// ("Charlie", 1)
/// )?;
/// ```
pub fn execute<P: Params>(&self, sql: &str, params: P) -> Result<i64> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = executor.execute_with_params(sql, params.into_params())?;
Ok(result.rows_affected())
}
/// Execute a query that returns rows
///
/// # Parameters
///
/// Parameters can be passed using:
/// - Empty tuple `()` for no parameters
/// - Tuple syntax `(value,)` for single parameter (note trailing comma)
/// - Tuple syntax `(1, "Alice")` for multiple parameters
/// - `params!` macro `params![1, "Alice"]`
///
/// # Examples
///
/// ```ignore
/// // Query all rows
/// for row in db.query("SELECT * FROM users", ())? {
/// let row = row?;
/// let id: i64 = row.get(0)?;
/// let name: String = row.get(1)?;
/// }
///
/// // Query with parameters
/// for row in db.query("SELECT * FROM users WHERE age > $1", (18,))? {
/// // ...
/// }
///
/// // Collect into Vec
/// let users: Vec<_> = db.query("SELECT * FROM users", ())?
/// .collect::<Result<Vec<_>, _>>()?;
/// ```
pub fn query<P: Params>(&self, sql: &str, params: P) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = executor.execute_with_params(sql, params.into_params())?;
Ok(Rows::new(result))
}
/// Execute a query and return a single value
///
/// This is a convenience method for queries that return a single row with a single column.
/// Returns an error if the query returns no rows.
///
/// # Examples
///
/// ```ignore
/// let count: i64 = db.query_one("SELECT COUNT(*) FROM users", ())?;
/// let name: String = db.query_one("SELECT name FROM users WHERE id = $1", (1,))?;
/// ```
pub fn query_one<T: FromValue, P: Params>(&self, sql: &str, params: P) -> Result<T> {
let row = self
.query(sql, params)?
.next()
.ok_or(Error::NoRowsReturned)??;
row.get(0)
}
/// Execute a query and return an optional single value
///
/// Like `query_one`, but returns `None` if no rows are returned.
///
/// # Examples
///
/// ```ignore
/// let name: Option<String> = db.query_opt("SELECT name FROM users WHERE id = $1", (999,))?;
/// assert!(name.is_none());
/// ```
pub fn query_opt<T: FromValue, P: Params>(&self, sql: &str, params: P) -> Result<Option<T>> {
match self.query(sql, params)?.next() {
Some(row) => Ok(Some(row?.get(0)?)),
None => Ok(None),
}
}
/// Execute a write statement with a timeout
///
/// Like `execute`, but cancels the query if it exceeds the timeout.
/// Timeout is specified in milliseconds. Use 0 for no timeout.
///
/// # Examples
///
/// ```ignore
/// // Execute with 5 second timeout
/// db.execute_with_timeout("DELETE FROM large_table WHERE old = true", (), 5000)?;
/// ```
pub fn execute_with_timeout<P: Params>(
&self,
sql: &str,
params: P,
timeout_ms: u64,
) -> Result<i64> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let param_values = params.into_params();
let ctx = ExecutionContextBuilder::new()
.params(param_values)
.timeout_ms(timeout_ms)
.build();
let result = executor.execute_with_context(sql, &ctx)?;
Ok(result.rows_affected())
}
/// Execute a query with a timeout
///
/// Like `query`, but cancels the query if it exceeds the timeout.
/// Timeout is specified in milliseconds. Use 0 for no timeout.
///
/// # Examples
///
/// ```ignore
/// // Query with 10 second timeout
/// for row in db.query_with_timeout("SELECT * FROM large_table", (), 10000)? {
/// // process row
/// }
/// ```
pub fn query_with_timeout<P: Params>(
&self,
sql: &str,
params: P,
timeout_ms: u64,
) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let param_values = params.into_params();
let ctx = ExecutionContextBuilder::new()
.params(param_values)
.timeout_ms(timeout_ms)
.build();
let result = executor.execute_with_context(sql, &ctx)?;
Ok(Rows::new(result))
}
/// Prepare a SQL statement for repeated execution
///
/// Prepared statements are more efficient when executing the same query
/// multiple times with different parameters.
///
/// # Examples
///
/// ```ignore
/// let stmt = db.prepare("SELECT * FROM users WHERE id = $1")?;
///
/// // Execute multiple times with different parameters
/// for id in 1..=10 {
/// for row in stmt.query((id,))? {
/// // ...
/// }
/// }
/// ```
pub fn prepare(&self, sql: &str) -> Result<Statement> {
self.ensure_open()?;
Statement::new(Arc::downgrade(&self.inner), sql.to_string(), self)
}
/// Create a Database from an existing Arc<DatabaseInner>.
/// Used by Statement to upgrade weak references.
pub(crate) fn from_inner(inner: Arc<DatabaseInner>) -> Self {
Database { inner }
}
/// Execute a statement with named parameters
///
/// Named parameters use the `:name` syntax in SQL queries.
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, named_params};
///
/// let db = Database::open("memory://")?;
/// db.execute("CREATE TABLE users (id INTEGER, name TEXT, age INTEGER)", ())?;
///
/// // Insert with named params
/// db.execute_named(
/// "INSERT INTO users VALUES (:id, :name, :age)",
/// named_params!{ id: 1, name: "Alice", age: 30 }
/// )?;
///
/// // Update with named params
/// db.execute_named(
/// "UPDATE users SET name = :name WHERE id = :id",
/// named_params!{ id: 1, name: "Alicia" }
/// )?;
/// ```
pub fn execute_named(&self, sql: &str, params: NamedParams) -> Result<i64> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = executor.execute_with_named_params(sql, params.into_inner())?;
Ok(result.rows_affected())
}
/// Execute a query with named parameters
///
/// Named parameters use the `:name` syntax in SQL queries.
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, named_params};
///
/// let db = Database::open("memory://")?;
/// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
/// db.execute("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')", ())?;
///
/// // Query with named params
/// for row in db.query_named(
/// "SELECT * FROM users WHERE name = :name",
/// named_params!{ name: "Alice" }
/// )? {
/// let row = row?;
/// println!("Found user: id={}", row.get::<i64>(0)?);
/// }
/// ```
pub fn query_named(&self, sql: &str, params: NamedParams) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = executor.execute_with_named_params(sql, params.into_inner())?;
Ok(Rows::new(result))
}
/// Execute a query with named parameters and a cancellation deadline.
///
/// This is the named-parameter counterpart of [`Database::query_with_timeout`].
/// A timeout of zero keeps the existing unlimited behavior.
pub fn query_named_with_timeout(
&self,
sql: &str,
params: NamedParams,
timeout_ms: u64,
) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let mut ctx = ExecutionContext::with_named_params(params.into_inner());
ctx.set_timeout_ms(timeout_ms);
let result = executor.execute_with_context(sql, &ctx)?;
Ok(Rows::new(result))
}
/// Execute one server request through the embedded facade contract.
#[doc(hidden)]
pub fn query_for_server(
&self,
sql: &str,
context: &super::ServerExecutionContext,
) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = executor.execute_with_context(sql, context.inner())?;
Ok(Rows::new(result))
}
/// Execute a query with named parameters and return a single value
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, named_params};
///
/// let count: i64 = db.query_one_named(
/// "SELECT COUNT(*) FROM users WHERE age > :min_age",
/// named_params!{ min_age: 18 }
/// )?;
/// ```
pub fn query_one_named<T: FromValue>(&self, sql: &str, params: NamedParams) -> Result<T> {
let mut rows = self.query_named(sql, params)?;
match rows.next() {
Some(Ok(row)) => row.get(0),
Some(Err(e)) => Err(e),
None => Err(Error::NoRowsReturned),
}
}
/// Execute a query and map results to structs
///
/// This method executes a query and converts each row to a struct
/// that implements the `FromRow` trait.
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, FromRow, ResultRow, Result};
///
/// struct User {
/// id: i64,
/// name: String,
/// }
///
/// impl FromRow for User {
/// fn from_row(row: &ResultRow) -> Result<Self> {
/// Ok(User {
/// id: row.get(0)?,
/// name: row.get(1)?,
/// })
/// }
/// }
///
/// let db = Database::open("memory://")?;
/// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
/// db.execute("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')", ())?;
///
/// // Query and map to structs
/// let users: Vec<User> = db.query_as("SELECT id, name FROM users", ())?;
/// assert_eq!(users.len(), 2);
/// assert_eq!(users[0].name, "Alice");
/// ```
pub fn query_as<T: FromRow, P: Params>(&self, sql: &str, params: P) -> Result<Vec<T>> {
let rows = self.query(sql, params)?;
rows.map(|r| r.and_then(|row| T::from_row(&row))).collect()
}
/// Execute a query with named parameters and map results to structs
///
/// # Examples
///
/// ```ignore
/// use radixdb::{Database, FromRow, ResultRow, Result, named_params};
///
/// struct Product {
/// id: i64,
/// name: String,
/// price: f64,
/// }
///
/// impl FromRow for Product {
/// fn from_row(row: &ResultRow) -> Result<Self> {
/// Ok(Product {
/// id: row.get(0)?,
/// name: row.get(1)?,
/// price: row.get(2)?,
/// })
/// }
/// }
///
/// let products: Vec<Product> = db.query_as_named(
/// "SELECT id, name, price FROM products WHERE price > :min_price",
/// named_params!{ min_price: 10.0 }
/// )?;
/// ```
pub fn query_as_named<T: FromRow>(&self, sql: &str, params: NamedParams) -> Result<Vec<T>> {
let rows = self.query_named(sql, params)?;
rows.map(|r| r.and_then(|row| T::from_row(&row))).collect()
}
/// Begin a new transaction with default isolation level
///
/// # Examples
///
/// ```ignore
/// let tx = db.begin()?;
/// tx.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
/// tx.commit()?;
/// ```
pub fn begin(&self) -> Result<Transaction> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let tx = executor.begin_transaction()?;
Ok(Transaction::new(tx, Arc::clone(&self.inner)))
}
/// Begin the read-only transaction used by the logical SQL exporter.
///
/// Lock ordering intentionally matches normal connection execution:
/// connection executor first, then the shared DDL fence. The returned
/// transaction retains that fence and streams rows from one MVCC snapshot.
#[doc(hidden)]
pub fn begin_logical_export(&self) -> Result<Transaction> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let engine = executor.engine().clone();
let plugin_registry = executor.plugin_registry();
let fence = engine.acquire_ddl_statement_fence(false);
let tx = executor.begin_transaction_with_isolation(IsolationLevel::SnapshotIsolation)?;
Ok(Transaction::new_logical_export(
tx,
engine,
plugin_registry,
Arc::clone(&self.inner),
fence,
))
}
/// Begin a new transaction with a specific isolation level
///
/// # Examples
///
/// ```ignore
/// use radixdb::IsolationLevel;
///
/// let tx = db.begin_with_isolation(IsolationLevel::SnapshotIsolation)?;
/// // All reads in this transaction see a consistent snapshot
/// tx.execute("UPDATE users SET balance = balance - 100 WHERE id = $1", (1,))?;
/// tx.commit()?;
/// ```
pub fn begin_with_isolation(&self, isolation: IsolationLevel) -> Result<Transaction> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let tx = executor.begin_transaction_with_isolation(isolation)?;
Ok(Transaction::new(tx, Arc::clone(&self.inner)))
}
/// Get the underlying storage engine
///
/// This is primarily for advanced use cases and testing.
pub fn engine(&self) -> &Arc<MVCCEngine> {
&self.inner.engine
}
/// Close the database connection
///
/// This removes the database from the global registry and closes the engine,
/// releasing the file lock immediately so another process can open the database.
///
/// Note: The engine is also closed automatically when all Database instances
/// are dropped.
pub fn close(&self) -> Result<()> {
if Arc::strong_count(&self.inner) != 1 {
return Err(Error::invalid_argument(
"cannot close database while a Transaction or retained connection owner is active",
));
}
// Explicit close is terminal for the shared engine, but registry
// ownership remains intact until every mandatory close stage succeeds.
self.inner.engine.close_engine()?;
let owner = self.owner_arc();
let mut registry = DATABASE_REGISTRY
.write()
.map_err(|_| Error::LockAcquisitionFailed("registry write".to_string()))?;
if matches!(
registry.get(&owner.registry_key),
Some(DatabaseRegistryEntry::Ready(entry)) if Arc::ptr_eq(entry, owner)
) {
registry.remove(&owner.registry_key);
}
Ok(())
}
/// Get a cached plan for a SQL statement (parse once, execute many times).
///
/// Returns a `CachedPlanRef` that can be stored and passed to
/// `execute_plan()` / `query_plan()` for zero-lookup execution.
pub fn cached_plan(&self, sql: &str) -> Result<CachedPlanRef> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.get_or_create_plan(sql)
}
/// Execute a pre-cached plan with positional parameters (no parsing, no cache lookup).
pub fn execute_plan<P: Params>(&self, plan: &CachedPlanRef, params: P) -> Result<i64> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let param_values = params.into_params();
let ctx = if param_values.is_empty() {
ExecutionContext::new()
} else {
ExecutionContext::with_params(param_values)
};
let result = executor.execute_with_cached_plan(plan, &ctx)?;
Ok(result.rows_affected())
}
/// Query using a pre-cached plan with positional parameters (no parsing, no cache lookup).
pub fn query_plan<P: Params>(&self, plan: &CachedPlanRef, params: P) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let param_values = params.into_params();
let ctx = if param_values.is_empty() {
ExecutionContext::new()
} else {
ExecutionContext::with_params(param_values)
};
let result = executor.execute_with_cached_plan(plan, &ctx)?;
Ok(Rows::new(result))
}
/// Execute a pre-cached plan with named parameters (no parsing, no cache lookup).
pub fn execute_named_plan(&self, plan: &CachedPlanRef, params: NamedParams) -> Result<i64> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let ctx = ExecutionContext::with_named_params(params.into_inner());
let result = executor.execute_with_cached_plan(plan, &ctx)?;
Ok(result.rows_affected())
}
/// Query using a pre-cached plan with named parameters (no parsing, no cache lookup).
pub fn query_named_plan(&self, plan: &CachedPlanRef, params: NamedParams) -> Result<Rows> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let ctx = ExecutionContext::with_named_params(params.into_inner());
let result = executor.execute_with_cached_plan(plan, &ctx)?;
Ok(Rows::new(result))
}
/// Check if a table exists
pub fn table_exists(&self, name: &str) -> Result<bool> {
self.ensure_open()?;
let engine = &self.inner.engine;
let tx = engine.begin_transaction()?;
match tx.get_table(name) {
Ok(_) => Ok(true),
Err(Error::TableNotFound(_)) => Ok(false),
Err(error) => Err(error),
}
}
/// Get the DSN this database was opened with
pub fn dsn(&self) -> &str {
&self.inner.owner.dsn
}
/// Set the default isolation level for new transactions
pub fn set_default_isolation_level(&self, level: IsolationLevel) -> Result<()> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.set_default_isolation_level(level);
Ok(())
}
/// Return this connection's default isolation for future transactions.
pub fn default_isolation_level(&self) -> Result<IsolationLevel> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
Ok(executor.default_isolation_level())
}
/// Create a backup snapshot of the database
///
/// This pins one complete catalog/data/index generation and copies every
/// reachable immutable member into a committed physical snapshot.
/// Normal persistence advances the same generation graph plus WAL.
///
/// In-memory databases reject this operation because no durable snapshot
/// artifact can be created.
pub fn create_snapshot(&self) -> Result<radixdb_storage::PhysicalSnapshotIdentity> {
use radixdb_storage::Engine;
self.ensure_open()?;
self.inner.engine.create_snapshot()
}
/// Restore the database from a backup snapshot.
///
/// If no identity is provided, restores from the latest snapshot.
/// Otherwise restores the snapshot with the specified stable identity.
///
/// This is a destructive operation that atomically replaces the current
/// catalog/data/index generation with the selected complete snapshot.
pub fn restore_snapshot(&self, snapshot_id: Option<&str>) -> Result<String> {
use radixdb_storage::Engine;
self.ensure_open()?;
// Cache admission is fallible, so acquire it before the destructive
// engine operation. A poisoned lock must never turn a committed
// restore into an error outcome.
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
let result = self.inner.engine.restore_snapshot(snapshot_id)?;
// Clear all query caches since all data has changed.
executor.clear_semantic_cache();
radixdb_executor::context::clear_scalar_subquery_cache();
radixdb_executor::context::clear_in_subquery_cache();
radixdb_executor::context::clear_semi_join_cache();
Ok(result)
}
/// Get the internal executor (for Statement use)
pub(crate) fn executor(&self) -> &Mutex<Executor> {
&self.inner.executor
}
#[doc(hidden)]
pub fn describe_query_output(
&self,
sql: &str,
) -> Result<Option<Vec<radixdb_executor::QueryOutputColumn>>> {
self.ensure_open()?;
self.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?
.describe_query_output(sql)
}
/// Whether this connection-local SQL executor owns a BEGIN transaction.
#[doc(hidden)]
pub fn has_active_sql_transaction(&self) -> Result<bool> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
Ok(executor.has_active_transaction())
}
/// Get semantic-cache statistics.
pub fn semantic_cache_stats(&self) -> Result<SemanticCacheStatsSnapshot> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
Ok(executor.semantic_cache_stats())
}
/// Clear all semantic-cache entries.
pub fn clear_semantic_cache(&self) -> Result<()> {
self.ensure_open()?;
let executor = self
.inner
.executor
.lock()
.map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
executor.clear_semantic_cache();
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::named_params;
use radixdb_core::Value;
#[test]
fn checkpoint_reopens_from_canonical_artifact_generation() {
let directory = tempfile::tempdir().expect("create database directory");
let database_path = directory.path().join("canonical-checkpoint");
let dsn = format!(
"file://{}?checkpoint_on_close=false",
database_path.display()
);
let database = Database::open(&dsn).expect("open persistent database");
database
.execute(
"CREATE TABLE items (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)",
(),
)
.expect("create table");
let catalog = database.engine().pin_catalog().expect("pin catalog");
assert!(
catalog
.graph()
.objects()
.any(|object| object.name().display().as_str() == "items"),
"CREATE TABLE must publish the logical catalog before returning"
);
database
.execute("INSERT INTO items VALUES (1, 'one'), (2, 'two')", ())
.expect("insert rows");
radixdb_storage::traits::Engine::force_checkpoint_cycle(database.engine().as_ref())
.expect("publish physical checkpoint");
database.close().expect("close database");
let mut pending = vec![database_path.clone()];
while let Some(directory) = pending.pop() {
for entry in std::fs::read_dir(&directory).expect("read artifact tree") {
let entry = entry.expect("read artifact member");
let path = entry.path();
if entry.file_type().expect("read artifact type").is_dir() {
pending.push(path);
continue;
}
let name = path
.file_name()
.and_then(|name| name.to_str())
.expect("artifact name is UTF-8");
assert!(!name.starts_with("ddl-"), "legacy DDL owner: {name}");
assert!(!name.ends_with(".vol"), "legacy volume: {name}");
assert!(!name.ends_with(".rpi"), "legacy postings: {name}");
}
}
assert!(!database_path.join("volumes").exists());
let reopened = Database::open(&dsn).expect("reopen persistent database");
let rows = reopened
.query("SELECT id, payload FROM items ORDER BY id", ())
.expect("query recovered rows")
.collect_vec()
.expect("collect recovered rows");
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].get::<i64>(0).expect("first id"), 1);
assert_eq!(rows[0].get::<String>(1).expect("first payload"), "one");
assert_eq!(rows[1].get::<i64>(0).expect("second id"), 2);
assert_eq!(rows[1].get::<String>(1).expect("second payload"), "two");
reopened.close().expect("close reopened database");
}
#[test]
fn transactional_index_rename_and_drop_survive_reopen() {
let directory = tempfile::tempdir().expect("create database directory");
let database_path = directory.path().join("transactional-index-ddl");
let dsn = format!(
"file://{}?checkpoint_on_close=false",
database_path.display()
);
let database = Database::open(&dsn).expect("open persistent database");
database
.execute(
"CREATE TABLE items (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)",
(),
)
.expect("create table");
database
.execute("CREATE INDEX idx_payload ON items(payload)", ())
.expect("create index");
database
.execute("ALTER INDEX idx_payload RENAME TO idx_payload_live", ())
.expect("rename index");
database.close().expect("close renamed-index database");
let reopened = Database::open(&dsn).expect("reopen renamed-index database");
let indexes = reopened
.schema()
.table("items")
.indexes()
.fetch()
.expect("fetch indexes after rename reopen");
assert!(indexes.iter().any(|index| index.name == "idx_payload_live"));
assert!(!indexes.iter().any(|index| index.name == "idx_payload"));
reopened
.execute("DROP INDEX idx_payload_live ON items", ())
.expect("drop renamed index");
reopened.close().expect("close dropped-index database");
let reopened = Database::open(&dsn).expect("reopen dropped-index database");
let indexes = reopened
.schema()
.table("items")
.indexes()
.fetch()
.expect("fetch indexes after drop reopen");
assert!(!indexes.iter().any(|index| index.name == "idx_payload_live"));
reopened.close().expect("close final database");
}
#[test]
fn ctas_and_view_catalog_survive_reopen_without_legacy_ddl_owner() {
let directory = tempfile::tempdir().expect("create database directory");
let database_path = directory.path().join("ctas-view-catalog");
let dsn = format!(
"file://{}?checkpoint_on_close=false&checkpoint_interval=0",
database_path.display()
);
let database = Database::open(&dsn).expect("open persistent database");
database
.execute(
"CREATE TABLE source_rows (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)",
(),
)
.expect("create source table");
database
.execute("INSERT INTO source_rows VALUES (1, 'one'), (2, 'two')", ())
.expect("insert source rows");
database
.execute(
"CREATE TABLE copied_rows AS SELECT id, payload FROM source_rows",
(),
)
.expect("create table as select");
database
.execute(
"CREATE VIEW copied_view AS SELECT id, payload FROM copied_rows",
(),
)
.expect("create view");
assert_eq!(
database
.query_one::<i64, _>("SELECT COUNT(*) FROM copied_view", ())
.expect("query live view"),
2
);
database.close().expect("close catalog database");
let reopened = Database::open(&dsn).expect("reopen catalog database");
assert_eq!(
reopened
.query_one::<i64, _>("SELECT COUNT(*) FROM copied_rows", ())
.expect("query reopened CTAS table"),
2
);
assert_eq!(
reopened
.query_one::<i64, _>("SELECT COUNT(*) FROM copied_view", ())
.expect("query reopened view"),
2
);
reopened
.execute("DROP VIEW copied_view", ())
.expect("drop view");
assert!(reopened.query("SELECT * FROM copied_view", ()).is_err());
reopened.close().expect("close dropped-view database");
let reopened = Database::open(&dsn).expect("reopen dropped-view database");
assert!(reopened.query("SELECT * FROM copied_view", ()).is_err());
reopened.close().expect("close final database");
}
#[test]
fn checkpoint_uses_the_actual_wal_generation_after_size_rotations() {
fn selected_wal_floor(root: &std::path::Path) -> u64 {
[
("CONTROL.0", radixdb_storage::v6::ControlSlotIndex::Zero),
("CONTROL.1", radixdb_storage::v6::ControlSlotIndex::One),
]
.into_iter()
.filter_map(|(name, slot)| {
let bytes = std::fs::read(root.join(name)).ok()?;
radixdb_storage::v6::decode_control_slot(&bytes, slot).ok()
})
.max_by_key(|control| control.database_generation())
.expect("at least one valid CONTROL")
.wal_replay_floor()
.generation()
.get()
}
fn active_wal_generations(root: &std::path::Path) -> Vec<u64> {
let mut generations = std::fs::read_dir(root.join("wal"))
.expect("read WAL directory")
.filter_map(std::result::Result::ok)
.filter_map(|entry| {
let name = entry.file_name();
let name = name.to_str()?;
let encoded = name.strip_prefix("wal-")?.strip_suffix(".log")?;
(encoded.len() == 16)
.then(|| u64::from_str_radix(encoded, 16).ok())
.flatten()
})
.collect::<Vec<_>>();
generations.sort_unstable();
generations
}
let directory = tempfile::tempdir().expect("create database directory");
let database_path = directory.path().join("rotated-wal-checkpoint");
let dsn = format!(
"file://{}?checkpoint_on_close=false&checkpoint_interval=0&wal_max_size=512",
database_path.display()
);
let database = Database::open(&dsn).expect("open persistent database");
database
.execute(
"CREATE TABLE items (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)",
(),
)
.expect("create table");
for id in 1..=12_i64 {
database
.execute(
"INSERT INTO items VALUES (?, ?)",
(id, format!("{id:04}-{}", "x".repeat(512))),
)
.expect("insert rotation row");
}
let wal_files_before_checkpoint = std::fs::read_dir(database_path.join("wal"))
.expect("read WAL directory")
.filter_map(std::result::Result::ok)
.filter(|entry| {
entry
.file_name()
.to_str()
.is_some_and(|name| name.starts_with("wal-") && name.ends_with(".log"))
})
.count();
assert!(
wal_files_before_checkpoint >= 3,
"test did not create multiple WAL rotations"
);
radixdb_storage::traits::Engine::force_checkpoint_cycle(database.engine().as_ref())
.expect("publish checkpoint at actual WAL generation");
let first_checkpoint_floor = selected_wal_floor(&database_path);
assert!(
first_checkpoint_floor > 1,
"checkpoint did not retain the actual rotated generation"
);
for id in 13..=24_i64 {
database
.execute(
"INSERT INTO items VALUES (?, ?)",
(id, format!("{id:04}-{}", "y".repeat(512))),
)
.expect("insert second rotation row");
}
radixdb_storage::traits::Engine::force_checkpoint_cycle(database.engine().as_ref())
.expect("publish second checkpoint and retire old rotation prefix");
let active_generations = active_wal_generations(&database_path);
assert!(
active_generations
.iter()
.all(|generation| *generation >= first_checkpoint_floor),
"WAL generations below retained floor {first_checkpoint_floor} leaked: {active_generations:?}"
);
database.close().expect("close database");
let reopened = Database::open(&dsn).expect("reopen rotated WAL database");
let rows = reopened
.query("SELECT COUNT(*) FROM items", ())
.expect("count recovered rows")
.collect_vec()
.expect("collect count row");
let count = rows
.first()
.expect("count row")
.get::<i64>(0)
.expect("count value");
assert_eq!(count, 24);
reopened.close().expect("close reopened database");
}
#[test]
fn compacted_tier_and_row_precedence_survive_reopen() {
fn data_artifact_count(root: &std::path::Path) -> usize {
let mut count = 0;
let mut pending = vec![root.to_path_buf()];
while let Some(directory) = pending.pop() {
for entry in std::fs::read_dir(directory).expect("read artifact directory") {
let entry = entry.expect("read artifact member");
let path = entry.path();
if entry.file_type().expect("read artifact type").is_dir() {
pending.push(path);
} else if path.extension().and_then(|value| value.to_str()) == Some("data") {
count += 1;
}
}
}
count
}
let directory = tempfile::tempdir().expect("create database directory");
let database_path = directory.path().join("canonical-compaction");
let dsn = format!(
"file://{}?checkpoint_on_close=false&checkpoint_interval=0&compact_threshold=2&target_volume_rows=65536",
database_path.display()
);
let database = Database::open(&dsn).expect("open persistent database");
database
.execute(
"CREATE TABLE items (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)",
(),
)
.expect("create table");
database
.execute("INSERT INTO items VALUES (1, 'old'), (2, 'two')", ())
.expect("insert first generation");
radixdb_storage::traits::Engine::force_checkpoint_cycle(database.engine().as_ref())
.expect("publish first L0 generation");
database
.execute("UPDATE items SET payload = 'new' WHERE id = 1", ())
.expect("update overlapping row");
database
.execute("INSERT INTO items VALUES (3, 'three')", ())
.expect("insert second generation");
radixdb_storage::traits::Engine::force_checkpoint_cycle(database.engine().as_ref())
.expect("publish second L0 generation");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
loop {
let stats = database.engine().runtime_stats_snapshot();
if stats.cold_l1_segments == 1
&& stats.cold_l0_segments == 0
&& stats.cold_unleveled_segments == 0
&& !stats.compaction_running
{
break;
}
assert!(
std::time::Instant::now() < deadline,
"canonical compaction did not settle: {stats:?}"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
let rows = database
.query("SELECT id, payload FROM items ORDER BY id", ())
.expect("query compacted rows")
.collect_vec()
.expect("collect compacted rows");
assert_eq!(rows.len(), 3);
assert_eq!(rows[0].get::<i64>(0).expect("first id"), 1);
assert_eq!(rows[0].get::<String>(1).expect("updated payload"), "new");
database.close().expect("close compacted database");
let artifact_count_before_reopen = data_artifact_count(&database_path);
let reopened = Database::open(&dsn).expect("reopen compacted database");
let recovered_stats = reopened.engine().runtime_stats_snapshot();
assert_eq!(recovered_stats.cold_unleveled_segments, 0);
assert_eq!(recovered_stats.cold_l0_segments, 0);
assert_eq!(recovered_stats.cold_l1_segments, 1);
let rows = reopened
.query("SELECT id, payload FROM items ORDER BY id", ())
.expect("query recovered compacted rows")
.collect_vec()
.expect("collect recovered compacted rows");
assert_eq!(rows.len(), 3);
assert_eq!(rows[0].get::<i64>(0).expect("first id"), 1);
assert_eq!(rows[0].get::<String>(1).expect("recovered payload"), "new");
radixdb_storage::traits::Engine::force_checkpoint_cycle(reopened.engine().as_ref())
.expect("request post-recovery maintenance");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
loop {
let stats = reopened.engine().runtime_stats_snapshot();
if stats.maintenance.compaction.completed >= 1 && !stats.compaction_running {
assert_eq!(stats.cold_unleveled_segments, 0);
assert_eq!(stats.cold_l0_segments, 0);
assert_eq!(stats.cold_l1_segments, 1);
break;
}
assert!(
std::time::Instant::now() < deadline,
"post-recovery compaction check did not settle: {stats:?}"
);
std::thread::sleep(std::time::Duration::from_millis(25));
}
reopened.close().expect("close reopened database");
assert!(
data_artifact_count(&database_path) <= artifact_count_before_reopen,
"stable L1 recovery must not publish another DATA artifact"
);
}
#[test]
fn named_query_timeout_preserves_named_parameter_binding() {
let database = Database::open_in_memory().expect("open database");
database
.execute("CREATE TABLE named_timeout (id INTEGER PRIMARY KEY)", ())
.expect("create table");
database
.execute("INSERT INTO named_timeout VALUES (7)", ())
.expect("insert row");
let mut rows = database
.query_named_with_timeout(
"SELECT id FROM named_timeout WHERE id = :id",
named_params! { id: 7 },
1_000,
)
.expect("query with named deadline");
assert!(rows.advance());
assert_eq!(rows.current_row().unwrap().get(0), Some(&Value::Integer(7)));
assert!(!rows.advance());
assert!(rows.error().is_none());
}
#[test]
fn r3_l02_batch_b_same_dsn_opens_have_independent_sql_transactions() {
let dsn = "memory://r3-l02-b-independent-opens";
let first = Database::open(dsn).expect("open first connection");
first
.execute(
"CREATE TABLE batch_b_connection (id INTEGER PRIMARY KEY, value INTEGER)",
(),
)
.expect("create table");
let second = Database::open(dsn).expect("open second connection");
first.execute("BEGIN", ()).expect("begin first connection");
first
.execute("INSERT INTO batch_b_connection VALUES (1, 10)", ())
.expect("insert in first connection");
let visible_to_second: i64 = second
.query_one("SELECT COUNT(*) FROM batch_b_connection", ())
.expect("query second connection");
assert_eq!(
visible_to_second, 0,
"a separately opened connection must not inherit the first connection's SQL transaction"
);
first.execute("ROLLBACK", ()).expect("rollback first");
}
#[test]
fn r3_l02_batch_b_engine_get_index_returns_the_existing_public_index() {
use radixdb_storage::Engine;
let database = Database::open_in_memory().expect("open database");
database
.execute(
"CREATE TABLE batch_b_index (id INTEGER PRIMARY KEY, value INTEGER)",
(),
)
.expect("create table");
database
.execute(
"CREATE INDEX idx_batch_b_value ON batch_b_index (value)",
(),
)
.expect("create index");
let index = Engine::get_index(
database.engine().as_ref(),
"batch_b_index",
"idx_batch_b_value",
)
.expect("public Engine::get_index must return an existing index");
assert_eq!(index.name(), "idx_batch_b_value");
}
#[test]
fn r2_l05_c_independent_database_open_does_not_wait_for_another_dsn() {
use std::sync::mpsc;
use std::time::Duration;
let temp = tempfile::tempdir().expect("temp dir");
let slow_dsn = format!("file://{}", temp.path().join("slow").display());
let fast_dsn = format!("file://{}", temp.path().join("fast").display());
let (entered_tx, entered_rx) = mpsc::channel();
let release = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
let release_for_hook = Arc::clone(&release);
let slow_for_hook = slow_dsn.clone();
let open_hook = DatabaseOpenTestHookGuard::install(Arc::new(move |dsn| {
if dsn != slow_for_hook {
return;
}
entered_tx.send(()).expect("report slow open admission");
let (lock, changed) = &*release_for_hook;
let mut released = lock.lock().expect("release lock");
while !*released {
released = changed.wait(released).expect("release wait");
}
}));
let slow_waiter_dsn = slow_dsn.clone();
let slow = std::thread::spawn(move || Database::open(&slow_dsn));
entered_rx
.recv_timeout(Duration::from_secs(2))
.expect("slow open reached deterministic recovery hook");
let same_dsn_waiter = std::thread::spawn(move || Database::open(&slow_waiter_dsn));
let (fast_tx, fast_rx) = mpsc::channel();
let fast = std::thread::spawn(move || {
fast_tx
.send(Database::open(&fast_dsn))
.expect("report independent open result");
});
let independent_completed = fast_rx.recv_timeout(Duration::from_millis(250));
let completed_before_release = independent_completed.is_ok();
{
let (lock, changed) = &*release;
*lock.lock().expect("release lock") = true;
changed.notify_all();
}
let slow_database = slow
.join()
.expect("slow open thread")
.expect("slow database opens");
let same_dsn_database = same_dsn_waiter
.join()
.expect("same DSN waiter thread")
.expect("same DSN waiter receives open outcome");
let fast_database = match independent_completed {
Ok(result) => result.expect("independent database opens"),
Err(_) => fast_rx
.recv_timeout(Duration::from_secs(2))
.expect("independent open eventually reports")
.expect("independent database opens after release"),
};
fast.join().expect("independent open thread");
drop(open_hook);
assert!(Arc::ptr_eq(
slow_database.owner_arc(),
same_dsn_database.owner_arc()
));
drop(same_dsn_database);
slow_database.close().expect("close slow database");
fast_database.close().expect("close fast database");
assert!(
completed_before_release,
"opening one DSN held the process-wide registry lock across recovery"
);
let blocked_parent = temp.path().join("not-a-directory");
std::fs::write(&blocked_parent, b"registry failure oracle")
.expect("create non-directory parent");
let failed_dsn = format!("file://{}", blocked_parent.join("database").display());
let failed_retry_dsn = failed_dsn.clone();
let failed_registry_key =
Database::canonical_registry_key(&failed_dsn).expect("canonical failed-open key");
let failed_waiter_dsn = failed_dsn.clone();
let (failed_entered_tx, failed_entered_rx) = mpsc::channel();
let failed_release = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
let failed_release_for_hook = Arc::clone(&failed_release);
let failed_for_hook = failed_dsn.clone();
let failure_hook = DatabaseOpenTestHookGuard::install(Arc::new(move |dsn| {
if dsn != failed_for_hook {
return;
}
failed_entered_tx
.send(())
.expect("report failed open admission");
let (lock, changed) = &*failed_release_for_hook;
let mut released = lock.lock().expect("failure release lock");
while !*released {
released = changed.wait(released).expect("failure release wait");
}
}));
let failed_owner = std::thread::spawn(move || Database::open(&failed_dsn));
failed_entered_rx
.recv_timeout(Duration::from_secs(2))
.expect("failed open reached hook");
let failed_waiter = std::thread::spawn(move || Database::open(&failed_waiter_dsn));
let waiter_deadline = std::time::Instant::now() + Duration::from_secs(2);
loop {
let waiter_joined = DATABASE_REGISTRY
.read()
.expect("registry read")
.get(&failed_registry_key)
.is_some_and(|entry| {
matches!(
entry,
DatabaseRegistryEntry::Opening(slot) if Arc::strong_count(slot) >= 3
)
});
if waiter_joined {
break;
}
assert!(
std::time::Instant::now() < waiter_deadline,
"same DSN waiter did not join opening slot"
);
std::thread::yield_now();
}
{
let (lock, changed) = &*failed_release;
*lock.lock().expect("failure release lock") = true;
changed.notify_all();
}
let owner_error = match failed_owner.join().expect("failed owner thread") {
Ok(_) => panic!("owner open through a non-directory parent must fail"),
Err(error) => error,
};
let waiter_error = match failed_waiter.join().expect("failed waiter thread") {
Ok(_) => panic!("waiter open through a non-directory parent must fail"),
Err(error) => error,
};
drop(failure_hook);
assert_eq!(owner_error.to_string(), waiter_error.to_string());
assert!(Database::open(&failed_retry_dsn).is_err());
}
#[test]
fn r2_l05_a_database_default_is_connection_local_and_explicit_isolation_wins() {
let db = Database::open_in_memory().expect("open database");
db.set_default_isolation_level(IsolationLevel::SnapshotIsolation)
.expect("set default isolation");
let default_tx = db.begin().expect("begin with database default");
assert_eq!(
db.engine().registry().get_isolation_level(default_tx.id()),
IsolationLevel::SnapshotIsolation
);
let explicit_tx = db
.begin_with_isolation(IsolationLevel::ReadCommitted)
.expect("begin explicit read committed");
assert_eq!(
db.engine().registry().get_isolation_level(explicit_tx.id()),
IsolationLevel::ReadCommitted
);
let clone = db.clone();
let clone_default = clone.begin().expect("clone uses its local default");
assert_eq!(
db.engine()
.registry()
.get_isolation_level(clone_default.id()),
IsolationLevel::ReadCommitted
);
}
#[test]
fn r2_l05_b_failed_close_is_retained_until_retry_and_fresh_open() {
let _failpoint_guard = radixdb_storage::test_failpoints::FailpointGuard::new();
radixdb_storage::test_failpoints::reset_all();
let temp = tempfile::tempdir().expect("temp dir");
let dsn = format!("file://{}", temp.path().join("close-retry").display());
let db = Database::open(&dsn).expect("open database");
db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)", ())
.expect("create table");
db.execute("INSERT INTO t VALUES (1)", ())
.expect("insert durable row");
radixdb_storage::test_failpoints::WAL_SYNC_FAIL
.store(true, std::sync::atomic::Ordering::Release);
let close_error = db.close().expect_err("WAL failure must fail close");
let reopen_error = match Database::open(&dsn) {
Ok(_) => panic!("close-failed registry owner must not reopen as ready"),
Err(error) => error,
};
assert_eq!(close_error.to_string(), reopen_error.to_string());
radixdb_storage::test_failpoints::WAL_SYNC_FAIL
.store(false, std::sync::atomic::Ordering::Release);
db.close().expect("retry close succeeds");
let reopened = Database::open(&dsn).expect("fresh engine opens after terminal close");
assert!(!Arc::ptr_eq(db.owner_arc(), reopened.owner_arc()));
let rows = reopened
.query("SELECT id FROM t", ())
.expect("query recovered row")
.collect_vec()
.expect("collect recovered row");
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get::<i64>(0).expect("recovered id"), 1);
reopened.close().expect("close fresh engine");
radixdb_storage::test_failpoints::reset_all();
}
#[test]
fn test_open_memory() {
let db = Database::open("memory://").unwrap();
assert_eq!(db.dsn(), "memory://");
}
#[test]
fn test_open_in_memory() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
.unwrap();
db.execute("INSERT INTO test VALUES ($1)", (1,)).unwrap();
for row in db.query("SELECT * FROM test", ()).unwrap() {
let row = row.unwrap();
let id: i64 = row.get(0).unwrap();
assert_eq!(id, 1);
}
}
#[test]
fn test_execute_and_query_new_api() {
let db = Database::open_in_memory().unwrap();
// Create table - no params
db.execute(
"CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)",
(),
)
.unwrap();
// Insert with tuple params
let affected = db
.execute(
"INSERT INTO users VALUES ($1, $2, $3), ($4, $5, $6)",
(1, "Alice", 30, 2, "Bob", 25),
)
.unwrap();
assert_eq!(affected, 2);
// Query with tuple params
let rows: Vec<_> = db
.query("SELECT * FROM users ORDER BY id", ())
.unwrap()
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].get::<i64>(0).unwrap(), 1);
assert_eq!(rows[0].get::<String>(1).unwrap(), "Alice");
assert_eq!(rows[0].get::<i64>(2).unwrap(), 30);
}
#[test]
fn test_query_one() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
.unwrap();
db.execute("INSERT INTO test VALUES ($1), ($2), ($3)", (1, 2, 3))
.unwrap();
let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
assert_eq!(count, 3);
}
#[test]
fn test_query_opt() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
.unwrap();
db.execute("INSERT INTO test VALUES ($1)", (1,)).unwrap();
// Found
let result: Option<i64> = db
.query_opt("SELECT id FROM test WHERE id = $1", (1,))
.unwrap();
assert_eq!(result, Some(1));
// Not found
let result: Option<i64> = db
.query_opt("SELECT id FROM test WHERE id = $1", (999,))
.unwrap();
assert_eq!(result, None);
}
#[test]
fn test_params_macro() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)", ())
.unwrap();
// Use params! macro
db.execute(
"INSERT INTO users VALUES ($1, $2)",
crate::params![1, "Alice"],
)
.unwrap();
let names: Vec<String> = db
.query("SELECT name FROM users WHERE id = $1", crate::params![1])
.unwrap()
.map(|r| r.and_then(|row| row.get(0)))
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap();
assert_eq!(names, vec!["Alice"]);
}
#[test]
fn test_parse_dsn() {
// Memory
let (scheme, path) = Database::parse_dsn("memory://").unwrap();
assert_eq!(scheme, "memory");
assert_eq!(path, "");
// File
let (scheme, path) = Database::parse_dsn("file:///tmp/test.db").unwrap();
assert_eq!(scheme, "file");
assert_eq!(path, "/tmp/test.db");
// File with params
let (scheme, path) = Database::parse_dsn("file:///tmp/test.db?sync=full").unwrap();
assert_eq!(scheme, "file");
assert_eq!(path, "/tmp/test.db?sync=full");
// Invalid
assert!(Database::parse_dsn("invalid").is_err());
assert!(Database::parse_dsn("unknown://test").is_err());
}
#[test]
fn test_parse_file_config_seal_hot_bytes() {
let (_path, config) = Database::parse_file_config(
"/tmp/test.db?copy_max_transaction_bytes=16384&max_compaction_jobs=4&storage_cpu_workers=3&page_cache_level=5&page_cache_max_bytes=1073741824&page_cache_memory_reserve=536870912&max_compaction_input_segments=6&max_compaction_input_bytes=33554432&max_compaction_output_bytes=50331648&compaction_job_time_budget_ms=30000&compaction_io_bytes_per_sec=8388608&compaction_disk_reserve_bytes=16777216&compaction_retry_cooldown_ms=5000&l0_soft_limit_segments=10&l0_hard_limit_segments=20&l0_soft_limit_bytes=67108864&l0_hard_limit_bytes=134217728&l0_soft_backpressure_wait_ms=25&seal_hot_bytes_threshold=1024&seal_incremental_hot_bytes_threshold=512&volume_cache_bytes=2048&read_queue_depth=8",
)
.unwrap();
assert_eq!(config.persistence.copy_max_transaction_bytes, 16_384);
assert_eq!(config.persistence.max_compaction_jobs, 4);
assert_eq!(config.persistence.storage_cpu_workers, 3);
assert_eq!(config.persistence.page_cache_level, 5);
assert_eq!(config.persistence.page_cache_max_bytes, 1_073_741_824);
assert_eq!(config.persistence.page_cache_memory_reserve, 536_870_912);
assert_eq!(config.persistence.max_compaction_input_segments, 6);
assert_eq!(config.persistence.max_compaction_input_bytes, 33_554_432);
assert_eq!(config.persistence.max_compaction_output_bytes, 50_331_648);
assert_eq!(config.persistence.compaction_job_time_budget_ms, 30_000);
assert_eq!(config.persistence.compaction_io_bytes_per_sec, 8_388_608);
assert_eq!(config.persistence.compaction_disk_reserve_bytes, 16_777_216);
assert_eq!(config.persistence.compaction_retry_cooldown_ms, 5_000);
assert_eq!(config.persistence.l0_soft_limit_segments, 10);
assert_eq!(config.persistence.l0_hard_limit_segments, 20);
assert_eq!(config.persistence.l0_soft_limit_bytes, 67_108_864);
assert_eq!(config.persistence.l0_hard_limit_bytes, 134_217_728);
assert_eq!(config.persistence.l0_soft_backpressure_wait_ms, 25);
assert_eq!(config.persistence.seal_hot_bytes_threshold, 1024);
assert_eq!(config.persistence.seal_incremental_hot_bytes_threshold, 512);
assert_eq!(config.persistence.volume_cache_bytes, 2048);
assert_eq!(config.persistence.read_queue_depth, 8);
let error = Database::parse_file_config(
"/tmp/test.db?l0_soft_limit_segments=8&l0_hard_limit_segments=8",
)
.expect_err("unordered L0 segment limits must fail closed");
assert!(error.to_string().contains("0 < soft < hard"));
let error = Database::parse_file_config("/tmp/test.db?max_compaction_jobs=0")
.expect_err("zero compaction workers must fail closed");
assert!(error.to_string().contains("max_compaction_jobs must be in"));
let error = Database::parse_file_config("/tmp/test.db?page_cache_level=11")
.expect_err("page cache levels above ten must fail closed");
assert!(error.to_string().contains("page_cache_level must be in"));
}
#[test]
fn file_config_rejects_retired_aliases_and_duplicate_canonical_options() {
for alias in [
"sync=full",
"snapshot_interval=60",
"sync_interval=10",
"snapshot_compression=on",
"seal_hot_bytes=1024",
"seal_incremental_hot_bytes=512",
"volume_cache=2048",
"compression_threshold=64",
"scan_prefetch_cache_bytes=4096",
"block_cache_bytes=8192",
] {
let error = Database::parse_file_config(&format!("/tmp/radix?{alias}"))
.expect_err("retired DSN alias must fail closed");
assert!(error.to_string().contains("unknown file database option"));
}
let error = Database::parse_file_config("/tmp/radix?sync_mode=full&sync_mode=none")
.expect_err("duplicate canonical option must not be order-dependent");
assert!(error.to_string().contains("duplicate file database option"));
}
#[test]
fn r2_l06_batch_a_rejects_obsolete_commit_batch_size() {
let error = Database::parse_file_config("/tmp/radix?commit_batch_size=2")
.expect_err("r2_l06_batch_a: removed durability knob must not be accepted");
assert!(error.to_string().contains("commit_batch_size"));
}
#[test]
fn r2_l06_batch_b_restore_never_commits_before_fallible_cache_admission() {
let dir = tempfile::tempdir().unwrap();
let dsn = format!("file://{}", dir.path().join("restore_cache").display());
let db = Database::open(&dsn).unwrap();
db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)", ())
.unwrap();
db.create_snapshot().unwrap();
db.execute("INSERT INTO t VALUES (1)", ()).unwrap();
let poisoned = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _executor = db.inner.executor.lock().unwrap();
panic!("poison executor before restore");
}));
assert!(poisoned.is_err());
assert!(db.restore_snapshot(None).is_err());
db.close().unwrap();
drop(db);
let reopened = Database::open(&dsn).unwrap();
let count: i64 = reopened.query_one("SELECT COUNT(*) FROM t", ()).unwrap();
assert_eq!(
count, 1,
"restore must not replace durable state before cache admission can succeed"
);
reopened.close().unwrap();
}
#[test]
fn test_from_value_types() {
assert_eq!(i64::from_value(&Value::Integer(42)).unwrap(), 42);
assert_eq!(f64::from_value(&Value::Float(3.5)).unwrap(), 3.5);
assert_eq!(
f64::from_value(&Value::decimal(12_345, 5, 2)).unwrap(),
123.45
);
assert_eq!(
String::from_value(&Value::Text("hello".into())).unwrap(),
"hello"
);
assert!(bool::from_value(&Value::Boolean(true)).unwrap());
// Optional
assert_eq!(
Option::<i64>::from_value(&Value::Integer(42)).unwrap(),
Some(42)
);
assert_eq!(
Option::<i64>::from_value(&Value::null_unknown()).unwrap(),
None
);
}
#[test]
fn test_cached_plan_insert_and_query() {
let db = Database::open_in_memory().unwrap();
db.execute(
"CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, score FLOAT)",
(),
)
.unwrap();
let insert_plan = db
.cached_plan("INSERT INTO test VALUES ($1, $2, $3)")
.unwrap();
// Batch insert using cached plan
db.execute_plan(&insert_plan, (1, "Alice", 95.5)).unwrap();
db.execute_plan(&insert_plan, (2, "Bob", 82.0)).unwrap();
db.execute_plan(&insert_plan, (3, "Charlie", 91.0)).unwrap();
// Query using cached plan
let query_plan = db
.cached_plan("SELECT name FROM test WHERE id = $1")
.unwrap();
let mut rows = db.query_plan(&query_plan, (2,)).unwrap();
let row = rows.next().unwrap().unwrap();
assert_eq!(row.get::<String>(0).unwrap(), "Bob");
}
#[test]
fn test_cached_plan_reuse() {
let db = Database::open_in_memory().unwrap();
db.execute(
"CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
(),
)
.unwrap();
// Get the same plan twice — second call should hit the cache
let plan1 = db.cached_plan("INSERT INTO test VALUES ($1, $2)").unwrap();
let plan2 = db.cached_plan("INSERT INTO test VALUES ($1, $2)").unwrap();
// Both should work independently
db.execute_plan(&plan1, (1, 100)).unwrap();
db.execute_plan(&plan2, (2, 200)).unwrap();
let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
assert_eq!(count, 2);
}
#[test]
fn test_cached_plan_update_delete() {
let db = Database::open_in_memory().unwrap();
db.execute(
"CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
(),
)
.unwrap();
db.execute("INSERT INTO test VALUES (1, 100)", ()).unwrap();
db.execute("INSERT INTO test VALUES (2, 200)", ()).unwrap();
// Update via cached plan
let update_plan = db
.cached_plan("UPDATE test SET value = $1 WHERE id = $2")
.unwrap();
let affected = db.execute_plan(&update_plan, (999, 1)).unwrap();
assert_eq!(affected, 1);
let val: i64 = db
.query_one("SELECT value FROM test WHERE id = 1", ())
.unwrap();
assert_eq!(val, 999);
// Delete via cached plan
let delete_plan = db.cached_plan("DELETE FROM test WHERE id = $1").unwrap();
let affected = db.execute_plan(&delete_plan, (2,)).unwrap();
assert_eq!(affected, 1);
let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
assert_eq!(count, 1);
}
#[test]
fn test_cached_plan_no_params() {
let db = Database::open_in_memory().unwrap();
db.execute(
"CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
(),
)
.unwrap();
db.execute("INSERT INTO test VALUES (1, 10)", ()).unwrap();
db.execute("INSERT INTO test VALUES (2, 20)", ()).unwrap();
let plan = db.cached_plan("SELECT COUNT(*) FROM test").unwrap();
let mut rows = db.query_plan(&plan, ()).unwrap();
let row = rows.next().unwrap().unwrap();
assert_eq!(row.get::<i64>(0).unwrap(), 2);
}
#[test]
fn test_cached_plan_named_params() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)", ())
.unwrap();
let plan = db
.cached_plan("INSERT INTO test VALUES (:id, :name)")
.unwrap();
db.execute_named_plan(&plan, named_params! { id: 1, name: "Alice" })
.unwrap();
db.execute_named_plan(&plan, named_params! { id: 2, name: "Bob" })
.unwrap();
let query_plan = db
.cached_plan("SELECT name FROM test WHERE id = :id")
.unwrap();
let mut rows = db
.query_named_plan(&query_plan, named_params! { id: 1 })
.unwrap();
let row = rows.next().unwrap().unwrap();
assert_eq!(row.get::<String>(0).unwrap(), "Alice");
}
#[test]
fn test_cached_plan_multi_statement_error() {
let db = Database::open_in_memory().unwrap();
db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
.unwrap();
// Multiple statements should fail
let result = db.cached_plan("INSERT INTO test VALUES (1); INSERT INTO test VALUES (2)");
assert!(result.is_err());
}
}