src/builder.rs
// Apache-2.0 licensed.
//! Builder for [`RedbCatalog`].
use std::collections::HashMap;
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use iceberg::io::{FileIOBuilder, StorageFactory};
use iceberg::{CatalogBuilder, Error, ErrorKind, Result};
use crate::catalog::RedbCatalog;
use crate::error::map_redb;
use crate::store::Store;
/// Property key for the redb database file path. May be set either via
/// [`RedbCatalogBuilder::db_path`] or in the `props` map passed to
/// [`iceberg::CatalogBuilder::load`].
pub const REDB_CATALOG_PROP_DB_PATH: &str = "redb.db-path";
/// Property key for the warehouse root URI. May be set either via
/// [`RedbCatalogBuilder::warehouse_location`] or in the `props` map passed to
/// [`iceberg::CatalogBuilder::load`].
pub const REDB_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
/// Property key for the L0 parsed-metadata cache budget, in bytes. May be set
/// via [`RedbCatalogBuilder::metadata_cache_bytes`] or in the `props` map. A
/// value of `0` disables the cache. Defaults to
/// [`crate::DEFAULT_METADATA_CACHE_BYTES`] (128 MiB) when unset.
pub const REDB_CATALOG_PROP_METADATA_CACHE_BYTES: &str = "redb.metadata-cache-bytes";
/// Property key for the L1.5 built-`Table` handle cache budget, in bytes. May be
/// set via [`RedbCatalogBuilder::table_handle_cache_bytes`] or in the `props`
/// map. A value of `0` disables the cache. Defaults to
/// [`crate::DEFAULT_TABLE_HANDLE_CACHE_BYTES`] (256 MiB) when unset.
pub const REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_BYTES: &str = "redb.table-handle-cache-bytes";
/// **Deprecated** property key: the L1.5 cache expressed as a handle COUNT.
/// An entry count cannot bound a cache whose entries are unbounded — see
/// [`crate::table_cache`]. Still honoured (converted to bytes at the worst case
/// per handle) so existing configs keep working, and ignored when
/// [`REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_BYTES`] is also set.
pub const REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_CAPACITY: &str = "redb.table-handle-cache-capacity";
/// Property key for write durability. Values: `immediate` (default, fsync per
/// commit), `eventual`, `none`. May be set via
/// [`RedbCatalogBuilder::durability`] or in the `props` map.
pub const REDB_CATALOG_PROP_DURABILITY: &str = "redb.durability";
/// Durability applied to every catalog-mutation commit. Maps onto redb's
/// durability levels.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum WriteDurability {
/// fsync on every commit — crash-safe. The default.
#[default]
Immediate,
/// Queued for persistence; persists shortly after `commit` returns. Higher
/// write throughput, may lose the last few commits on a hard crash (the
/// database stays internally consistent).
Eventual,
/// No fsync — fastest writes, durability is best-effort. Intended for bulk
/// import / benchmarking. Note: redb only frees pages at higher durability
/// levels, so exclusive `None` use grows the file.
None,
}
impl WriteDurability {
pub(crate) fn to_redb(self) -> redb::Durability {
match self {
WriteDurability::Immediate => redb::Durability::Immediate,
WriteDurability::Eventual => redb::Durability::Eventual,
WriteDurability::None => redb::Durability::None,
}
}
fn parse(s: &str) -> Option<Self> {
match s.trim().to_ascii_lowercase().as_str() {
"immediate" => Some(Self::Immediate),
"eventual" => Some(Self::Eventual),
"none" => Some(Self::None),
_ => None,
}
}
}
/// Builder for [`RedbCatalog`].
#[derive(Debug, Default)]
pub struct RedbCatalogBuilder {
db_path: Option<PathBuf>,
warehouse_location: Option<String>,
storage_factory: Option<Arc<dyn StorageFactory>>,
metadata_cache_bytes: Option<u64>,
table_handle_cache_bytes: Option<u64>,
durability: Option<WriteDurability>,
props: HashMap<String, String>,
/// Set by the iceberg `CatalogBuilder` hooks skade does not support yet
/// (table encryption, a custom runtime); `load` refuses by name instead of
/// silently ignoring them. See [`CatalogBuilder for RedbCatalogBuilder`].
unsupported: Vec<&'static str>,
}
impl RedbCatalogBuilder {
/// Set the redb database file path.
pub fn db_path(mut self, path: impl Into<PathBuf>) -> Self {
self.db_path = Some(path.into());
self
}
/// Set the warehouse root URI (e.g. `file:///var/.../warehouse`).
pub fn warehouse_location(mut self, loc: impl Into<String>) -> Self {
self.warehouse_location = Some(loc.into());
self
}
/// Set the L0 parsed-metadata cache budget in bytes (e.g. `64 * 1024 * 1024`
/// for 64 MiB). `0` disables the cache. Defaults to
/// [`crate::DEFAULT_METADATA_CACHE_BYTES`] (128 MiB) when unset.
pub fn metadata_cache_bytes(mut self, bytes: u64) -> Self {
self.metadata_cache_bytes = Some(bytes);
self
}
/// Set the L1.5 built-`Table` handle cache budget in bytes (e.g.
/// `64 * 1024 * 1024` for 64 MiB). `0` disables it. Defaults to
/// [`crate::DEFAULT_TABLE_HANDLE_CACHE_BYTES`] (256 MiB) when unset. A warm
/// `load_table` hit then costs an `Arc`-sharing `Table` clone (~100 ns)
/// instead of rebuilding iceberg's per-`Table` `ObjectCache` (~14 µs).
pub fn table_handle_cache_bytes(mut self, bytes: u64) -> Self {
self.table_handle_cache_bytes = Some(bytes);
self
}
/// **Deprecated** — set the L1.5 cache as a handle COUNT. A count bounds
/// nothing here: one handle pins its parsed metadata plus a 32 MiB-by-default
/// iceberg `ObjectCache`. The count is converted to the byte budget it could
/// pin in the worst case; use [`Self::table_handle_cache_bytes`] instead.
#[deprecated(
since = "0.9.11",
note = "an entry count cannot bound a cache whose entries are unbounded; use table_handle_cache_bytes"
)]
pub fn table_handle_cache_capacity(mut self, capacity: u64) -> Self {
self.table_handle_cache_bytes = Some(crate::table_cache::bytes_for_handle_count(capacity));
self
}
/// Set write durability for catalog mutations. Defaults to
/// [`WriteDurability::Immediate`] (fsync per commit).
pub fn durability(mut self, durability: WriteDurability) -> Self {
self.durability = Some(durability);
self
}
/// Bulk-set arbitrary properties. Equivalent to repeated `prop` calls.
pub fn props(mut self, props: HashMap<String, String>) -> Self {
for (k, v) in props {
self.props.insert(k, v);
}
self
}
/// Set a single property.
pub fn prop(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.props.insert(key.into(), value.into());
self
}
}
impl CatalogBuilder for RedbCatalogBuilder {
type C = RedbCatalog;
fn with_storage_factory(mut self, storage_factory: Arc<dyn StorageFactory>) -> Self {
self.storage_factory = Some(storage_factory);
self
}
// iceberg-rust main (2026-10-02 rebase) added these two to the trait. skade
// writes plaintext and builds its tables on the ambient tokio runtime, so
// both are REFUSED at `load` rather than accepted and ignored: an ignored
// KMS factory would write plaintext for a table that asked to be encrypted.
fn with_kms_client_factory(
mut self,
_kms_client_factory: Arc<dyn iceberg::encryption::kms::KmsClientFactory>,
) -> Self {
self.unsupported.push("table encryption (with_kms_client_factory)");
self
}
fn with_runtime(mut self, _runtime: iceberg::Runtime) -> Self {
self.unsupported.push("a custom runtime (with_runtime)");
self
}
fn load(
mut self,
name: impl Into<String>,
props: HashMap<String, String>,
) -> impl Future<Output = Result<Self::C>> + Send {
for (k, v) in props {
self.props.insert(k, v);
}
if let Some(p) = self.props.remove(REDB_CATALOG_PROP_DB_PATH) {
self.db_path = Some(PathBuf::from(p));
}
if let Some(w) = self.props.remove(REDB_CATALOG_PROP_WAREHOUSE) {
self.warehouse_location = Some(w);
}
let cache_bytes_prop = self.props.remove(REDB_CATALOG_PROP_METADATA_CACHE_BYTES);
let table_cache_bytes_prop = self.props.remove(REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_BYTES);
let table_cache_count_prop = self
.props
.remove(REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_CAPACITY);
let durability_prop = self.props.remove(REDB_CATALOG_PROP_DURABILITY);
let name = name.into();
let db_path = self.db_path.clone();
let warehouse_location = self.warehouse_location.clone();
let storage_factory = self.storage_factory.clone();
let metadata_cache_bytes = self.metadata_cache_bytes;
let table_handle_cache_bytes = self.table_handle_cache_bytes;
let durability = self.durability;
let unsupported = std::mem::take(&mut self.unsupported);
async move {
if !unsupported.is_empty() {
return Err(Error::new(
ErrorKind::FeatureUnsupported,
format!("redb catalog does not support {}", unsupported.join(" or ")),
));
}
if name.trim().is_empty() {
return Err(Error::new(
ErrorKind::DataInvalid,
"Catalog name cannot be empty",
));
}
let db_path = db_path.ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"redb catalog requires `{REDB_CATALOG_PROP_DB_PATH}` to be set, \
either via RedbCatalogBuilder::db_path or props"
),
)
})?;
let warehouse_location = warehouse_location.ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"redb catalog requires `{REDB_CATALOG_PROP_WAREHOUSE}` to be set, \
either via RedbCatalogBuilder::warehouse_location or props"
),
)
})?;
let factory = storage_factory.ok_or_else(|| {
Error::new(
ErrorKind::Unexpected,
"StorageFactory must be provided for RedbCatalog. \
Use `with_storage_factory` to configure it.",
)
})?;
// A `props` entry overrides the typed builder field (consistent with
// db-path / warehouse handling above); otherwise fall back to the
// field, then the crate default.
let metadata_cache_bytes = match cache_bytes_prop {
Some(s) => s.trim().parse::<u64>().map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
format!(
"`{REDB_CATALOG_PROP_METADATA_CACHE_BYTES}` must be a non-negative \
integer number of bytes, got `{s}`"
),
)
})?,
None => metadata_cache_bytes.unwrap_or(crate::DEFAULT_METADATA_CACHE_BYTES),
};
// Bytes win over the deprecated count, which wins over the typed
// builder field, which falls back to the crate default.
let table_handle_cache_bytes = match (table_cache_bytes_prop, table_cache_count_prop) {
(Some(s), _) => s.trim().parse::<u64>().map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
format!(
"`{REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_BYTES}` must be a \
non-negative integer number of bytes, got `{s}`"
),
)
})?,
(None, Some(s)) => {
let n = s.trim().parse::<u64>().map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
format!(
"`{REDB_CATALOG_PROP_TABLE_HANDLE_CACHE_CAPACITY}` must be a \
non-negative integer, got `{s}`"
),
)
})?;
crate::table_cache::bytes_for_handle_count(n)
}
(None, None) => table_handle_cache_bytes
.unwrap_or(crate::DEFAULT_TABLE_HANDLE_CACHE_BYTES),
};
let durability = match durability_prop {
Some(s) => WriteDurability::parse(&s).ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"`{REDB_CATALOG_PROP_DURABILITY}` must be one of \
immediate|eventual|none, got `{s}`"
),
)
})?,
None => durability.unwrap_or_default(),
};
let fileio = FileIOBuilder::new(factory).build();
if let Some(parent) = db_path.parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent).map_err(|e| {
Error::new(
ErrorKind::Unexpected,
format!("failed to create catalog dir {}: {e}", parent.display()),
)
})?;
}
}
let store = Store::open(db_path, durability.to_redb()).map_err(map_redb)?;
Ok(RedbCatalog {
name,
warehouse_location,
fileio,
store,
meta_cache: crate::meta_cache::MetadataCache::new(metadata_cache_bytes),
table_cache: crate::table_cache::TableHandleCache::new(table_handle_cache_bytes),
})
}
}
}