Skip to content

Commit 8e6edbd

Browse files
Implement snapshots (#67)
* add(snapshots): implement index snapshots for kd tree and flat index * add(snapshots): implement snapshots for hnsw index * format code * remove repr(packed) because bincode already packs structs * fix(snapshots): remove unwraps and handle errors * refactor(snapshots): improve error handling in worker thread * fix merge conflicts * add(snapshot): add architectural details in README.md
1 parent 1154cda commit 8e6edbd

44 files changed

Lines changed: 3468 additions & 600 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Cargo.lock‎

Lines changed: 676 additions & 277 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ members = [
88
"crates/http",
99
"crates/tui",
1010
"crates/grpc",
11+
"crates/snapshot",
1112
]
1213

1314
[workspace.package]
@@ -51,5 +52,6 @@ grpc = { path = "crates/grpc" }
5152
http = { path = "crates/http" }
5253
index = { path = "crates/index" }
5354
server = { path = "crates/server" }
55+
snapshot = { path = "crates/snapshot" }
5456
storage = { path = "crates/storage" }
5557
tui = { path = "crates/tui" }

‎crates/api/Cargo.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ license.workspace = true
1010
defs.workspace = true
1111
index.workspace = true
1212
snafu.workspace = true
13+
snapshot.workspace = true
1314
storage.workspace = true
1415
tempfile.workspace = true
1516
uuid.workspace = true

‎crates/api/src/lib.rs‎

Lines changed: 198 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
1-
use defs::{Dimension, IndexedVector, Similarity};
2-
1+
use defs::{DbError, Dimension, IndexedVector, Similarity, SnapshottableDb};
32
use defs::{DenseVector, Payload, Point, PointId};
43
use index::hnsw::HnswIndex;
5-
use std::path::PathBuf;
4+
use index::kd_tree::index::KDTree;
5+
use std::path::{Path, PathBuf};
6+
use tempfile::tempdir;
67
// use std::sync::atomic::{AtomicU64, Ordering};
78
use std::sync::{Arc, RwLock};
89

9-
use index::flat::FlatIndex;
10+
use index::flat::index::FlatIndex;
1011
use index::{IndexType, VectorIndex};
12+
use snapshot::Snapshot;
1113
use storage::rocks_db::RocksDbStorage;
1214
use storage::{StorageEngine, StorageType, VectorPage};
1315

@@ -152,6 +154,33 @@ impl VectorDb {
152154
}
153155
}
154156

