1use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
5use noxu_tree::{KeyComparatorFn, Tree};
6use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
7use std::sync::{Arc, RwLock};
8
9use crate::dup_key_data;
10use crate::throughput_stats::ThroughputStats;
11
12use crate::{DatabaseConfig, DatabaseId, DbType};
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16enum DeleteState {
17 NotDeleted,
18 DeletedCleanupInListHarvest,
19 DeletedCleanupLogHarvest,
20 Deleted,
21}
22
23const DUPS_ENABLED: u8 = 0x01;
25const TEMPORARY_BIT: u8 = 0x02;
26const IS_REPLICATED_BIT: u8 = 0x04;
27const NOT_REPLICATED_BIT: u8 = 0x08;
28const PREFIXING_ENABLED: u8 = 0x10;
29
30pub struct DatabaseImpl {
34 id: DatabaseId,
36 name: String,
38 db_type: DbType,
40 flags: u8,
42 delete_state: DeleteState,
44 dirty: AtomicBool,
46 max_tree_entries_per_node: i32,
48 reference_count: AtomicI64,
50 tree: Option<DatabaseTree>,
53 real_tree: Option<Arc<RwLock<Tree>>>,
62 deferred_write: bool,
68 entry_count: Arc<AtomicU64>,
77 pub throughput: Arc<ThroughputStats>,
83 btree_comparator_id: Option<String>,
91 duplicate_comparator_id: Option<String>,
95 triggers: Vec<Arc<dyn crate::trigger::Trigger>>,
104}
105
106#[derive(Debug)]
114pub struct DatabaseTree {
115 root_lsn: u64,
117}
118
119impl Default for DatabaseTree {
120 fn default() -> Self {
121 Self::new()
122 }
123}
124
125impl DatabaseTree {
126 pub fn new() -> Self {
127 DatabaseTree { root_lsn: noxu_util::NULL_LSN.as_u64() }
128 }
129 pub fn get_root_lsn(&self) -> u64 {
130 self.root_lsn
131 }
132 pub fn set_root_lsn(&mut self, lsn: u64) {
133 self.root_lsn = lsn;
134 }
135}
136
137impl DatabaseImpl {
138 pub fn new(
140 id: DatabaseId,
141 name: String,
142 db_type: DbType,
143 config: &DatabaseConfig,
144 ) -> Self {
145 let mut flags = 0u8;
146 if config.sorted_duplicates {
147 flags |= DUPS_ENABLED;
148 }
149 if config.temporary {
150 flags |= TEMPORARY_BIT;
151 }
152 if config.key_prefixing {
153 flags |= PREFIXING_ENABLED;
154 }
155
156 let max_entries = config.node_max_entries as usize;
157 let btree_comparator_id =
158 config.btree_comparator.as_ref().map(|c| c.identity.clone());
159 let duplicate_comparator_id =
160 config.duplicate_comparator.as_ref().map(|c| c.identity.clone());
161 let real_tree = Self::build_tree(id, max_entries, config);
162 let mut real_tree = real_tree;
168 real_tree.set_key_prefixing(config.key_prefixing);
169 DatabaseImpl {
170 id,
171 name,
172 db_type,
173 flags,
174 delete_state: DeleteState::NotDeleted,
175 dirty: AtomicBool::new(false),
176 max_tree_entries_per_node: config.node_max_entries,
177 reference_count: AtomicI64::new(0),
178 tree: Some(DatabaseTree::new()),
179 real_tree: Some(Arc::new(RwLock::new(real_tree))),
180 deferred_write: config.deferred_write,
181 entry_count: Arc::new(AtomicU64::new(0)),
182 throughput: ThroughputStats::new(),
183 btree_comparator_id,
184 duplicate_comparator_id,
185 triggers: config.triggers.clone(),
186 }
187 }
188
189 fn build_tree(
206 id: DatabaseId,
207 max_entries: usize,
208 config: &DatabaseConfig,
209 ) -> Tree {
210 let btree_fn: Option<KeyComparatorFn> =
211 config.btree_comparator.as_ref().map(|c| c.func.clone());
212 if config.sorted_duplicates {
213 let dup_fn: Option<KeyComparatorFn> =
214 config.duplicate_comparator.as_ref().map(|c| c.func.clone());
215 let dup_cmp: KeyComparatorFn =
216 Arc::new(move |a: &[u8], b: &[u8]| {
217 dup_key_data::cmp_two_part_keys(
218 a,
219 b,
220 |x, y| match &btree_fn {
221 Some(f) => f(x, y),
222 None => x.cmp(y),
223 },
224 |x, y| match &dup_fn {
225 Some(f) => f(x, y),
226 None => x.cmp(y),
227 },
228 )
229 });
230 Tree::new_with_comparator(id.id() as u64, max_entries, dup_cmp)
231 } else if let Some(btree_fn) = btree_fn {
232 Tree::new_with_comparator(id.id() as u64, max_entries, btree_fn)
233 } else {
234 Tree::new(id.id() as u64, max_entries)
235 }
236 }
237
238 pub fn get_id(&self) -> DatabaseId {
240 self.id
241 }
242 pub fn get_name(&self) -> &str {
243 &self.name
244 }
245 pub fn get_db_type(&self) -> DbType {
246 self.db_type
247 }
248
249 pub fn is_deferred_write(&self) -> bool {
253 self.deferred_write
254 }
255
256 pub fn get_sorted_duplicates(&self) -> bool {
258 self.flags & DUPS_ENABLED != 0
259 }
260
261 pub fn btree_comparator_id(&self) -> Option<&str> {
265 self.btree_comparator_id.as_deref()
266 }
267
268 pub fn duplicate_comparator_id(&self) -> Option<&str> {
272 self.duplicate_comparator_id.as_deref()
273 }
274
275 pub fn triggers(&self) -> &[Arc<dyn crate::trigger::Trigger>] {
279 &self.triggers
280 }
281
282 pub fn has_user_triggers(&self) -> bool {
287 !self.triggers.is_empty()
288 }
289
290 pub fn is_ln_immediately_obsolete(&self) -> bool {
300 self.get_sorted_duplicates()
301 }
304 pub fn is_temporary(&self) -> bool {
305 self.flags & TEMPORARY_BIT != 0
306 }
307 pub fn get_key_prefixing(&self) -> bool {
308 self.flags & PREFIXING_ENABLED != 0
309 }
310 pub fn is_replicated(&self) -> bool {
311 self.flags & IS_REPLICATED_BIT != 0
312 }
313 pub fn set_replicated(&mut self, replicated: bool) {
319 if replicated {
320 self.flags |= IS_REPLICATED_BIT;
321 self.flags &= !NOT_REPLICATED_BIT;
322 } else {
323 self.flags |= NOT_REPLICATED_BIT;
324 self.flags &= !IS_REPLICATED_BIT;
325 }
326 }
327
328 pub fn is_deleted(&self) -> bool {
330 self.delete_state == DeleteState::Deleted
331 }
332 pub fn is_deleting(&self) -> bool {
333 self.delete_state != DeleteState::NotDeleted
334 }
335 pub fn start_delete(&mut self) {
336 self.delete_state = DeleteState::DeletedCleanupInListHarvest;
337 }
338 pub fn finish_delete(&mut self) {
339 self.delete_state = DeleteState::Deleted;
340 }
341
342 pub fn is_dirty(&self) -> bool {
344 self.dirty.load(Ordering::Relaxed)
345 }
346 pub fn set_dirty(&self) {
347 self.dirty.store(true, Ordering::Relaxed);
348 }
349 pub fn clear_dirty(&self) {
350 self.dirty.store(false, Ordering::Relaxed);
351 }
352
353 pub fn increment_reference_count(&self) {
355 self.reference_count.fetch_add(1, Ordering::Relaxed);
356 }
357 pub fn decrement_reference_count(&self) {
358 self.reference_count.fetch_sub(1, Ordering::Relaxed);
359 }
360 pub fn reference_count(&self) -> i64 {
361 self.reference_count.load(Ordering::Relaxed)
362 }
363
364 pub fn entry_count(&self) -> u64 {
369 self.entry_count.load(Ordering::Relaxed)
370 }
371
372 pub fn increment_entry_count(&self) {
374 self.entry_count.fetch_add(1, Ordering::Relaxed);
375 }
376
377 pub fn decrement_entry_count(&self) {
379 loop {
381 let cur = self.entry_count.load(Ordering::Relaxed);
382 if cur == 0 {
383 break;
384 }
385 if self
386 .entry_count
387 .compare_exchange_weak(
388 cur,
389 cur - 1,
390 Ordering::Relaxed,
391 Ordering::Relaxed,
392 )
393 .is_ok()
394 {
395 break;
396 }
397 }
398 }
399
400 pub fn get_tree(&self) -> Option<&DatabaseTree> {
402 self.tree.as_ref()
403 }
404 pub fn get_tree_mut(&mut self) -> Option<&mut DatabaseTree> {
405 self.tree.as_mut()
406 }
407
408 pub fn get_real_tree(
422 &self,
423 ) -> Option<std::sync::RwLockReadGuard<'_, Tree>> {
424 self.real_tree.as_ref()?.read().ok()
425 }
426
427 pub fn get_real_tree_arc(&self) -> Option<Arc<RwLock<Tree>>> {
430 self.real_tree.clone()
431 }
432
433 pub fn update_key_expiration(
439 &self,
440 key: &[u8],
441 expiration_hours: u32,
442 ) -> bool {
443 self.real_tree
444 .as_ref()
445 .and_then(|arc| arc.read().ok())
446 .map(|t| t.update_key_expiration(key, expiration_hours))
447 .unwrap_or(false)
448 }
449
450 pub fn collect_btree_stats(&self) -> Option<noxu_tree::TreeStats> {
458 self.real_tree
459 .as_ref()
460 .and_then(|arc| arc.read().ok())
461 .map(|t| t.collect_stats())
462 }
463
464 pub fn set_recovered_tree(&mut self, mut tree: Tree) {
469 let count = tree.count_entries();
472 self.entry_count.store(count, std::sync::atomic::Ordering::Relaxed);
473 let mut had_comparator = false;
476 if let Some(ref current_arc) = self.real_tree
477 && let Ok(mut current) = current_arc.write()
478 && let Some(cmp) = current.take_comparator()
479 {
480 tree.set_comparator(cmp);
481 had_comparator = true;
482 }
483 tree.set_key_prefixing(self.flags & PREFIXING_ENABLED != 0);
490 if had_comparator {
497 tree.resort_under_comparator();
498 }
499 self.real_tree = Some(Arc::new(RwLock::new(tree)));
500 }
501
502 pub fn set_memory_counter(
509 &mut self,
510 counter: std::sync::Arc<std::sync::atomic::AtomicI64>,
511 ) {
512 if let Some(tree_arc) = self.real_tree.as_ref()
513 && let Ok(mut tree) = tree_arc.write()
514 {
515 tree.set_memory_counter(counter);
516 }
517 }
518
519 pub fn set_tree_compact_max_key_length(&mut self, len: i32) {
523 if let Some(tree_arc) = self.real_tree.as_ref()
524 && let Ok(mut tree) = tree_arc.write()
525 {
526 tree.set_compact_max_key_length(len);
527 }
528 }
529
530 pub fn max_tree_entries_per_node(&self) -> i32 {
532 self.max_tree_entries_per_node
533 }
534
535 pub fn log_size(&self) -> usize {
538 8 + 4 + self.name.len() + 1 + 4 + 8 }
544
545 pub fn write_to_log(&self, buf: &mut Vec<u8>) -> std::io::Result<()> {
546 buf.write_i64::<BigEndian>(self.id.id())?;
547 buf.write_u32::<BigEndian>(self.name.len() as u32)?;
548 buf.extend_from_slice(self.name.as_bytes());
549 buf.write_u8(self.flags)?;
550 buf.write_i32::<BigEndian>(self.max_tree_entries_per_node)?;
551 let root_lsn = self
552 .tree
553 .as_ref()
554 .map_or(noxu_util::NULL_LSN.as_u64(), |t| t.root_lsn);
555 buf.write_u64::<BigEndian>(root_lsn)?;
556 Ok(())
557 }
558
559 pub fn read_from_log(buf: &[u8]) -> std::io::Result<Self> {
560 fn type_for_db_name(name: &str) -> DbType {
562 match name {
563 "_jeIdMap" | "_noxuIdMap" => DbType::Id,
564 "_jeNameMap" | "_noxuNameMap" => DbType::Name,
565 "_jeUtilization" | "_noxuUtilization" => DbType::Utilization,
566 _ => DbType::User,
567 }
568 }
569 use std::io::Cursor;
570
571 let mut cursor = Cursor::new(buf);
572 let id = cursor.read_i64::<BigEndian>()?;
573 let name_len = cursor.read_u32::<BigEndian>()? as usize;
574
575 let name_start = cursor.position() as usize;
577 let name_end = name_start + name_len;
578 if name_end > buf.len() {
579 return Err(std::io::Error::new(
580 std::io::ErrorKind::UnexpectedEof,
581 "Buffer too short for name",
582 ));
583 }
584 let name = String::from_utf8(buf[name_start..name_end].to_vec())
585 .map_err(|e| {
586 std::io::Error::new(std::io::ErrorKind::InvalidData, e)
587 })?;
588 cursor.set_position(name_end as u64);
589
590 let flags = cursor.read_u8()?;
591 let max_entries = cursor.read_i32::<BigEndian>()?;
592 let root_lsn = cursor.read_u64::<BigEndian>()?;
593
594 let db_type = type_for_db_name(&name);
595
596 let mut tree = DatabaseTree::new();
597 tree.root_lsn = root_lsn;
598
599 let real_tree = Tree::new(id as u64, max_entries as usize);
600 Ok(DatabaseImpl {
601 id: DatabaseId::new(id),
602 name,
603 db_type,
604 flags,
605 delete_state: DeleteState::NotDeleted,
606 dirty: AtomicBool::new(false),
607 max_tree_entries_per_node: max_entries,
608 reference_count: AtomicI64::new(0),
609 tree: Some(tree),
610 real_tree: Some(Arc::new(RwLock::new(real_tree))),
611 deferred_write: false, entry_count: Arc::new(AtomicU64::new(0)),
613 throughput: ThroughputStats::new(),
614 btree_comparator_id: None,
615 duplicate_comparator_id: None,
616 triggers: Vec::new(),
620 })
621 }
622}
623
624impl std::fmt::Debug for DatabaseImpl {
625 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
626 f.debug_struct("DatabaseImpl")
627 .field("id", &self.id)
628 .field("name", &self.name)
629 .field("db_type", &self.db_type)
630 .field("flags", &self.flags)
631 .field("delete_state", &self.delete_state)
632 .finish()
633 }
634}
635
636#[cfg(test)]
637#[expect(clippy::field_reassign_with_default)]
638mod tests {
639 use super::*;
640
641 fn make_config() -> DatabaseConfig {
642 DatabaseConfig::default()
643 }
644
645 #[test]
646 fn test_new_database() {
647 let config = make_config();
648 let db = DatabaseImpl::new(
649 DatabaseId::new(100),
650 "test_db".to_string(),
651 DbType::User,
652 &config,
653 );
654
655 assert_eq!(db.get_id(), DatabaseId::new(100));
656 assert_eq!(db.get_name(), "test_db");
657 assert_eq!(db.get_db_type(), DbType::User);
658 assert!(!db.is_deleted());
659 assert!(!db.is_deleting());
660 assert_eq!(db.reference_count(), 0);
661 }
662
663 #[test]
664 fn test_sorted_duplicates_flag() {
665 let mut config = DatabaseConfig::default();
666 config.sorted_duplicates = false;
667 let db1 = DatabaseImpl::new(
668 DatabaseId::new(1),
669 "db1".to_string(),
670 DbType::User,
671 &config,
672 );
673 assert!(!db1.get_sorted_duplicates());
674
675 config.sorted_duplicates = true;
676 let db2 = DatabaseImpl::new(
677 DatabaseId::new(2),
678 "db2".to_string(),
679 DbType::User,
680 &config,
681 );
682 assert!(db2.get_sorted_duplicates());
683 }
684
685 #[test]
686 fn test_temporary_flag() {
687 let mut config = DatabaseConfig::default();
688 config.temporary = false;
689 let db1 = DatabaseImpl::new(
690 DatabaseId::new(1),
691 "db1".to_string(),
692 DbType::User,
693 &config,
694 );
695 assert!(!db1.is_temporary());
696
697 config.temporary = true;
698 let db2 = DatabaseImpl::new(
699 DatabaseId::new(2),
700 "db2".to_string(),
701 DbType::User,
702 &config,
703 );
704 assert!(db2.is_temporary());
705 }
706
707 #[test]
708 fn test_key_prefixing_flag() {
709 let mut config = DatabaseConfig::default();
710 config.key_prefixing = false;
711 let db1 = DatabaseImpl::new(
712 DatabaseId::new(1),
713 "db1".to_string(),
714 DbType::User,
715 &config,
716 );
717 assert!(!db1.get_key_prefixing());
718
719 config.key_prefixing = true;
720 let db2 = DatabaseImpl::new(
721 DatabaseId::new(2),
722 "db2".to_string(),
723 DbType::User,
724 &config,
725 );
726 assert!(db2.get_key_prefixing());
727 }
728
729 #[test]
730 fn test_set_recovered_tree_preserves_key_prefixing() {
731 let mut config = DatabaseConfig::default();
737 config.key_prefixing = true;
738 let mut db = DatabaseImpl::new(
739 DatabaseId::new(7),
740 "kp_recover".to_string(),
741 DbType::User,
742 &config,
743 );
744 let recovered = Tree::new(7, 256);
746 assert!(!recovered.key_prefixing, "recovered tree starts false");
747 db.set_recovered_tree(recovered);
748 let t = db.get_real_tree_arc().expect("real tree");
750 assert!(
751 t.read().unwrap().key_prefixing,
752 "GAP-5: set_recovered_tree must preserve key_prefixing=true"
753 );
754 }
755
756 #[test]
757 fn test_delete_state_transitions() {
758 let config = make_config();
759 let mut db = DatabaseImpl::new(
760 DatabaseId::new(1),
761 "db".to_string(),
762 DbType::User,
763 &config,
764 );
765
766 assert!(!db.is_deleted());
767 assert!(!db.is_deleting());
768
769 db.start_delete();
770 assert!(!db.is_deleted());
771 assert!(db.is_deleting());
772
773 db.finish_delete();
774 assert!(db.is_deleted());
775 assert!(db.is_deleting());
776 }
777
778 #[test]
779 fn test_dirty_tracking() {
780 let config = make_config();
781 let db = DatabaseImpl::new(
782 DatabaseId::new(1),
783 "db".to_string(),
784 DbType::User,
785 &config,
786 );
787
788 assert!(!db.is_dirty());
789
790 db.set_dirty();
791 assert!(db.is_dirty());
792
793 db.clear_dirty();
794 assert!(!db.is_dirty());
795 }
796
797 #[test]
798 fn test_reference_counting() {
799 let config = make_config();
800 let db = DatabaseImpl::new(
801 DatabaseId::new(1),
802 "db".to_string(),
803 DbType::User,
804 &config,
805 );
806
807 assert_eq!(db.reference_count(), 0);
808
809 db.increment_reference_count();
810 assert_eq!(db.reference_count(), 1);
811
812 db.increment_reference_count();
813 assert_eq!(db.reference_count(), 2);
814
815 db.decrement_reference_count();
816 assert_eq!(db.reference_count(), 1);
817
818 db.decrement_reference_count();
819 assert_eq!(db.reference_count(), 0);
820 }
821
822 #[test]
823 fn test_serialization_round_trip() {
824 let mut config = DatabaseConfig::default();
825 config.sorted_duplicates = true;
826 config.key_prefixing = true;
827 config.node_max_entries = 256;
828
829 let db = DatabaseImpl::new(
830 DatabaseId::new(42),
831 "my_database".to_string(),
832 DbType::User,
833 &config,
834 );
835
836 let mut buf = Vec::new();
837 db.write_to_log(&mut buf).unwrap();
838
839 let db2 = DatabaseImpl::read_from_log(&buf).unwrap();
840
841 assert_eq!(db2.get_id(), DatabaseId::new(42));
842 assert_eq!(db2.get_name(), "my_database");
843 assert!(db2.get_sorted_duplicates());
844 assert!(db2.get_key_prefixing());
845 assert_eq!(db2.max_tree_entries_per_node(), 256);
846 }
847
848 #[test]
849 fn test_tree_access() {
850 let config = make_config();
851 let mut db = DatabaseImpl::new(
852 DatabaseId::new(1),
853 "db".to_string(),
854 DbType::User,
855 &config,
856 );
857
858 {
860 let tree = db.get_tree().unwrap();
861 assert_eq!(tree.get_root_lsn(), noxu_util::NULL_LSN.as_u64());
862 }
863
864 {
866 let tree = db.get_tree_mut().unwrap();
867 tree.set_root_lsn(12345);
868 }
869
870 {
872 let tree = db.get_tree().unwrap();
873 assert_eq!(tree.get_root_lsn(), 12345);
874 }
875 }
876
877 #[test]
878 fn test_log_size() {
879 let config = make_config();
880 let db = DatabaseImpl::new(
881 DatabaseId::new(1),
882 "test".to_string(),
883 DbType::User,
884 &config,
885 );
886
887 let expected_size = 8 + 4 + 4 + 1 + 4 + 8; assert_eq!(db.log_size(), expected_size);
889 }
890}