reifydb_store_multi/store/
drop.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::{common::CommitVersion, encoded::key::EncodedKey, interface::store::EntryKind};
5
6use crate::{Result, tier::TierStorage};
7
8#[derive(Debug, Clone)]
9pub struct DropEntry {
10	pub key: EncodedKey,
11
12	pub version: CommitVersion,
13
14	pub value_bytes: u64,
15}
16
17pub(crate) fn find_keys_to_drop<S: TierStorage>(
18	storage: &S,
19	table: EntryKind,
20	key: &[u8],
21	pending_version: Option<CommitVersion>,
22) -> Result<Vec<DropEntry>> {
23	let all_versions = storage.get_all_versions(table, key)?;
24
25	let mut versioned_entries: Vec<(CommitVersion, u64)> = all_versions
26		.into_iter()
27		.map(|(version, value)| {
28			let value_bytes = value.as_ref().map(|v| v.len() as u64).unwrap_or(0);
29			(version, value_bytes)
30		})
31		.collect();
32
33	if let Some(pending_ver) = pending_version
34		&& !versioned_entries.iter().any(|(v, _)| *v == pending_ver)
35	{
36		versioned_entries.push((pending_ver, 0));
37	}
38
39	versioned_entries.sort_by(|a, b| b.0.cmp(&a.0));
40
41	let mut entries_to_drop = Vec::with_capacity(versioned_entries.len().saturating_sub(1));
42	let drop_key = EncodedKey::new(key.to_vec());
43
44	for (idx, (entry_version, value_bytes)) in versioned_entries.into_iter().enumerate() {
45		let should_drop = idx > 0;
46
47		if should_drop {
48			if Some(entry_version) == pending_version {
49				continue;
50			}
51
52			entries_to_drop.push(DropEntry {
53				key: drop_key.clone(),
54				version: entry_version,
55				value_bytes,
56			});
57		}
58	}
59
60	Ok(entries_to_drop)
61}
62
63#[cfg(test)]
64pub mod tests {
65	use std::collections::HashMap;
66
67	use reifydb_value::util::cowvec::CowVec;
68
69	use super::*;
70	use crate::tier::commit::buffer::MultiCommitBufferTier;
71
72	/// Create versioned test entries for a key
73	fn setup_versioned_entries(storage: &MultiCommitBufferTier, table: EntryKind, key: &[u8], versions: &[u64]) {
74		for v in versions {
75			let entries = vec![(EncodedKey::new(key.to_vec()), Some(CowVec::new(vec![*v as u8])))];
76			storage.set(CommitVersion(*v), HashMap::from([(table, entries)])).unwrap();
77		}
78	}
79
80	/// Extract version numbers from the drop entries
81	fn extract_dropped_versions(entries: &[DropEntry]) -> Vec<u64> {
82		entries.iter().map(|e| e.version.0).collect()
83	}
84
85	#[test]
86	fn test_drop_historical_versions() {
87		let storage = MultiCommitBufferTier::memory();
88		let table = EntryKind::Multi;
89		let key = b"test_key";
90
91		// Versions: 1, 5, 10, 20, 100
92		setup_versioned_entries(&storage, table, key, &[1, 5, 10, 20, 100]);
93
94		// Should drop all except 100
95		let to_drop = find_keys_to_drop(&storage, table, key, None).unwrap();
96
97		assert_eq!(to_drop.len(), 4);
98		let versions = extract_dropped_versions(&to_drop);
99		assert!(versions.contains(&1));
100		assert!(versions.contains(&5));
101		assert!(versions.contains(&10));
102		assert!(versions.contains(&20));
103		assert!(!versions.contains(&100));
104	}
105
106	#[test]
107	fn test_keep_latest_with_pending() {
108		let storage = MultiCommitBufferTier::memory();
109		let table = EntryKind::Multi;
110		let key = b"test_key";
111
112		// Existing: 1, 5, 10. Pending: 20.
113		setup_versioned_entries(&storage, table, key, &[1, 5, 10]);
114
115		// Should keep 20 (pending) and drop 1, 5, 10
116		let to_drop = find_keys_to_drop(&storage, table, key, Some(CommitVersion(20))).unwrap();
117
118		assert_eq!(to_drop.len(), 3);
119		let versions = extract_dropped_versions(&to_drop);
120		assert!(versions.contains(&1));
121		assert!(versions.contains(&5));
122		assert!(versions.contains(&10));
123		assert!(!versions.contains(&20));
124	}
125
126	#[test]
127	fn test_single_version_no_drop() {
128		let storage = MultiCommitBufferTier::memory();
129		let table = EntryKind::Multi;
130		let key = b"test_key";
131
132		setup_versioned_entries(&storage, table, key, &[42]);
133
134		// Only one version exists, should drop nothing
135		let to_drop = find_keys_to_drop(&storage, table, key, None).unwrap();
136		assert!(to_drop.is_empty());
137	}
138
139	#[test]
140	fn test_empty_storage() {
141		let storage = MultiCommitBufferTier::memory();
142		let table = EntryKind::Multi;
143		let key = b"nonexistent";
144
145		let to_drop = find_keys_to_drop(&storage, table, key, None).unwrap();
146		assert!(to_drop.is_empty());
147	}
148}
reifydb_store_multi/store/drop.rs

reifydb_store_multi/store/
drop.rs