157+
impl SnapshottableDb for VectorDb {
158+
fn create_snapshot(&self, dir_path: &Path) -> Result<PathBuf, DbError> {
159+
if !dir_path.is_dir() {
160+
return Err(DbError::SnapshotError(format!(
161+
"Invalid path: {}",
162+
dir_path.display()
163+
)));
164+
}
165+
166+
let index_snapshot = self
167+
.index
168+
.read()
169+
.map_err(|_| DbError::LockError)?
170+
.snapshot()?;
171+
172+
let tempdir = tempdir().unwrap();
173+
let storage_checkpoint = self.storage.checkpoint_at(tempdir.path()).map_err(|e| {
174+
DbError::StorageCheckpointError(format!("Could not create storage checkpoint: {e}"))
175+
})?;
176+
177+
let snapshot = Snapshot::new(index_snapshot, storage_checkpoint, self.dimension)?;
178+
let snapshot_path = snapshot.save(dir_path)?;
179+
180+
Ok(snapshot_path)
181+
}
182+
}
183+
155184
#[derive(Debug)]
156185
pub struct DbConfig {
157186
pub storage_type: StorageType,
@@ -161,6 +190,28 @@ pub struct DbConfig {
161190
pub similarity: Similarity,
162191
}
163192

193+
#[derive(Debug)]
194+
pub struct DbRestoreConfig {
195+
pub data_path: PathBuf,
196+
pub snapshot_path: PathBuf,
197+
}
198+
199+
impl DbRestoreConfig {
200+
pub fn new(data_path: PathBuf, snapshot_path: PathBuf) -> Self {
201+
Self {
202+
data_path,
203+
snapshot_path,
204+
}
205+
}
206+
}
207+
208+
pub fn restore_from_snapshot(config: &DbRestoreConfig) -> Result<VectorDb, DbError> {
209+
// restore the index from the snapshot
210+
let (storage_engine, index, dimensions) =
211+
Snapshot::load(&config.snapshot_path, &config.data_path)?;
212+
Ok(VectorDb::_new(storage_engine, index, dimensions))
213+
}
214+
164215
pub fn init_api(config: DbConfig) -> Result<VectorDb> {
165216
// Initialize the storage engine
166217
let storage = match config.storage_type {
@@ -171,11 +222,11 @@ pub fn init_api(config: DbConfig) -> Result<VectorDb> {
171222
// Initialize the vector index
172223
let index: Arc<RwLock<dyn VectorIndex>> = match config.index_type {
173224
IndexType::Flat => Arc::new(RwLock::new(FlatIndex::new())),
225+
IndexType::KDTree => Arc::new(RwLock::new(KDTree::build_empty(config.dimension))),
174226
IndexType::HNSW => Arc::new(RwLock::new(HnswIndex::new(
175227
config.similarity,
176228
config.dimension,
177229
))),
178-
_ => Arc::new(RwLock::new(FlatIndex::new())),
179230
};
180231

181232
// Init the db
@@ -192,8 +243,11 @@ mod tests {
192243

193244
// TODO: Add more exhaustive tests
194245

246+
use std::sync::Mutex;
247+
195248
use super::*;
196249
use defs::ContentType;
250+
use snapshot::{engine::SnapshotEngine, registry::local::LocalRegistry};
197251
use tempfile::{TempDir, tempdir};
198252

199253
// Helper function to create a test database
@@ -419,4 +473,143 @@ mod tests {
419473
let inserted = db.build_index().unwrap();
420474
assert_eq!(inserted, 10);
421475
}
476+
477+
#[test]
478+
fn test_create_and_load_snapshot() {
479+
let (old_db, temp_dir) = create_test_db();
480+
481+
let v1 = vec![0.0, 1.0, 2.0];
482+
let v2 = vec![3.0, 4.0, 5.0];
483+
let v3 = vec![6.0, 7.0, 8.0];
484+
485+
let id1 = old_db
486+
.insert(
487+
v1.clone(),
488+
Payload {
489+
content_type: ContentType::Text,
490+
content: "test".to_string(),
491+
},
492+
)
493+
.unwrap();
494+
495+
let id2 = old_db
496+
.insert(
497+
v2.clone(),
498+
Payload {
499+
content_type: ContentType::Text,
500+
content: "test".to_string(),
501+
},
502+
)
503+
.unwrap();
504+
505+
let temp_snapshot_dir = tempdir().unwrap();
506+
let snapshot_path = old_db.create_snapshot(temp_snapshot_dir.path()).unwrap();
507+
508+
// insert v3 after snapshot
509+
let id3 = old_db
510+
.insert(
511+
v3.clone(),
512+
Payload {
513+
content_type: ContentType::Text,
514+
content: "test".to_string(),
515+
},
516+
)
517+
.unwrap();
518+
519+
let reload_config = DbRestoreConfig {
520+
data_path: temp_dir.path().to_path_buf(),
521+
snapshot_path,
522+
};
523+
524+
std::mem::drop(old_db);
525+
let loaded_db = restore_from_snapshot(&reload_config).unwrap();
526+
527+
assert!(loaded_db.get(id1).unwrap_or(None).is_some());
528+
assert!(loaded_db.get(id2).unwrap_or(None).is_some());
529+
assert!(loaded_db.get(id3).unwrap_or(None).is_none()); // v3 was inserted after snapshot was taken
530+
531+
// vector restore check
532+
assert!(loaded_db.get(id1).unwrap().unwrap().vector.unwrap() == v1);
533+
assert!(loaded_db.get(id2).unwrap().unwrap().vector.unwrap() == v2);
534+
}
535+
536+
#[test]
537+
fn test_snapshot_engine() {
538+
let (_db, _temp_dir) = create_test_db();
539+
let db = Arc::new(Mutex::new(_db));
540+
541+
let registry_tempdir = tempdir().unwrap();
542+
543+
let registry = Arc::new(Mutex::new(
544+
LocalRegistry::new(registry_tempdir.path()).unwrap(),
545+
));
546+
547+
let last_k = 4;
548+
let mut se = SnapshotEngine::new(last_k, db.clone(), registry.clone());
549+
550+
let v1 = vec![0.0, 1.0, 2.0];
551+
let v2 = vec![3.0, 4.0, 5.0];
552+
let v3 = vec![6.0, 7.0, 8.0];
553+
554+
let test_vectors = vec![v1.clone(), v2.clone(), v3.clone()];
555+
let mut inserted_ids = Vec::new();
556+
557+
for (i, vector) in test_vectors.clone().into_iter().enumerate() {
558+
se.snapshot().unwrap();
559+
let id = db
560+
.lock()
561+
.unwrap()
562+
.insert(
563+
vector.clone(),
564+
Payload {
565+
content_type: ContentType::Text,
566+
content: format!("{}", i),
567+
},
568+
)
569+
.unwrap();
570+
inserted_ids.push(id);
571+
}
572+
se.snapshot().unwrap();
573+
let snapshots = se.list_alive_snapshots().unwrap();
574+
575+
// asserting these cases:
576+
// snapshot 0 : no vectors
577+
// snapshot 1 : v1
578+
// snapshot 2 : v1, v2
579+
// snapshot 3 : v1, v2, v3
580+
581+
std::mem::drop(db);
582+
std::mem::drop(se);
583+
584+
for (i, snapshot) in snapshots.iter().enumerate() {
585+
let temp_dir = tempdir().unwrap();
586+
let db = restore_from_snapshot(&DbRestoreConfig {
587+
data_path: temp_dir.path().to_path_buf(),
588+
snapshot_path: snapshot.path.clone(),
589+
})
590+
.unwrap();
591+
for j in 0..i {
592+
// test if point is present
593+
assert!(db.get(inserted_ids[j]).unwrap_or(None).is_some());
594+
// test vector restore
595+
assert!(
596+
db.get(inserted_ids[j]).unwrap().unwrap().vector.unwrap() == test_vectors[j]
597+
);
598+
// test payload restore
599+
assert!(
600+
db.get(inserted_ids[j])
601+
.unwrap()
602+
.unwrap()
603+
.payload
604+
.unwrap()
605+
.content
606+
== format!("{}", j)
607+
);
608+
}
609+
for absent_id in inserted_ids.iter().skip(i) {
610+
assert!(db.get(*absent_id).unwrap_or(None).is_none());
611+
}
612+
std::mem::drop(db);
613+
}
614+
}
422615
}

‎crates/defs/src/error.rs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,29 @@
11
use std::io;
22

3+
use crate::{Dimension, PointId};
4+
#[derive(Debug, PartialEq, Eq)]
5+
pub enum DbError {
6+
ParseError,
7+
StorageError(String),
8+
SerializationError(String),
9+
DeserializationError,
10+
IndexError(String),
11+
LockError,
12+
IndexInitError, //TODO: Change this
13+
UnsupportedSimilarity,
14+
DimensionMismatch,
15+
SnapshotError(String),
16+
StorageInitializationError,
17+
StorageCheckpointError(String),
18+
InvalidMagicBytes(String),
19+
VectorNotFound(uuid::Uuid),
20+
SnapshotRegistryError(String),
21+
SnapshotEngineError(String),
22+
InvalidDimension { expected: Dimension, got: Dimension },
23+
PointAlreadyExists { id: PointId },
24+
PointNotFound { id: PointId },
25+
}
26+
327
#[derive(Debug)]
428
pub enum ServerError {
529
Bind(io::Error),

‎crates/defs/src/lib.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,10 @@ pub mod types;
33

44
// Without re-exports, users would need to write defs::types::SomeType instead of just defs::SomeType. Re-exports simplify the API by flattening the module hierarchy. The * means "everything public" from that module.
55
pub use error::*;
6+
use std::path::{Path, PathBuf};
67
pub use types::*;
8+
9+
// hoisted trait so it can be used by the snapshots crate
10+
pub trait SnapshottableDb: Send + Sync {
11+
fn create_snapshot(&self, dir_path: &Path) -> Result<PathBuf, DbError>;
12+
}

‎crates/defs/src/types.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ pub type Dimension = usize;
1515
// Sparse vector implementation not supported yet. Refer lib/sparse/src/common/sparse_vector.rs
1616
pub type DenseVector = Vec<Element>;
1717

18+
pub type Magic = [u8; 4];
19+
1820
pub enum StoredVector {
1921
Dense(DenseVector),
2022
}
@@ -45,7 +47,7 @@ pub struct IndexedVector {
4547
pub vector: DenseVector,
4648
}
4749

48-
#[derive(Debug, Deserialize, Copy, Clone)]
50+
#[derive(Debug, Serialize, Deserialize, Copy, Clone)]
4951
pub enum Similarity {
5052
Euclidean,
5153
Manhattan,

‎crates/grpc/src/error.rs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,21 @@ impl From<storage::error::StorageError> for GrpcError {
120120
StorageError::Deserialization { id, source: _ } => GrpcError::Internal {
121121
message: format!("failed to deserialize point {}", id),
122122
},
123+
StorageError::RocksDbInitialization {} => GrpcError::Internal {
124+
message: "failed to initialize storage".to_string(),
125+
},
126+
StorageError::RocksDbCheckpointMsg { msg } => GrpcError::Internal {
127+
message: format!("checkpoint error: {}", msg),
128+
},
129+
StorageError::RocksDbCheckpointIo { msg, source: _ } => GrpcError::Internal {
130+
message: format!("checkpoint io error: {}", msg),
131+
},
132+
StorageError::RocksDbCheckpoint { source: _ } => GrpcError::Internal {
133+
message: "checkpoint error".to_string(),
134+
},
135+
StorageError::RocksDbFlush { source: _ } => GrpcError::Internal {
136+
message: "flush error".to_string(),
137+
},
123138
}
124139
}
125140
}

‎crates/http/src/handler.rs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,12 @@ fn api_error_to_response(err: &ApiError) -> (StatusCode, String) {
126126
| StorageError::RocksDbDelete { .. }
127127
| StorageError::RocksDbIteration { .. }
128128
| StorageError::Serialization { .. }
129-
| StorageError::Deserialization { .. } => {
129+
| StorageError::Deserialization { .. }
130+
| StorageError::RocksDbCheckpoint { .. }
131+
| StorageError::RocksDbFlush { .. }
132+
| StorageError::RocksDbInitialization { .. }
133+
| StorageError::RocksDbCheckpointMsg { .. }
134+
| StorageError::RocksDbCheckpointIo { .. } => {
130135
(StatusCode::INTERNAL_SERVER_ERROR, source.to_string())
131136
}
132137
},

‎crates/index/Cargo.toml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,10 @@ edition.workspace = true
77
license.workspace = true
88

99
[dependencies]
10+
bincode.workspace = true
1011
defs.workspace = true
1112
rand.workspace = true
13+
serde.workspace = true
1214
snafu.workspace = true
15+
storage.workspace = true
1316
uuid.workspace = true

0 commit comments

Comments
 (0)