reifydb_cdc/storage/
memory.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::{BTreeMap, Bound},
6	sync::Arc,
7};
8
9use reifydb_core::{
10	common::CommitVersion,
11	interface::cdc::{Cdc, CdcBatch},
12};
13use reifydb_runtime::sync::rwlock::RwLock;
14
15use super::{CdcStorage, CdcStorageResult, DropBeforeResult, DroppedCdcEntry, normalize_range_inclusive};
16
17#[derive(Clone)]
18pub struct MemoryCdcStorage {
19	inner: Arc<RwLock<BTreeMap<CommitVersion, Cdc>>>,
20}
21
22impl MemoryCdcStorage {
23	pub fn new() -> Self {
24		Self {
25			inner: Arc::new(RwLock::new(BTreeMap::new())),
26		}
27	}
28
29	pub fn with_entries(entries: impl IntoIterator<Item = Cdc>) -> Self {
30		let map: BTreeMap<CommitVersion, Cdc> = entries.into_iter().map(|cdc| (cdc.version, cdc)).collect();
31		Self {
32			inner: Arc::new(RwLock::new(map)),
33		}
34	}
35
36	pub fn len(&self) -> usize {
37		self.inner.read().len()
38	}
39
40	pub fn is_empty(&self) -> bool {
41		self.inner.read().is_empty()
42	}
43
44	pub fn clear(&self) {
45		self.inner.write().clear();
46	}
47}
48
49impl Default for MemoryCdcStorage {
50	fn default() -> Self {
51		Self::new()
52	}
53}
54
55impl CdcStorage for MemoryCdcStorage {
56	fn write(&self, cdc: &Cdc) -> CdcStorageResult<()> {
57		self.inner.write().insert(cdc.version, cdc.clone());
58		Ok(())
59	}
60
61	fn read(&self, version: CommitVersion) -> CdcStorageResult<Option<Cdc>> {
62		Ok(self.inner.read().get(&version).cloned())
63	}
64
65	fn read_range(
66		&self,
67		start: Bound<CommitVersion>,
68		end: Bound<CommitVersion>,
69		batch_size: u64,
70	) -> CdcStorageResult<CdcBatch> {
71		let Some((lo_inc, hi_inc)) = normalize_range_inclusive(start, end) else {
72			return Ok(CdcBatch {
73				items: Vec::new(),
74				has_more: false,
75			});
76		};
77		let guard = self.inner.read();
78		let (items, has_more) = collect_range_into(&guard, lo_inc, hi_inc, batch_size as usize);
79		Ok(CdcBatch {
80			items,
81			has_more,
82		})
83	}
84
85	fn count(&self, version: CommitVersion) -> CdcStorageResult<usize> {
86		Ok(self.inner.read().get(&version).map(|cdc| cdc.system_changes.len()).unwrap_or(0))
87	}
88
89	fn min_version(&self) -> CdcStorageResult<Option<CommitVersion>> {
90		Ok(self.inner.read().keys().next().copied())
91	}
92
93	fn max_version(&self) -> CdcStorageResult<Option<CommitVersion>> {
94		Ok(self.inner.read().keys().next_back().copied())
95	}
96
97	fn drop_before(&self, version: CommitVersion, limit: usize) -> CdcStorageResult<DropBeforeResult> {
98		let mut guard = self.inner.write();
99		let mut keys_to_remove: Vec<_> = guard.range(..version).take(limit).map(|(k, _)| *k).collect();
100		let more_remaining = keys_to_remove.len() == limit && guard.range(..version).nth(limit).is_some();
101		let count = keys_to_remove.len();
102		let entries = collect_dropped_entries(&guard, &keys_to_remove);
103		for key in keys_to_remove.drain(..) {
104			guard.remove(&key);
105		}
106		Ok(DropBeforeResult {
107			count,
108			entries,
109			more_remaining,
110		})
111	}
112}
113
114#[inline]
115fn collect_range_into(
116	guard: &BTreeMap<CommitVersion, Cdc>,
117	lo_inc: CommitVersion,
118	hi_inc: CommitVersion,
119	batch_size: usize,
120) -> (Vec<Cdc>, bool) {
121	let mut items: Vec<Cdc> = Vec::with_capacity(batch_size.min(64));
122	for (count, (_, cdc)) in guard.range(lo_inc..=hi_inc).enumerate() {
123		if count >= batch_size {
124			return (items, true);
125		}
126		items.push(cdc.clone());
127	}
128	(items, false)
129}
130
131#[inline]
132fn collect_dropped_entries(guard: &BTreeMap<CommitVersion, Cdc>, keys: &[CommitVersion]) -> Vec<DroppedCdcEntry> {
133	let mut entries = Vec::new();
134	for key in keys {
135		if let Some(cdc) = guard.get(key) {
136			for sys_change in &cdc.system_changes {
137				entries.push(DroppedCdcEntry {
138					key: sys_change.key().clone(),
139					value_bytes: sys_change.value_bytes() as u64,
140				});
141			}
142		}
143	}
144	entries
145}
reifydb_cdc/storage/memory.rs

reifydb_cdc/storage/
memory.rs
