use crate::config::{Durability, FjallConfig, Strategy};
use crate::store::FjallStore;
use fjall::CompressionType;
use nitrite::common::{NitriteModule, NitritePlugin, PluginRegistrar};
use nitrite::errors::NitriteResult;
use nitrite::store::{NitriteStore, StoreEventListener, StoreModule};
pub struct FjallModule {
store_config: FjallConfig,
}
impl FjallModule {
#[inline]
pub fn with_config() -> FjallModuleBuilder {
FjallModuleBuilder::new()
}
}
impl NitriteModule for FjallModule {
fn plugins(&self) -> NitriteResult<Vec<NitritePlugin>> {
let store = self.get_store()?;
let plugin = store.as_plugin();
Ok(vec![plugin])
}
fn load(&self, plugin_registrar: &PluginRegistrar) -> NitriteResult<()> {
let store = self.get_store()?;
plugin_registrar.register_store_plugin(store)
}
}
impl StoreModule for FjallModule {
fn get_store(&self) -> NitriteResult<NitriteStore> {
let store = FjallStore::new(self.store_config.clone());
Ok(NitriteStore::new(store))
}
}
pub struct FjallModuleBuilder {
store_config: FjallConfig,
}
impl Default for FjallModuleBuilder {
fn default() -> Self {
Self::new()
}
}
impl FjallModuleBuilder {
#[inline]
pub fn new() -> FjallModuleBuilder {
FjallModuleBuilder {
store_config: FjallConfig::new(),
}
}
#[inline]
pub fn production_preset(self) -> Self {
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(4);
self
.block_cache_capacity(256 * 1024 * 1024)
.blob_cache_capacity(64 * 1024 * 1024)
.max_write_buffer_size(128 * 1024 * 1024)
.max_memtable_size(32 * 1024 * 1024)
.flush_workers(cpus)
.compaction_workers((cpus / 2).max(1))
.bloom_filter_bits(10)
.fsync_frequency(100) .compression_type(CompressionType::Lz4)
}
#[inline]
pub fn high_throughput_preset(self) -> Self {
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(4);
self
.block_cache_capacity(512 * 1024 * 1024)
.blob_cache_capacity(128 * 1024 * 1024)
.max_write_buffer_size(256 * 1024 * 1024)
.max_memtable_size(64 * 1024 * 1024)
.flush_workers(cpus)
.compaction_workers(cpus)
.manual_journal_persist(true)
.kv_separated(true)
.bloom_filter_bits(0)
}
#[inline]
pub fn low_memory_preset(self) -> Self {
self
.block_cache_capacity(16 * 1024 * 1024)
.blob_cache_capacity(8 * 1024 * 1024)
.max_write_buffer_size(32 * 1024 * 1024)
.max_memtable_size(8 * 1024 * 1024)
.flush_workers(1)
.compaction_workers(1)
.bloom_filter_bits(10)
}
#[inline]
pub fn db_path(self, db_path: &str) -> Self {
self.store_config.set_db_path(db_path);
self
}
#[inline]
pub fn manual_journal_persist(self, manual_journal_persist: bool) -> Self {
self.store_config.set_manual_journal_persist(manual_journal_persist);
self
}
#[inline]
pub fn flush_workers(self, flush_workers_count: usize) -> Self {
self.store_config.set_flush_workers(flush_workers_count);
self
}
#[inline]
pub fn compaction_workers(self, compaction_workers_count: usize) -> Self {
self.store_config.set_compaction_workers(compaction_workers_count);
self
}
#[inline]
pub fn block_cache_capacity(self, block_cache_capacity: u64) -> Self {
self.store_config.set_block_cache_capacity(block_cache_capacity);
self
}
#[inline]
pub fn blob_cache_capacity(self, blob_cache_capacity: u64) -> Self {
self.store_config.set_blob_cache_capacity(blob_cache_capacity);
self
}
#[inline]
pub fn max_journaling_size(self, max_journaling_size: u64) -> Self {
self.store_config.set_max_journaling_size(max_journaling_size);
self
}
#[inline]
pub fn max_write_buffer_size(self, max_write_buffer_size: u64) -> Self {
self.store_config.set_max_write_buffer_size(max_write_buffer_size);
self
}
#[inline]
pub fn fsync_frequency(self, fsync_frequency: u16) -> Self {
self.store_config.set_fsync_frequency(fsync_frequency);
self
}
#[inline]
pub fn durability(self, durability: Durability) -> Self {
self.store_config.set_durability(durability);
self
}
#[inline]
pub fn event_listener(self, listener: StoreEventListener) -> Self {
self.store_config.add_event_listener(listener);
self
}
#[inline]
pub fn bloom_filter_bits(self, bloom_filter_bits: u8) -> Self {
self.store_config.set_bloom_filter_bits(bloom_filter_bits as i8);
self
}
#[inline]
pub fn compression_type(self, compression_type: CompressionType) -> Self {
self.store_config.set_compression_type(compression_type);
self
}
#[inline]
pub fn compaction_strategy(self, compaction_strategy: Strategy) -> Self {
self.store_config.set_compaction_strategy(compaction_strategy);
self
}
#[inline]
pub fn max_memtable_size(self, max_memtable_size: u32) -> Self {
self.store_config.set_max_memtable_size(max_memtable_size);
self
}
#[inline]
pub fn block_size(self, block_size: u32) -> Self {
self.store_config.set_block_size(block_size);
self
}
#[inline]
pub fn kv_separated(self, kv_separated: bool) -> Self {
self.store_config.set_kv_separated(kv_separated);
self
}
#[inline]
pub fn staleness_threshold(self, staleness_threshold: f32) -> Self {
self.store_config.set_staleness_threshold(staleness_threshold);
self
}
#[inline]
pub fn build(self) -> FjallModule {
FjallModule {
store_config: self.store_config,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use fjall::CompressionType;
use nitrite::common::PluginRegistrar;
#[inline(never)]
fn black_box<T>(x: T) -> T {
x
}
#[test]
fn test_fjall_module_with_config() {
let builder = FjallModule::with_config();
assert!(builder.store_config.db_path().is_empty());
}
#[test]
fn test_fjall_module_plugins() {
let module = FjallModule {
store_config: FjallConfig::new(),
};
let plugins = module.plugins();
assert!(plugins.is_ok());
assert_eq!(plugins.unwrap().len(), 1);
}
#[test]
fn test_fjall_module_load() {
let module = FjallModule {
store_config: FjallConfig::new(),
};
let registrar = PluginRegistrar::default();
let result = module.load(®istrar);
assert!(result.is_ok());
}
#[test]
fn test_fjall_module_get_store() {
let module = FjallModule {
store_config: FjallConfig::new(),
};
let store = module.get_store();
assert!(store.is_ok());
}
#[test]
fn test_fjall_module_builder() {
let builder = FjallModuleBuilder::new()
.db_path("test_path")
.manual_journal_persist(true)
.flush_workers(4)
.compaction_workers(2)
.block_cache_capacity(1024)
.blob_cache_capacity(2048)
.max_journaling_size(4096)
.max_write_buffer_size(8192)
.fsync_frequency(10)
.event_listener(StoreEventListener::new(|_| Ok(())))
.bloom_filter_bits(8)
.compression_type(CompressionType::None)
.compaction_strategy(Strategy::default())
.max_memtable_size(1024)
.block_size(4096)
.kv_separated(true)
.staleness_threshold(0.5)
.build();
assert_eq!(builder.store_config.db_path(), "test_path");
assert!(builder.store_config.manual_journal_persist());
assert_eq!(builder.store_config.flush_workers(), 4);
assert_eq!(builder.store_config.compaction_workers(), 2);
assert_eq!(builder.store_config.block_cache_capacity(), 1024);
assert_eq!(builder.store_config.blob_cache_capacity(), 2048);
assert_eq!(builder.store_config.max_journaling_size(), 4096);
assert_eq!(builder.store_config.max_write_buffer_size(), 8192);
assert_eq!(builder.store_config.fsync_frequency(), 10);
assert_eq!(builder.store_config.bloom_filter_bits(), 8);
assert_eq!(builder.store_config.compression_type(), CompressionType::None);
assert_eq!(builder.store_config.max_memtable_size(), 1024);
assert_eq!(builder.store_config.block_size(), 4096);
assert!(builder.store_config.kv_separated());
assert_eq!(builder.store_config.staleness_threshold(), 0.5);
}
#[test]
fn test_fjall_module_creation_perf() {
for _ in 0..1000 {
let module = black_box(FjallModule {
store_config: FjallConfig::new(),
});
black_box(module);
}
}
#[test]
fn test_fjall_module_builder_chain_perf() {
for _ in 0..500 {
let _module = black_box(
FjallModuleBuilder::new()
.db_path("test_path")
.flush_workers(4)
.compaction_workers(2)
.block_cache_capacity(256 * 1_024 * 1_024)
.blob_cache_capacity(64 * 1_024 * 1_024)
.build()
);
}
}
#[test]
fn test_builder_inline_efficiency() {
for _ in 0..1000 {
let builder = black_box(FjallModuleBuilder::new());
let builder = black_box(builder.db_path("path1"));
let builder = black_box(builder.flush_workers(8));
let builder = black_box(builder.block_cache_capacity(512 * 1_024 * 1_024));
let module = black_box(builder.build());
black_box(module);
}
}
#[test]
fn test_builder_single_allocation_per_chain() {
for _ in 0..500 {
let module = black_box(
FjallModuleBuilder::new()
.db_path("test")
.manual_journal_persist(true)
.flush_workers(4)
.compaction_workers(2)
.build()
);
black_box(module.store_config.db_path());
}
}
#[test]
fn test_fjall_module_creation_efficiency() {
for _ in 0..5000 {
let module = black_box(FjallModule {
store_config: FjallConfig::new(),
});
black_box(module);
}
}
#[test]
fn test_builder_with_cache_settings_perf() {
for _ in 0..1000 {
let module = black_box(
FjallModuleBuilder::new()
.block_cache_capacity(256 * 1_024 * 1_024)
.blob_cache_capacity(64 * 1_024 * 1_024)
.max_write_buffer_size(128 * 1_024 * 1_024)
.build()
);
black_box(module);
}
}
#[test]
fn test_builder_method_call_count() {
for _ in 0..500 {
let builder = FjallModuleBuilder::new();
let builder = builder.db_path("path");
let builder = builder.flush_workers(4);
let builder = builder.compaction_workers(2);
let builder = builder.block_cache_capacity(256 * 1_024 * 1_024);
let builder = builder.bloom_filter_bits(8);
let builder = builder.kv_separated(true);
let _module = black_box(builder.build());
}
}
#[test]
fn test_store_module_get_store_perf() {
let module = FjallModule {
store_config: FjallConfig::new(),
};
for _ in 0..500 {
let store = black_box(module.get_store().ok());
black_box(store);
}
}
#[test]
fn test_plugins_method_efficiency() {
let module = FjallModule {
store_config: FjallConfig::new(),
};
for _ in 0..500 {
let plugins = black_box(module.plugins().ok());
black_box(plugins);
}
}
#[test]
fn test_production_preset() {
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(4);
let module = FjallModuleBuilder::new()
.production_preset()
.db_path("prod_test")
.build();
assert_eq!(module.store_config.db_path(), "prod_test");
assert_eq!(module.store_config.block_cache_capacity(), 256 * 1024 * 1024);
assert_eq!(module.store_config.blob_cache_capacity(), 64 * 1024 * 1024);
assert_eq!(module.store_config.max_write_buffer_size(), 128 * 1024 * 1024);
assert_eq!(module.store_config.max_memtable_size(), 32 * 1024 * 1024);
assert_eq!(module.store_config.flush_workers(), cpus);
assert_eq!(module.store_config.compaction_workers(), (cpus / 2).max(1));
assert_eq!(module.store_config.bloom_filter_bits(), 10);
assert_eq!(module.store_config.fsync_frequency(), 100);
assert_eq!(module.store_config.compression_type(), CompressionType::Lz4);
}
#[test]
fn test_high_throughput_preset() {
let cpus = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(4);
let module = FjallModuleBuilder::new()
.high_throughput_preset()
.db_path("throughput_test")
.build();
assert_eq!(module.store_config.block_cache_capacity(), 512 * 1024 * 1024);
assert_eq!(module.store_config.blob_cache_capacity(), 128 * 1024 * 1024);
assert_eq!(module.store_config.max_write_buffer_size(), 256 * 1024 * 1024);
assert_eq!(module.store_config.max_memtable_size(), 64 * 1024 * 1024);
assert_eq!(module.store_config.flush_workers(), cpus);
assert_eq!(module.store_config.compaction_workers(), cpus);
assert!(module.store_config.manual_journal_persist());
assert!(module.store_config.kv_separated());
assert_eq!(module.store_config.bloom_filter_bits(), 0);
}
#[test]
fn test_low_memory_preset() {
let module = FjallModuleBuilder::new()
.low_memory_preset()
.db_path("low_mem_test")
.build();
assert_eq!(module.store_config.block_cache_capacity(), 16 * 1024 * 1024);
assert_eq!(module.store_config.blob_cache_capacity(), 8 * 1024 * 1024);
assert_eq!(module.store_config.max_write_buffer_size(), 32 * 1024 * 1024);
assert_eq!(module.store_config.max_memtable_size(), 8 * 1024 * 1024);
assert_eq!(module.store_config.flush_workers(), 1);
assert_eq!(module.store_config.compaction_workers(), 1);
assert_eq!(module.store_config.bloom_filter_bits(), 10);
}
#[test]
fn test_preset_override() {
let module = FjallModuleBuilder::new()
.production_preset()
.block_cache_capacity(1024 * 1024 * 1024) .build();
assert_eq!(module.store_config.block_cache_capacity(), 1024 * 1024 * 1024);
assert_eq!(module.store_config.bloom_filter_bits(), 10);
}
#[test]
fn test_preset_chain_performance() {
for _ in 0..500 {
let module = black_box(
FjallModuleBuilder::new()
.production_preset()
.db_path("perf_test")
.build()
);
black_box(module);
}
}
}