use crate::distance::DistanceMetric;
use crate::error::{Result, RetrieveError};
#[cfg(feature = "serde")]
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
#[cfg(feature = "serde")]
use std::io::{BufReader, BufWriter, Write};
#[cfg(feature = "serde")]
use std::path::Path;
#[cfg(feature = "serde")]
const LSM_FORMAT_VERSION: u32 = 1;
#[derive(Debug, Clone)]
pub struct LsmConfig {
pub dimension: usize,
pub buffer_capacity: usize,
pub size_ratio: usize,
pub max_levels: usize,
pub hnsw_m: usize,
pub hnsw_ef_construction: usize,
pub ef_search: usize,
pub distance_metric: DistanceMetric,
}
impl Default for LsmConfig {
fn default() -> Self {
Self {
dimension: 128,
buffer_capacity: 10_000,
size_ratio: 10,
max_levels: 5,
hnsw_m: 16,
hnsw_ef_construction: 200,
ef_search: 100,
distance_metric: DistanceMetric::Cosine,
}
}
}
#[derive(Debug)]
struct Level {
vectors: Vec<f32>,
doc_ids: Vec<u32>,
count: usize,
#[cfg(feature = "hnsw")]
hnsw: Option<crate::hnsw::HNSWIndex>,
}
#[cfg(feature = "serde")]
#[derive(Debug, Serialize, Deserialize)]
struct LsmManifest {
version: u32,
config: PersistedLsmConfig,
level_counts: Vec<usize>,
tombstone_count: usize,
total_inserts: u64,
total_deletes: u64,
total_compactions: u64,
}
#[cfg(feature = "serde")]
#[derive(Debug, Serialize, Deserialize)]
struct PersistedLsmConfig {
dimension: usize,
buffer_capacity: usize,
size_ratio: usize,
max_levels: usize,
hnsw_m: usize,
hnsw_ef_construction: usize,
ef_search: usize,
distance_metric: DistanceMetric,
}
#[cfg(feature = "serde")]
impl From<&LsmConfig> for PersistedLsmConfig {
fn from(config: &LsmConfig) -> Self {
Self {
dimension: config.dimension,
buffer_capacity: config.buffer_capacity,
size_ratio: config.size_ratio,
max_levels: config.max_levels,
hnsw_m: config.hnsw_m,
hnsw_ef_construction: config.hnsw_ef_construction,
ef_search: config.ef_search,
distance_metric: config.distance_metric,
}
}
}
#[cfg(feature = "serde")]
impl PersistedLsmConfig {
fn into_config(self) -> Result<LsmConfig> {
if self.dimension == 0 {
return Err(RetrieveError::FormatError(
"LSM manifest has zero dimension".into(),
));
}
if self.buffer_capacity == 0 {
return Err(RetrieveError::FormatError(
"LSM manifest has zero buffer_capacity".into(),
));
}
if self.size_ratio == 0 {
return Err(RetrieveError::FormatError(
"LSM manifest has zero size_ratio".into(),
));
}
if self.max_levels == 0 {
return Err(RetrieveError::FormatError(
"LSM manifest has zero max_levels".into(),
));
}
Ok(LsmConfig {
dimension: self.dimension,
buffer_capacity: self.buffer_capacity,
size_ratio: self.size_ratio,
max_levels: self.max_levels,
hnsw_m: self.hnsw_m,
hnsw_ef_construction: self.hnsw_ef_construction,
ef_search: self.ef_search,
distance_metric: self.distance_metric,
})
}
}
impl Level {
fn new() -> Self {
Self {
vectors: Vec::new(),
doc_ids: Vec::new(),
count: 0,
#[cfg(feature = "hnsw")]
hnsw: None,
}
}
fn is_empty(&self) -> bool {
self.count == 0
}
fn memory_usage(&self) -> crate::memory::MemoryReport {
let mut report = crate::memory::MemoryReport {
vectors_bytes: self.vectors.capacity() * std::mem::size_of::<f32>(),
graph_bytes: 0,
quantized_bytes: 0,
metadata_bytes: self.doc_ids.capacity() * std::mem::size_of::<u32>(),
};
#[cfg(feature = "hnsw")]
if let Some(hnsw) = &self.hnsw {
let hnsw_report = hnsw.memory_usage();
report.vectors_bytes += hnsw_report.vectors_bytes;
report.graph_bytes += hnsw_report.graph_bytes;
report.quantized_bytes += hnsw_report.quantized_bytes;
report.metadata_bytes += hnsw_report.metadata_bytes;
}
report
}
fn brute_force_search(
&self,
query: &[f32],
k: usize,
dimension: usize,
tombstones: &HashSet<u32>,
metric: DistanceMetric,
) -> Vec<(u32, f32)> {
let mut results: Vec<(u32, f32)> = (0..self.count)
.filter_map(|i| {
let doc_id = self.doc_ids[i];
if tombstones.contains(&doc_id) {
return None;
}
let start = i * dimension;
let vec = &self.vectors[start..start + dimension];
let dist = metric.distance(query, vec);
Some((doc_id, dist))
})
.collect();
results.sort_unstable_by(|a, b| a.1.total_cmp(&b.1));
results.truncate(k);
results
}
}
pub struct LsmIndex {
config: LsmConfig,
levels: Vec<Level>,
tombstones: HashSet<u32>,
total_inserts: u64,
total_deletes: u64,
total_compactions: u64,
}
impl LsmIndex {
pub fn new(config: LsmConfig) -> Self {
let mut levels = Vec::with_capacity(config.max_levels);
levels.push(Level::new()); Self {
config,
levels,
tombstones: HashSet::new(),
total_inserts: 0,
total_deletes: 0,
total_compactions: 0,
}
}
#[cfg(feature = "serde")]
pub fn save_to_dir(&self, output_dir: impl AsRef<Path>) -> Result<()> {
let output_dir = output_dir.as_ref();
std::fs::create_dir_all(output_dir)?;
self.validate_snapshot_shape()?;
let mut tombstones: Vec<u32> = self.tombstones.iter().copied().collect();
tombstones.sort_unstable();
let manifest = LsmManifest {
version: LSM_FORMAT_VERSION,
config: PersistedLsmConfig::from(&self.config),
level_counts: self.levels.iter().map(|level| level.count).collect(),
tombstone_count: tombstones.len(),
total_inserts: self.total_inserts,
total_deletes: self.total_deletes,
total_compactions: self.total_compactions,
};
write_json_atomic(&output_dir.join("manifest.json"), &manifest)?;
write_u32_atomic(&output_dir.join("tombstones.bin"), &tombstones)?;
for (level_idx, level) in self.levels.iter().enumerate() {
write_f32_atomic(
&output_dir.join(format!("level_{level_idx}_vectors.bin")),
&level.vectors,
)?;
write_u32_atomic(
&output_dir.join(format!("level_{level_idx}_doc_ids.bin")),
&level.doc_ids,
)?;
}
Ok(())
}
#[cfg(feature = "serde")]
pub fn load_from_dir(input_dir: impl AsRef<Path>) -> Result<Self> {
let input_dir = input_dir.as_ref();
let manifest: LsmManifest = read_json(&input_dir.join("manifest.json"))?;
if manifest.version != LSM_FORMAT_VERSION {
return Err(RetrieveError::FormatError(format!(
"unsupported LSM format version {}",
manifest.version
)));
}
if manifest.level_counts.is_empty() {
return Err(RetrieveError::FormatError(
"LSM manifest has no levels".into(),
));
}
let config = manifest.config.into_config()?;
if manifest.level_counts.len() > config.max_levels {
return Err(RetrieveError::FormatError(format!(
"LSM manifest level count {} exceeds max_levels {}",
manifest.level_counts.len(),
config.max_levels
)));
}
let tombstones =
read_u32_exact(&input_dir.join("tombstones.bin"), manifest.tombstone_count)?
.into_iter()
.collect();
let mut levels = Vec::with_capacity(manifest.level_counts.len());
for (level_idx, &count) in manifest.level_counts.iter().enumerate() {
let vector_len = count.checked_mul(config.dimension).ok_or_else(|| {
RetrieveError::FormatError(format!("LSM level {level_idx} vector length overflow"))
})?;
let vectors = read_f32_exact(
&input_dir.join(format!("level_{level_idx}_vectors.bin")),
vector_len,
)?;
let doc_ids = read_u32_exact(
&input_dir.join(format!("level_{level_idx}_doc_ids.bin")),
count,
)?;
levels.push(Self::level_from_snapshot(
&config, level_idx, vectors, doc_ids,
)?);
}
Ok(Self {
config,
levels,
tombstones,
total_inserts: manifest.total_inserts,
total_deletes: manifest.total_deletes,
total_compactions: manifest.total_compactions,
})
}
#[cfg(feature = "serde")]
fn validate_snapshot_shape(&self) -> Result<()> {
let dim = self.config.dimension;
if dim == 0 {
return Err(RetrieveError::InvalidParameter(
"cannot save LSM index with zero dimension".into(),
));
}
if self.levels.is_empty() {
return Err(RetrieveError::InvalidParameter(
"cannot save LSM index with no levels".into(),
));
}
if self.levels.len() > self.config.max_levels {
return Err(RetrieveError::InvalidParameter(format!(
"LSM level count {} exceeds max_levels {}",
self.levels.len(),
self.config.max_levels
)));
}
for (level_idx, level) in self.levels.iter().enumerate() {
let expected_vectors = level.count.checked_mul(dim).ok_or_else(|| {
RetrieveError::InvalidParameter(format!(
"LSM level {level_idx} vector length overflow"
))
})?;
if level.vectors.len() != expected_vectors {
return Err(RetrieveError::InvalidParameter(format!(
"LSM level {level_idx} has {} vector scalars, expected {}",
level.vectors.len(),
expected_vectors
)));
}
if level.doc_ids.len() != level.count {
return Err(RetrieveError::InvalidParameter(format!(
"LSM level {level_idx} has {} doc ids, expected {}",
level.doc_ids.len(),
level.count
)));
}
}
Ok(())
}
#[cfg(feature = "serde")]
fn level_from_snapshot(
config: &LsmConfig,
level_idx: usize,
vectors: Vec<f32>,
doc_ids: Vec<u32>,
) -> Result<Level> {
let count = doc_ids.len();
let expected_vectors = count.checked_mul(config.dimension).ok_or_else(|| {
RetrieveError::FormatError(format!("LSM level {level_idx} vector length overflow"))
})?;
if vectors.len() != expected_vectors {
return Err(RetrieveError::FormatError(format!(
"LSM level {level_idx} has {} vector scalars, expected {}",
vectors.len(),
expected_vectors
)));
}
#[cfg(feature = "hnsw")]
let hnsw = if level_idx > 0 && count > 0 {
let mut hnsw = crate::hnsw::HNSWIndex::builder(config.dimension)
.m(config.hnsw_m)
.ef_construction(config.hnsw_ef_construction)
.metric(config.distance_metric)
.auto_normalize(false)
.build()?;
for (i, &doc_id) in doc_ids.iter().enumerate() {
let start = i * config.dimension;
hnsw.add_slice(doc_id, &vectors[start..start + config.dimension])?;
}
hnsw.build()?;
Some(hnsw)
} else {
None
};
Ok(Level {
vectors,
doc_ids,
count,
#[cfg(feature = "hnsw")]
hnsw,
})
}
pub fn insert(&mut self, doc_id: u32, vector: Vec<f32>) -> Result<()> {
self.insert_slice(doc_id, &vector)
}
pub fn insert_slice(&mut self, doc_id: u32, vector: &[f32]) -> Result<()> {
if vector.len() != self.config.dimension {
return Err(RetrieveError::DimensionMismatch {
query_dim: vector.len(),
doc_dim: self.config.dimension,
});
}
self.tombstones.remove(&doc_id);
let stored = self.prepare_vector(vector);
self.levels[0].vectors.extend_from_slice(&stored);
self.levels[0].doc_ids.push(doc_id);
self.levels[0].count += 1;
self.total_inserts += 1;
if self.levels[0].count >= self.config.buffer_capacity {
self.compact()?;
}
Ok(())
}
pub fn delete(&mut self, doc_id: u32) {
self.tombstones.insert(doc_id);
self.total_deletes += 1;
}
fn prepare_vector(&self, vector: &[f32]) -> Vec<f32> {
if self.config.distance_metric != DistanceMetric::Cosine {
return vector.to_vec();
}
let norm: f32 = vector.iter().map(|x| x * x).sum::<f32>().sqrt();
if norm > 1e-10 {
vector.iter().map(|x| x / norm).collect()
} else {
vector.to_vec()
}
}
pub fn search(&self, query: &[f32], k: usize) -> Result<Vec<(u32, f32)>> {
self.search_with_ef(query, k, self.config.ef_search)
}
pub fn search_with_ef(
&self,
query: &[f32],
k: usize,
ef_search: usize,
) -> Result<Vec<(u32, f32)>> {
if query.len() != self.config.dimension {
return Err(RetrieveError::DimensionMismatch {
query_dim: query.len(),
doc_dim: self.config.dimension,
});
}
let query_normalized = self.prepare_vector(query);
let query = query_normalized.as_slice();
let mut all_results: Vec<(u32, f32)> = Vec::new();
for (level_idx, level) in self.levels.iter().enumerate() {
if level.is_empty() {
continue;
}
let level_results = if level_idx == 0 {
level.brute_force_search(
query,
k,
self.config.dimension,
&self.tombstones,
self.config.distance_metric,
)
} else {
#[cfg(feature = "hnsw")]
{
if let Some(ref hnsw) = level.hnsw {
let ef = ef_search.max(k);
match hnsw.search(query, k, ef) {
Ok(results) => results
.into_iter()
.filter(|(id, _)| !self.tombstones.contains(id))
.collect(),
Err(_) => Vec::new(),
}
} else {
level.brute_force_search(
query,
k,
self.config.dimension,
&self.tombstones,
self.config.distance_metric,
)
}
}
#[cfg(not(feature = "hnsw"))]
{
level.brute_force_search(
query,
k,
self.config.dimension,
&self.tombstones,
self.config.distance_metric,
)
}
};
all_results.extend(level_results);
}
let mut seen = HashSet::new();
all_results.sort_unstable_by(|a, b| a.1.total_cmp(&b.1));
all_results.retain(|(id, _)| seen.insert(*id));
all_results.truncate(k);
Ok(all_results)
}
pub fn compact(&mut self) -> Result<()> {
if self.levels[0].is_empty() {
return Ok(());
}
let l0 = std::mem::replace(&mut self.levels[0], Level::new());
if self.levels.len() < 2 {
self.levels.push(Level::new());
}
self.merge_into_level(l0, 1)?;
self.cascade_compact(1)?;
self.total_compactions += 1;
Ok(())
}
fn merge_into_level(&mut self, source: Level, target_idx: usize) -> Result<()> {
let dim = self.config.dimension;
let mut merged_vectors: Vec<f32> = Vec::new();
let mut merged_ids: Vec<u32> = Vec::new();
if target_idx < self.levels.len() {
let target = &self.levels[target_idx];
for i in 0..target.count {
let doc_id = target.doc_ids[i];
if !self.tombstones.contains(&doc_id) {
let start = i * dim;
merged_vectors.extend_from_slice(&target.vectors[start..start + dim]);
merged_ids.push(doc_id);
}
}
}
for i in 0..source.count {
let doc_id = source.doc_ids[i];
if !self.tombstones.contains(&doc_id) {
let start = i * dim;
merged_vectors.extend_from_slice(&source.vectors[start..start + dim]);
merged_ids.push(doc_id);
}
}
let merged_count = merged_ids.len();
#[cfg(feature = "hnsw")]
let hnsw = if merged_count > 0 {
let mut hnsw = crate::hnsw::HNSWIndex::builder(dim)
.m(self.config.hnsw_m)
.ef_construction(self.config.hnsw_ef_construction)
.metric(self.config.distance_metric)
.auto_normalize(false) .build()?;
for (i, &doc_id) in merged_ids.iter().enumerate() {
let start = i * dim;
hnsw.add_slice(doc_id, &merged_vectors[start..start + dim])?;
}
hnsw.build()?;
Some(hnsw)
} else {
None
};
while self.levels.len() <= target_idx {
self.levels.push(Level::new());
}
self.levels[target_idx] = Level {
vectors: merged_vectors,
doc_ids: merged_ids,
count: merged_count,
#[cfg(feature = "hnsw")]
hnsw,
};
Ok(())
}
fn cascade_compact(&mut self, level_idx: usize) -> Result<()> {
if level_idx >= self.config.max_levels - 1 {
return Ok(()); }
let level_size = self.levels.get(level_idx).map_or(0, |l| l.count);
let next_size = self.levels.get(level_idx + 1).map_or(0, |l| l.count);
let should_compact = if next_size == 0 {
level_size >= self.config.buffer_capacity * self.config.size_ratio
} else {
level_size >= self.config.size_ratio * next_size
};
if should_compact {
let source = std::mem::replace(&mut self.levels[level_idx], Level::new());
while self.levels.len() <= level_idx + 1 {
self.levels.push(Level::new());
}
self.merge_into_level(source, level_idx + 1)?;
self.cascade_compact(level_idx + 1)?;
}
Ok(())
}
pub fn len(&self) -> usize {
let total: usize = self.levels.iter().map(|l| l.count).sum();
total.saturating_sub(self.tombstones.len())
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn num_levels(&self) -> usize {
self.levels.len()
}
pub fn level_sizes(&self) -> Vec<usize> {
self.levels.iter().map(|l| l.count).collect()
}
pub fn stats(&self) -> LsmStats {
LsmStats {
total_inserts: self.total_inserts,
total_deletes: self.total_deletes,
total_compactions: self.total_compactions,
num_levels: self.levels.len(),
level_sizes: self.level_sizes(),
tombstone_count: self.tombstones.len(),
}
}
pub fn memory_usage(&self) -> crate::memory::MemoryReport {
let mut report = crate::memory::MemoryReport {
vectors_bytes: 0,
graph_bytes: 0,
quantized_bytes: 0,
metadata_bytes: self.levels.capacity() * std::mem::size_of::<Level>()
+ self.tombstones.capacity() * std::mem::size_of::<u32>(),
};
for level in &self.levels {
let level_report = level.memory_usage();
report.vectors_bytes += level_report.vectors_bytes;
report.graph_bytes += level_report.graph_bytes;
report.quantized_bytes += level_report.quantized_bytes;
report.metadata_bytes += level_report.metadata_bytes;
}
report
}
pub fn force_merge_all(&mut self) -> Result<()> {
if self.levels.is_empty() {
return Ok(());
}
let dim = self.config.dimension;
let mut all_vectors: Vec<f32> = Vec::new();
let mut all_ids: Vec<u32> = Vec::new();
for level in &self.levels {
for i in 0..level.count {
let doc_id = level.doc_ids[i];
if !self.tombstones.contains(&doc_id) {
let start = i * dim;
all_vectors.extend_from_slice(&level.vectors[start..start + dim]);
all_ids.push(doc_id);
}
}
}
self.levels.clear();
self.levels.push(Level::new());
let count = all_ids.len();
if count == 0 {
return Ok(());
}
#[cfg(feature = "hnsw")]
let hnsw = {
let mut hnsw = crate::hnsw::HNSWIndex::builder(dim)
.m(self.config.hnsw_m)
.ef_construction(self.config.hnsw_ef_construction)
.metric(self.config.distance_metric)
.auto_normalize(false)
.build()?;
for (i, &doc_id) in all_ids.iter().enumerate() {
let start = i * dim;
hnsw.add_slice(doc_id, &all_vectors[start..start + dim])?;
}
hnsw.build()?;
Some(hnsw)
};
self.levels.push(Level {
vectors: all_vectors,
doc_ids: all_ids,
count,
#[cfg(feature = "hnsw")]
hnsw,
});
self.tombstones.clear();
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct LsmStats {
pub total_inserts: u64,
pub total_deletes: u64,
pub total_compactions: u64,
pub num_levels: usize,
pub level_sizes: Vec<usize>,
pub tombstone_count: usize,
}
#[cfg(feature = "serde")]
fn write_json_atomic<T: Serialize>(path: &Path, value: &T) -> Result<()> {
write_atomic(path, |writer| {
serde_json::to_writer_pretty(writer, value)
.map_err(|e| std::io::Error::other(e.to_string()))
})
}
#[cfg(feature = "serde")]
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<T> {
let file = std::fs::File::open(path)?;
serde_json::from_reader(BufReader::new(file))
.map_err(|e| RetrieveError::FormatError(e.to_string()))
}
#[cfg(feature = "serde")]
fn write_f32_atomic(path: &Path, values: &[f32]) -> Result<()> {
write_atomic(path, |writer| {
for value in values {
writer.write_all(&value.to_le_bytes())?;
}
Ok(())
})
}
#[cfg(feature = "serde")]
fn read_f32_exact(path: &Path, expected_len: usize) -> Result<Vec<f32>> {
let bytes = std::fs::read(path)?;
let expected_bytes = expected_len
.checked_mul(std::mem::size_of::<f32>())
.ok_or_else(|| RetrieveError::FormatError("f32 byte length overflow".into()))?;
if bytes.len() != expected_bytes {
return Err(RetrieveError::FormatError(format!(
"{} size mismatch: expected {} bytes, got {}",
path.display(),
expected_bytes,
bytes.len()
)));
}
Ok(bytes
.chunks_exact(4)
.map(|chunk| f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]))
.collect())
}
#[cfg(feature = "serde")]
fn write_u32_atomic(path: &Path, values: &[u32]) -> Result<()> {
write_atomic(path, |writer| {
for value in values {
writer.write_all(&value.to_le_bytes())?;
}
Ok(())
})
}
#[cfg(feature = "serde")]
fn read_u32_exact(path: &Path, expected_len: usize) -> Result<Vec<u32>> {
let bytes = std::fs::read(path)?;
let expected_bytes = expected_len
.checked_mul(std::mem::size_of::<u32>())
.ok_or_else(|| RetrieveError::FormatError("u32 byte length overflow".into()))?;
if bytes.len() != expected_bytes {
return Err(RetrieveError::FormatError(format!(
"{} size mismatch: expected {} bytes, got {}",
path.display(),
expected_bytes,
bytes.len()
)));
}
Ok(bytes
.chunks_exact(4)
.map(|chunk| u32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]))
.collect())
}
#[cfg(feature = "serde")]
fn write_atomic(
path: &Path,
write: impl FnOnce(&mut BufWriter<std::fs::File>) -> std::io::Result<()>,
) -> Result<()> {
let tmp_path = path.with_extension("tmp");
{
let file = std::fs::File::create(&tmp_path)?;
let mut writer = BufWriter::new(file);
write(&mut writer)?;
writer.flush()?;
}
std::fs::rename(&tmp_path, path)?;
Ok(())
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use super::*;
fn make_config(dim: usize) -> LsmConfig {
LsmConfig {
dimension: dim,
buffer_capacity: 20,
size_ratio: 5,
max_levels: 4,
hnsw_m: 8,
hnsw_ef_construction: 50,
ef_search: 50,
distance_metric: DistanceMetric::L2,
}
}
fn make_l2_config(dim: usize, buffer_capacity: usize) -> LsmConfig {
LsmConfig {
buffer_capacity,
..make_config(dim)
}
}
fn make_vector(dim: usize, seed: u32) -> Vec<f32> {
(0..dim)
.map(|i| (seed as f32 * 0.1 + i as f32 * 0.01).sin())
.collect()
}
#[test]
fn insert_and_search_l0() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..10u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
assert_eq!(index.level_sizes(), vec![10]);
let results = index.search(&make_vector(8, 0), 3).unwrap();
assert!(!results.is_empty());
assert_eq!(results[0].0, 0); }
#[test]
fn l2_l0_preserves_vector_magnitude() {
let mut index = LsmIndex::new(make_l2_config(2, 10));
index.insert(0, vec![1.0, 0.0]).unwrap();
index.insert(1, vec![2.0, 0.0]).unwrap();
let results = index.search(&[2.0, 0.0], 2).unwrap();
assert_eq!(results[0].0, 1);
}
#[test]
fn l2_compacted_level_preserves_vector_magnitude() {
let mut index = LsmIndex::new(make_l2_config(2, 2));
index.insert(0, vec![1.0, 0.0]).unwrap();
index.insert(1, vec![2.0, 0.0]).unwrap();
assert!(
index.level_sizes().len() >= 2,
"expected compaction into L1"
);
let results = index.search_with_ef(&[2.0, 0.0], 2, 20).unwrap();
assert_eq!(results[0].0, 1);
}
#[test]
fn compaction_moves_to_l1() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..25u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
let sizes = index.level_sizes();
assert!(sizes.len() >= 2, "expected at least 2 levels: {sizes:?}");
assert!(
sizes[1] > 0,
"L1 should have vectors after compaction: {sizes:?}"
);
let results = index.search(&make_vector(8, 5), 3).unwrap();
assert!(!results.is_empty());
}
#[test]
fn delete_filters_from_search() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..10u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
index.delete(0);
index.delete(1);
let results = index.search(&make_vector(8, 0), 10).unwrap();
for (id, _) in &results {
assert!(*id != 0 && *id != 1, "deleted ID {id} in results");
}
}
#[test]
fn delete_survives_compaction() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..25u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
index.delete(5);
for i in 25..50u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
let results = index.search(&make_vector(8, 5), 50).unwrap();
for (id, _) in &results {
assert_ne!(*id, 5, "deleted ID 5 in results after compaction");
}
}
#[test]
fn force_merge_all() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..50u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
index.delete(10);
index.force_merge_all().unwrap();
let sizes = index.level_sizes();
assert_eq!(sizes[0], 0, "L0 should be empty after merge");
assert_eq!(sizes[1], 49, "L1 should have 49 vectors (50 - 1 deleted)");
assert_eq!(index.stats().tombstone_count, 0);
let results = index.search(&make_vector(8, 0), 3).unwrap();
assert!(!results.is_empty());
}
#[test]
fn reinsert_after_delete() {
let mut index = LsmIndex::new(make_config(8));
index.insert(0, make_vector(8, 0)).unwrap();
index.delete(0);
index.insert(0, make_vector(8, 100)).unwrap();
let results = index.search(&make_vector(8, 100), 1).unwrap();
assert_eq!(results[0].0, 0);
}
#[cfg(feature = "serde")]
#[test]
fn snapshot_roundtrip_preserves_levels_tombstones_and_counters() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..35u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
index.delete(5);
index.delete(7);
let before_sizes = index.level_sizes();
let before_stats = index.stats();
let dir = tempfile::tempdir().unwrap();
index.save_to_dir(dir.path()).unwrap();
let loaded = LsmIndex::load_from_dir(dir.path()).unwrap();
assert_eq!(loaded.level_sizes(), before_sizes);
assert_eq!(loaded.stats().total_inserts, before_stats.total_inserts);
assert_eq!(loaded.stats().total_deletes, before_stats.total_deletes);
assert_eq!(
loaded.stats().total_compactions,
before_stats.total_compactions
);
assert_eq!(loaded.stats().tombstone_count, 2);
let results = loaded.search(&make_vector(8, 5), 20).unwrap();
for (id, _) in results {
assert_ne!(id, 5);
assert_ne!(id, 7);
}
}
#[cfg(feature = "serde")]
#[test]
fn load_rejects_future_snapshot_version() {
let mut index = LsmIndex::new(make_config(4));
index.insert(0, make_vector(4, 0)).unwrap();
let dir = tempfile::tempdir().unwrap();
index.save_to_dir(dir.path()).unwrap();
let manifest_path = dir.path().join("manifest.json");
let mut manifest: serde_json::Value =
serde_json::from_slice(&std::fs::read(&manifest_path).unwrap()).unwrap();
manifest["version"] = serde_json::json!(LSM_FORMAT_VERSION + 1);
std::fs::write(
&manifest_path,
serde_json::to_vec_pretty(&manifest).unwrap(),
)
.unwrap();
let err = match LsmIndex::load_from_dir(dir.path()) {
Ok(_) => panic!("future LSM snapshot version should be rejected"),
Err(err) => err,
};
match err {
RetrieveError::FormatError(message) => {
assert!(message.contains("unsupported LSM format version"));
}
other => panic!("expected format error, got {other:?}"),
}
}
#[test]
fn empty_search() {
let index = LsmIndex::new(make_config(8));
let results = index.search(&make_vector(8, 0), 5).unwrap();
assert!(results.is_empty());
}
#[test]
fn stats_tracking() {
let mut index = LsmIndex::new(make_config(8));
for i in 0..30u32 {
index.insert(i, make_vector(8, i)).unwrap();
}
index.delete(0);
let stats = index.stats();
assert_eq!(stats.total_inserts, 30);
assert_eq!(stats.total_deletes, 1);
assert!(stats.total_compactions >= 1);
assert_eq!(stats.tombstone_count, 1);
}
}