1use crate::backend::{BackendPartition, BackendWriteOp};
2use crate::config::VantaConfig;
3use crate::error::{Result, VantaError};
4use crate::executor::{ExecutionResult, Executor};
5use crate::index::{cosine_sim_f32, set_prefetch_mode};
7use crate::node::{DistanceMetric, FieldValue, UnifiedNode, VectorRepresentations};
8use crate::storage::{IndexRebuildReport, StorageEngine};
9use parking_lot::RwLock;
10use serde::{Deserialize, Serialize};
11use std::collections::{BTreeMap, BTreeSet};
12use std::fs::File;
13use std::hash::Hasher;
14use std::io::{BufRead, BufReader, BufWriter, Write};
15use std::path::Path;
16use std::sync::Arc;
17use std::time::{Instant, SystemTime, UNIX_EPOCH};
18use twox_hash::XxHash64;
19
20const RESERVED_PREFIX: &str = "__vanta_";
21const FIELD_NAMESPACE: &str = "__vanta_namespace";
22const FIELD_KEY: &str = "__vanta_key";
23const FIELD_PAYLOAD: &str = "__vanta_payload";
24const FIELD_CREATED_AT_MS: &str = "__vanta_created_at_ms";
25const FIELD_UPDATED_AT_MS: &str = "__vanta_updated_at_ms";
26const FIELD_VERSION: &str = "__vanta_version";
27const FIELD_EXPIRES_AT_MS: &str = "__vanta_expires_at_ms";
28const EXPORT_SCHEMA_VERSION: u32 = 1;
29const DERIVED_INDEX_SCHEMA_VERSION: u32 = 1;
30const DERIVED_INDEX_STATE_KEY: &[u8] = b"derived_index_state";
31const TEXT_INDEX_STATE_KEY: &[u8] = b"text_index_state";
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub enum VantaRuntimeProfile {
39 Enterprise,
40 Performance,
41 LowResource,
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub enum VantaStorageTier {
47 Hot,
48 Cold,
49}
50
51#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
53pub enum VantaValue {
54 String(String),
55 Int(i64),
56 Float(f64),
57 Bool(bool),
58 DateTime(chrono::DateTime<chrono::Utc>),
59 ListString(Vec<String>),
60 ListInt(Vec<i64>),
61 ListFloat(Vec<f64>),
62 ListBool(Vec<bool>),
63 ListDateTime(Vec<chrono::DateTime<chrono::Utc>>),
64 Null,
65}
66
67impl VantaValue {
68 pub fn to_index_values(&self) -> Vec<VantaValue> {
69 match self {
70 VantaValue::ListString(vec) => {
71 vec.iter().map(|s| VantaValue::String(s.clone())).collect()
72 }
73 VantaValue::ListInt(vec) => vec.iter().map(|&i| VantaValue::Int(i)).collect(),
74 VantaValue::ListFloat(vec) => vec.iter().map(|&f| VantaValue::Float(f)).collect(),
75 VantaValue::ListBool(vec) => vec.iter().map(|&b| VantaValue::Bool(b)).collect(),
76 VantaValue::ListDateTime(vec) => {
77 vec.iter().map(|&dt| VantaValue::DateTime(dt)).collect()
78 }
79 other => vec![other.clone()],
80 }
81 }
82}
83
84pub type VantaFields = BTreeMap<String, VantaValue>;
86
87pub type VantaMemoryMetadata = VantaFields;
89
90#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
92pub struct VantaMemoryInput {
93 pub namespace: String,
94 pub key: String,
95 pub payload: String,
96 pub metadata: VantaMemoryMetadata,
97 pub vector: Option<Vec<f32>>,
98 pub ttl_ms: Option<u64>,
102}
103
104impl VantaMemoryInput {
105 pub fn new(
106 namespace: impl Into<String>,
107 key: impl Into<String>,
108 payload: impl Into<String>,
109 ) -> Self {
110 Self {
111 namespace: namespace.into(),
112 key: key.into(),
113 payload: payload.into(),
114 metadata: VantaMemoryMetadata::new(),
115 vector: None,
116 ttl_ms: None,
117 }
118 }
119}
120
121#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
123pub struct VantaMemoryRecord {
124 pub namespace: String,
125 pub key: String,
126 pub payload: String,
127 pub metadata: VantaMemoryMetadata,
128 pub created_at_ms: u64,
129 pub updated_at_ms: u64,
130 pub version: u64,
131 pub node_id: u64,
132 pub vector: Option<Vec<f32>>,
133 pub expires_at_ms: Option<u64>,
136}
137
138#[derive(Debug, Clone, PartialEq)]
140pub struct VantaMemoryListOptions {
141 pub filters: VantaMemoryMetadata,
142 pub limit: usize,
143 pub cursor: Option<usize>,
144}
145
146impl Default for VantaMemoryListOptions {
147 fn default() -> Self {
148 Self {
149 filters: VantaMemoryMetadata::new(),
150 limit: 100,
151 cursor: None,
152 }
153 }
154}
155
156#[derive(Debug, Clone, PartialEq)]
158pub struct VantaMemoryListPage {
159 pub records: Vec<VantaMemoryRecord>,
160 pub next_cursor: Option<usize>,
161}
162
163#[derive(Debug, Clone, PartialEq)]
165pub struct VantaMemorySearchRequest {
166 pub namespace: String,
167 pub query_vector: Vec<f32>,
168 pub filters: VantaMemoryMetadata,
169 pub text_query: Option<String>,
170 pub top_k: usize,
171 pub distance_metric: DistanceMetric,
173 pub explain: bool,
175}
176
177impl Default for VantaMemorySearchRequest {
178 fn default() -> Self {
179 Self {
180 namespace: String::new(),
181 query_vector: Vec::new(),
182 filters: Default::default(),
183 text_query: None,
184 top_k: 10,
185 distance_metric: DistanceMetric::Cosine,
186 explain: false,
187 }
188 }
189}
190
191#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
193pub struct VantaMemorySearchHit {
194 pub record: VantaMemoryRecord,
195 pub score: f32,
196 pub explanation: Option<VantaSearchExplanationHit>,
197}
198
199#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
201pub struct VantaIndexRebuildReport {
202 pub scanned_nodes: u64,
203 pub indexed_vectors: u64,
204 pub skipped_tombstones: u64,
205 pub duration_ms: u64,
206 pub derived_rebuild_ms: u64,
207 pub index_path: String,
208 pub success: bool,
209}
210
211#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
213pub struct VantaExportReport {
214 pub records_exported: u64,
215 pub namespaces: Vec<String>,
216 pub path: String,
217 pub duration_ms: u64,
218}
219
220#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
222pub struct VantaImportReport {
223 pub inserted: u64,
224 pub updated: u64,
225 pub skipped: u64,
226 pub errors: u64,
227 pub duration_ms: u64,
228}
229
230#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
232pub struct VantaTextIndexRepairReport {
233 pub record_count: u64,
234 pub posting_entries: u64,
235 pub doc_stats_entries: u64,
236 pub term_stats_entries: u64,
237 pub namespace_stats_entries: u64,
238 pub duration_ms: u64,
239 pub success: bool,
240}
241
242#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
244pub struct VantaOperationalMetrics {
245 pub startup_ms: u64,
246 pub wal_replay_ms: u64,
247 pub wal_records_replayed: u64,
248 pub ann_rebuild_ms: u64,
249 pub ann_rebuild_scanned_nodes: u64,
250 pub derived_rebuild_ms: u64,
251 pub text_index_rebuild_ms: u64,
252 pub text_postings_written: u64,
253 pub text_index_repairs: u64,
254 pub text_lexical_queries: u64,
255 pub text_lexical_query_ms: u64,
256 pub text_candidates_scored: u64,
257 pub text_consistency_audits: u64,
258 pub text_consistency_audit_failures: u64,
259 pub hybrid_query_ms: u64,
260 pub hybrid_candidates_fused: u64,
261 pub planner_hybrid_queries: u64,
262 pub planner_text_only_queries: u64,
263 pub planner_vector_only_queries: u64,
264 pub records_exported: u64,
265 pub records_imported: u64,
266 pub import_errors: u64,
267 pub derived_prefix_scans: u64,
268 pub derived_full_scan_fallbacks: u64,
269 pub process_rss_bytes: u64,
271 pub process_virtual_bytes: u64,
272 pub hnsw_nodes_count: u64,
273 pub hnsw_logical_bytes: u64,
274 pub mmap_resident_bytes: Option<u64>,
275 pub volatile_cache_entries: u64,
276 pub volatile_cache_cap_bytes: u64,
277}
278
279#[cfg(debug_assertions)]
280#[derive(Debug, Clone, PartialEq, Eq)]
281#[doc(hidden)]
282pub struct VantaMemorySearchDebugReport {
283 pub route: String,
284 pub budget: usize,
285 pub text_candidates: usize,
286 pub vector_candidates: usize,
287 pub fused_candidates: usize,
288 pub top_identities: Vec<String>,
289}
290
291#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
292pub struct VantaHybridFusionReport {
293 pub text_candidates: usize,
294 pub vector_candidates: usize,
295 pub fused_candidates: usize,
296 pub rrf_k: usize,
297}
298
299#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
300pub struct VantaSearchExplanation {
301 pub route: String,
302 pub hits: Vec<VantaSearchExplanationHit>,
303 pub fusion_report: Option<VantaHybridFusionReport>,
304}
305
306#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
307pub struct VantaSearchExplanationHit {
308 pub identity: String,
309 pub score: f32,
310 pub snippet: Option<String>,
311 pub matched_tokens: Vec<String>,
312 pub matched_phrases: Vec<String>,
313 pub bm25_terms: Vec<VantaBm25TermContribution>,
314 pub rrf_text_rank: Option<usize>,
315 pub rrf_vector_rank: Option<usize>,
316}
317
318#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
319pub struct VantaBm25TermContribution {
320 pub token: String,
321 pub tf: u32,
322 pub df: u64,
323 pub doc_len: u32,
324 pub contribution: f32,
325}
326
327#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
328struct DerivedIndexState {
329 schema_version: u32,
330 rebuilt_at_ms: u64,
331 record_count: u64,
332 namespace_entries: u64,
333 payload_entries: u64,
334}
335
336#[derive(Debug, Clone, PartialEq, Eq)]
337struct DerivedIndexRebuildReport {
338 record_count: u64,
339 namespace_entries: u64,
340 payload_entries: u64,
341 duration_ms: u64,
342}
343
344#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
345struct TextIndexState {
346 schema_version: u32,
347 tokenizer: String,
348 tokenizer_version: u32,
349 key_format: String,
350 rebuilt_at_ms: u64,
351 record_count: u64,
352 posting_entries: u64,
353 doc_stats_entries: u64,
354 term_stats_entries: u64,
355 namespace_stats_entries: u64,
356}
357
358#[derive(Debug, Clone, PartialEq, Eq)]
359struct TextIndexRebuildReport {
360 record_count: u64,
361 posting_entries: u64,
362 doc_stats_entries: u64,
363 term_stats_entries: u64,
364 namespace_stats_entries: u64,
365 duration_ms: u64,
366}
367
368#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
369struct TextIndexCounts {
370 record_count: u64,
371 posting_entries: u64,
372 doc_stats_entries: u64,
373 term_stats_entries: u64,
374 namespace_stats_entries: u64,
375 unknown_entries: u64,
376}
377
378#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
379struct TextIndexMutationReport {
380 postings_written: u64,
381 doc_stats_delta: i64,
382 term_stats_delta: i64,
383 namespace_stats_delta: i64,
384}
385
386#[derive(Debug, Clone, Default, PartialEq, Eq)]
387struct ExpectedTextIndexEntries {
388 entries: BTreeMap<Vec<u8>, Vec<u8>>,
389 counts: TextIndexCounts,
390 records_scanned: u64,
391 namespaces: BTreeSet<String>,
392}
393
394#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
399pub struct VantaTextIndexAuditReport {
400 pub schema_version: u32,
401 pub tokenizer: String,
402 pub tokenizer_version: u32,
403 pub key_format: String,
404 pub namespace_filter: Option<String>,
405 pub namespaces_audited: Vec<String>,
406 pub records_scanned: u64,
407 pub expected_entries: u64,
408 pub actual_entries: u64,
409 pub missing_entries: u64,
410 pub unexpected_entries: u64,
411 pub value_mismatches: u64,
412 pub unreadable_entries: u64,
413 pub mismatches: u64,
414 pub deep_audit: bool,
415 pub position_errors: u64,
416 pub tf_errors: u64,
417 pub df_errors: u64,
418 pub doc_len_errors: u64,
419 pub logical_corruptions: u64,
420 pub state_valid: bool,
421 pub state_status: String,
422 pub duration_ms: u64,
423 pub passed: bool,
424 pub status: String,
425}
426
427#[derive(Debug, Clone, Serialize, Deserialize)]
428struct VantaMemoryExportLine {
429 schema_version: u32,
430 namespace: String,
431 key: String,
432 payload: String,
433 metadata: VantaMemoryMetadata,
434 vector: Option<Vec<f32>>,
435 created_at_ms: u64,
436 updated_at_ms: u64,
437 version: u64,
438 expires_at_ms: Option<u64>,
439}
440
441#[derive(Debug, Clone, PartialEq)]
443pub struct VantaEdgeRecord {
444 pub target: u64,
445 pub label: String,
446 pub weight: f32,
447}
448
449#[derive(Debug, Clone, PartialEq)]
451pub struct VantaNodeInput {
452 pub id: u64,
453 pub content: Option<String>,
454 pub vector: Option<Vec<f32>>,
455 pub fields: VantaFields,
456}
457
458impl VantaNodeInput {
459 pub fn new(id: u64) -> Self {
460 Self {
461 id,
462 content: None,
463 vector: None,
464 fields: VantaFields::new(),
465 }
466 }
467}
468
469#[derive(Debug, Clone, PartialEq)]
471pub struct VantaNodeRecord {
472 pub id: u64,
473 pub fields: VantaFields,
474 pub vector: Option<Vec<f32>>,
475 pub vector_dimensions: usize,
476 pub edges: Vec<VantaEdgeRecord>,
477 pub confidence_score: f32,
478 pub importance: f32,
479 pub hits: u32,
480 pub last_accessed: u64,
481 pub epoch: u32,
482 pub tier: VantaStorageTier,
483 pub is_alive: bool,
484}
485
486#[derive(Debug, Clone, PartialEq)]
488pub struct VantaSearchHit {
489 pub node_id: u64,
490 pub distance: f32,
491}
492
493#[derive(Debug, Clone, PartialEq)]
495pub enum VantaQueryResult {
496 Read(Vec<VantaNodeRecord>),
497 Write {
498 affected_nodes: usize,
499 message: String,
500 node_id: Option<u64>,
501 },
502 StaleContext {
503 node_id: u64,
504 },
505}
506
507#[derive(Debug, Clone, PartialEq, Eq)]
509pub struct VantaCapabilities {
510 pub runtime_profile: VantaRuntimeProfile,
511 pub persistence: bool,
512 pub vector_search: bool,
513 pub iql_queries: bool,
514 pub read_only: bool,
515}
516
517#[derive(Clone)]
519pub struct VantaEmbedded {
520 engine: Arc<RwLock<Option<Arc<StorageEngine>>>>,
521 config: VantaConfig,
522}
523
524fn now_ms() -> u64 {
525 SystemTime::now()
526 .duration_since(UNIX_EPOCH)
527 .unwrap_or_default()
528 .as_millis() as u64
529}
530
531fn memory_node_id(namespace: &str, key: &str) -> u64 {
532 let mut hasher = XxHash64::default();
533 hasher.write(namespace.as_bytes());
534 hasher.write(&[0]);
535 hasher.write(key.as_bytes());
536 hasher.finish()
537}
538
539fn validate_namespace(namespace: &str) -> Result<()> {
540 if namespace.is_empty() {
541 return Err(VantaError::Execution(
542 "namespace must not be empty".to_string(),
543 ));
544 }
545 if namespace.len() > 128 {
546 return Err(VantaError::Execution(
547 "namespace must be at most 128 bytes".to_string(),
548 ));
549 }
550 if !namespace
551 .bytes()
552 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'/' | b'-'))
553 {
554 return Err(VantaError::Execution(
555 "namespace may contain only A-Z, a-z, 0-9, '.', '_', '/', '-'".to_string(),
556 ));
557 }
558 Ok(())
559}
560
561fn validate_key(key: &str) -> Result<()> {
562 if key.is_empty() {
563 return Err(VantaError::Execution("key must not be empty".to_string()));
564 }
565 if key.len() > 512 {
566 return Err(VantaError::Execution(
567 "key must be at most 512 bytes".to_string(),
568 ));
569 }
570 if key.as_bytes().contains(&0) {
571 return Err(VantaError::Execution(
572 "key must not contain NUL bytes".to_string(),
573 ));
574 }
575 Ok(())
576}
577
578fn validate_metadata(metadata: &VantaMemoryMetadata) -> Result<()> {
579 if let Some(key) = metadata.keys().find(|key| key.starts_with(RESERVED_PREFIX)) {
580 return Err(VantaError::Execution(format!(
581 "metadata key '{}' is reserved for VantaDB internals",
582 key
583 )));
584 }
585 if let Some(key) = metadata.keys().find(|key| key.as_bytes().contains(&0)) {
586 return Err(VantaError::Execution(format!(
587 "metadata key '{}' must not contain NUL bytes",
588 key
589 )));
590 }
591 Ok(())
592}
593
594fn namespace_index_key(namespace: &str, key: &str) -> Vec<u8> {
595 let mut index_key = Vec::with_capacity(namespace.len() + 1 + key.len());
596 index_key.extend_from_slice(namespace.as_bytes());
597 index_key.push(0);
598 index_key.extend_from_slice(key.as_bytes());
599 index_key
600}
601
602fn namespace_index_prefix(namespace: &str) -> Vec<u8> {
603 let mut prefix = Vec::with_capacity(namespace.len() + 1);
604 prefix.extend_from_slice(namespace.as_bytes());
605 prefix.push(0);
606 prefix
607}
608
609fn encoded_scalar_value(value: &VantaValue) -> Result<Vec<u8>> {
610 match value {
611 VantaValue::String(value) => {
612 let mut encoded = b"s:".to_vec();
613 encoded.extend_from_slice(value.as_bytes());
614 Ok(encoded)
615 }
616 VantaValue::Int(value) => Ok(format!("i:{value}").into_bytes()),
617 VantaValue::Float(value) => Ok(format!("f:{:016x}", value.to_bits()).into_bytes()),
618 VantaValue::Bool(value) => {
619 if *value {
620 Ok(b"b:1".to_vec())
621 } else {
622 Ok(b"b:0".to_vec())
623 }
624 }
625 VantaValue::DateTime(dt) => {
626 let mut encoded = b"d:".to_vec();
627 encoded.extend_from_slice(
628 dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)
629 .as_bytes(),
630 );
631 Ok(encoded)
632 }
633 VantaValue::ListString(_)
634 | VantaValue::ListInt(_)
635 | VantaValue::ListFloat(_)
636 | VantaValue::ListBool(_)
637 | VantaValue::ListDateTime(_) => Err(VantaError::Execution(
638 "Cannot encode list value as scalar index key".to_string(),
639 )),
640 VantaValue::Null => Ok(b"n:".to_vec()),
641 }
642}
643
644fn payload_index_prefix(namespace: &str, field: &str, value: &VantaValue) -> Result<Vec<u8>> {
645 let encoded = encoded_scalar_value(value)?;
646 let mut prefix = Vec::with_capacity(namespace.len() + field.len() + encoded.len() + 3);
647 prefix.extend_from_slice(namespace.as_bytes());
648 prefix.push(0);
649 prefix.extend_from_slice(field.as_bytes());
650 prefix.push(0);
651 prefix.extend_from_slice(&encoded);
652 prefix.push(0);
653 Ok(prefix)
654}
655
656fn payload_index_key(
657 namespace: &str,
658 field: &str,
659 value: &VantaValue,
660 key: &str,
661) -> Result<Vec<u8>> {
662 let mut index_key = payload_index_prefix(namespace, field, value)?;
663 index_key.extend_from_slice(key.as_bytes());
664 Ok(index_key)
665}
666
667fn node_id_bytes(node_id: u64) -> Vec<u8> {
668 node_id.to_le_bytes().to_vec()
669}
670
671fn decode_node_id(bytes: &[u8]) -> Option<u64> {
672 if bytes.len() != std::mem::size_of::<u64>() {
673 return None;
674 }
675 let mut id = [0u8; 8];
676 id.copy_from_slice(bytes);
677 Some(u64::from_le_bytes(id))
678}
679
680fn get_string_field(fields: &VantaFields, key: &str) -> Option<String> {
681 match fields.get(key) {
682 Some(VantaValue::String(value)) => Some(value.clone()),
683 _ => None,
684 }
685}
686
687fn get_u64_field(fields: &VantaFields, key: &str) -> Option<u64> {
688 match fields.get(key) {
689 Some(VantaValue::Int(value)) if *value >= 0 => Some(*value as u64),
690 _ => None,
691 }
692}
693
694fn memory_record_from_node(node: UnifiedNode) -> Option<VantaMemoryRecord> {
695 if !node.is_alive() {
696 return None;
697 }
698
699 let mut fields: VantaFields = node
700 .relational
701 .into_iter()
702 .map(|(key, value)| (key, value.into()))
703 .collect();
704
705 let namespace = get_string_field(&fields, FIELD_NAMESPACE)?;
706 let key = get_string_field(&fields, FIELD_KEY)?;
707 let payload = get_string_field(&fields, FIELD_PAYLOAD)?;
708 let created_at_ms = get_u64_field(&fields, FIELD_CREATED_AT_MS)?;
709 let updated_at_ms = get_u64_field(&fields, FIELD_UPDATED_AT_MS)?;
710 let version = get_u64_field(&fields, FIELD_VERSION)?;
711 let expires_at_ms = get_u64_field(&fields, FIELD_EXPIRES_AT_MS);
712
713 fields.remove(FIELD_NAMESPACE);
714 fields.remove(FIELD_KEY);
715 fields.remove(FIELD_PAYLOAD);
716 fields.remove(FIELD_CREATED_AT_MS);
717 fields.remove(FIELD_UPDATED_AT_MS);
718 fields.remove(FIELD_VERSION);
719 fields.remove(FIELD_EXPIRES_AT_MS);
720
721 if let Some(deadline) = expires_at_ms {
724 if deadline > 0 {
725 let now = now_ms();
726 if now > deadline {
727 return None;
728 }
729 }
730 }
731
732 let vector = match node.vector {
733 VectorRepresentations::Full(vector) => Some(vector),
734 _ => None,
735 };
736
737 Some(VantaMemoryRecord {
738 namespace,
739 key,
740 payload,
741 metadata: fields,
742 created_at_ms,
743 updated_at_ms,
744 version,
745 node_id: node.id,
746 vector,
747 expires_at_ms,
748 })
749}
750
751fn memory_record_to_node(record: &VantaMemoryRecord) -> UnifiedNode {
752 let mut node = UnifiedNode::new(record.node_id);
753 node.set_field(
754 FIELD_NAMESPACE,
755 FieldValue::String(record.namespace.clone()),
756 );
757 node.set_field(FIELD_KEY, FieldValue::String(record.key.clone()));
758 node.set_field(FIELD_PAYLOAD, FieldValue::String(record.payload.clone()));
759 node.set_field(
760 FIELD_CREATED_AT_MS,
761 FieldValue::Int(record.created_at_ms as i64),
762 );
763 node.set_field(
764 FIELD_UPDATED_AT_MS,
765 FieldValue::Int(record.updated_at_ms as i64),
766 );
767 node.set_field(FIELD_VERSION, FieldValue::Int(record.version as i64));
768
769 if let Some(expires_at) = record.expires_at_ms {
770 node.set_field(FIELD_EXPIRES_AT_MS, FieldValue::Int(expires_at as i64));
771 }
772
773 for (key, value) in record.metadata.clone() {
774 node.set_field(key, value.into());
775 }
776
777 if let Some(vector) = record.vector.clone().filter(|vector| !vector.is_empty()) {
778 node.vector = VectorRepresentations::Full(vector);
779 node.flags.set(crate::node::NodeFlags::HAS_VECTOR);
780 }
781
782 node
783}
784
785fn export_line_from_record(record: VantaMemoryRecord) -> VantaMemoryExportLine {
786 VantaMemoryExportLine {
787 schema_version: EXPORT_SCHEMA_VERSION,
788 namespace: record.namespace,
789 key: record.key,
790 payload: record.payload,
791 metadata: record.metadata,
792 vector: record.vector,
793 created_at_ms: record.created_at_ms,
794 updated_at_ms: record.updated_at_ms,
795 version: record.version,
796 expires_at_ms: record.expires_at_ms,
797 }
798}
799
800fn record_from_export_line(line: VantaMemoryExportLine) -> Result<VantaMemoryRecord> {
801 if line.schema_version != EXPORT_SCHEMA_VERSION {
802 return Err(VantaError::Execution(format!(
803 "unsupported memory export schema_version {}",
804 line.schema_version
805 )));
806 }
807
808 let node_id = memory_node_id(&line.namespace, &line.key);
809 Ok(VantaMemoryRecord {
810 namespace: line.namespace,
811 key: line.key,
812 payload: line.payload,
813 metadata: line.metadata,
814 created_at_ms: line.created_at_ms,
815 updated_at_ms: line.updated_at_ms,
816 version: line.version,
817 node_id,
818 vector: line.vector,
819 expires_at_ms: line.expires_at_ms,
820 })
821}
822
823fn matches_memory_filters(record: &VantaMemoryRecord, filters: &VantaMemoryMetadata) -> bool {
824 filters
825 .iter()
826 .all(|(key, expected)| record.metadata.get(key) == Some(expected))
827}
828
829impl VantaEmbedded {
830 pub fn from_engine(engine: Arc<StorageEngine>) -> Self {
831 let config = engine.config.clone();
832 Self {
833 engine: Arc::new(RwLock::new(Some(engine))),
834 config,
835 }
836 }
837
838 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
839 let config = VantaConfig {
840 storage_path: path.as_ref().to_string_lossy().into_owned(),
841 ..Default::default()
842 };
843 Self::open_with_config(config)
844 }
845
846 pub fn open_with_config(config: VantaConfig) -> Result<Self> {
847 let final_config = config.clone();
848 set_prefetch_mode(config.prefetch_mode);
849
850 let engine = StorageEngine::open_with_config(
851 &final_config.storage_path,
852 Some(final_config.clone()),
853 )?;
854 let embedded = Self {
855 engine: Arc::new(RwLock::new(Some(Arc::new(engine)))),
856 config: final_config,
857 };
858 if !embedded.config.read_only {
859 embedded.ensure_derived_indexes_current()?;
860 embedded.ensure_text_index_current()?;
861 }
862 Ok(embedded)
863 }
864
865 fn engine_handle(&self) -> Result<Arc<StorageEngine>> {
866 self.engine.read().clone().ok_or(VantaError::NotInitialized)
867 }
868
869 fn load_derived_index_state(engine: &StorageEngine) -> Result<Option<DerivedIndexState>> {
870 let Some(bytes) = engine
871 .get_from_partition(BackendPartition::InternalMetadata, DERIVED_INDEX_STATE_KEY)?
872 else {
873 return Ok(None);
874 };
875 bincode::serde::decode_from_slice(&bytes, bincode::config::standard())
876 .map(|(val, _)| Some(val))
877 .map_err(|err| {
878 VantaError::SerializationError(format!("derived index state decode error: {err}"))
879 })
880 }
881
882 fn write_derived_index_state(engine: &StorageEngine, state: &DerivedIndexState) -> Result<()> {
883 let bytes = bincode::serde::encode_to_vec(state, bincode::config::standard())
884 .map_err(|err| VantaError::SerializationError(err.to_string()))?;
885 engine.put_to_partition(
886 BackendPartition::InternalMetadata,
887 DERIVED_INDEX_STATE_KEY,
888 &bytes,
889 )
890 }
891
892 fn load_text_index_state(engine: &StorageEngine) -> Result<Option<TextIndexState>> {
893 let Some(bytes) =
894 engine.get_from_partition(BackendPartition::InternalMetadata, TEXT_INDEX_STATE_KEY)?
895 else {
896 return Ok(None);
897 };
898 bincode::serde::decode_from_slice(&bytes, bincode::config::standard())
899 .map(|(val, _)| Some(val))
900 .map_err(|err| {
901 VantaError::SerializationError(format!("text index state decode error: {err}"))
902 })
903 }
904
905 fn write_text_index_state(engine: &StorageEngine, state: &TextIndexState) -> Result<()> {
906 let bytes = bincode::serde::encode_to_vec(state, bincode::config::standard())
907 .map_err(|err| VantaError::SerializationError(err.to_string()))?;
908 engine.put_to_partition(
909 BackendPartition::InternalMetadata,
910 TEXT_INDEX_STATE_KEY,
911 &bytes,
912 )
913 }
914
915 fn fresh_text_index_state(counts: TextIndexCounts) -> TextIndexState {
916 let spec = crate::text_index::TextIndexSpec::default();
917 TextIndexState {
918 schema_version: spec.schema_version,
919 tokenizer: spec.tokenizer.name.to_string(),
920 tokenizer_version: spec.tokenizer.version,
921 key_format: spec.key_format.to_string(),
922 rebuilt_at_ms: now_ms(),
923 record_count: counts.record_count,
924 posting_entries: counts.posting_entries,
925 doc_stats_entries: counts.doc_stats_entries,
926 term_stats_entries: counts.term_stats_entries,
927 namespace_stats_entries: counts.namespace_stats_entries,
928 }
929 }
930
931 fn text_index_state_matches_spec(state: &TextIndexState) -> bool {
932 let spec = crate::text_index::TextIndexSpec::default();
933 state.schema_version == spec.schema_version
934 && state.tokenizer == spec.tokenizer.name
935 && state.tokenizer_version == spec.tokenizer.version
936 && state.key_format == spec.key_format
937 }
938
939 fn count_memory_records(engine: &StorageEngine) -> Result<u64> {
940 let mut count = 0u64;
941 for node in engine.scan_nodes()? {
942 if memory_record_from_node(node).is_some() {
943 count += 1;
944 }
945 }
946 Ok(count)
947 }
948
949 fn expected_text_index_counts(engine: &StorageEngine) -> Result<TextIndexCounts> {
950 let mut counts = TextIndexCounts::default();
951 let mut terms = BTreeSet::new();
952 let mut namespaces = BTreeSet::new();
953
954 for node in engine.scan_nodes()? {
955 if let Some(record) = memory_record_from_node(node) {
956 counts.record_count += 1;
957 counts.posting_entries += crate::text_index::posting_count(&record.payload);
958 counts.doc_stats_entries += 1;
959 namespaces.insert(record.namespace.clone());
960 for token in crate::text_index::unique_tokens(&record.payload) {
961 terms.insert((record.namespace.clone(), token));
962 }
963 }
964 }
965
966 counts.term_stats_entries = terms.len() as u64;
967 counts.namespace_stats_entries = namespaces.len() as u64;
968 Ok(counts)
969 }
970
971 fn current_text_index_counts(engine: &StorageEngine) -> Result<TextIndexCounts> {
972 let mut counts = TextIndexCounts::default();
973 for (key, _value) in engine.scan_partition(BackendPartition::TextIndex)? {
974 if !crate::text_index::is_internal_key(&key) {
975 counts.posting_entries += 1;
976 continue;
977 }
978
979 if crate::text_index::is_doc_stats_key(&key) {
980 counts.doc_stats_entries += 1;
981 } else if crate::text_index::is_term_stats_key(&key) {
982 counts.term_stats_entries += 1;
983 } else if crate::text_index::is_namespace_stats_key(&key) {
984 counts.namespace_stats_entries += 1;
985 } else {
986 counts.unknown_entries += 1;
987 }
988 }
989 Ok(counts)
990 }
991
992 fn current_derived_index_counts(engine: &StorageEngine) -> Result<(u64, u64)> {
993 let namespace_entries = engine
994 .scan_partition(BackendPartition::NamespaceIndex)?
995 .len() as u64;
996 let payload_entries = engine.scan_partition(BackendPartition::PayloadIndex)?.len() as u64;
997 Ok((namespace_entries, payload_entries))
998 }
999
1000 fn derived_put_ops(record: &VantaMemoryRecord) -> Result<Vec<BackendWriteOp>> {
1001 let mut ops = Vec::new();
1002 ops.push(BackendWriteOp::Put {
1003 partition: BackendPartition::NamespaceIndex,
1004 key: namespace_index_key(&record.namespace, &record.key),
1005 value: node_id_bytes(record.node_id),
1006 });
1007
1008 for (field, value) in &record.metadata {
1009 for val in value.to_index_values() {
1010 ops.push(BackendWriteOp::Put {
1011 partition: BackendPartition::PayloadIndex,
1012 key: payload_index_key(&record.namespace, field, &val, &record.key)?,
1013 value: node_id_bytes(record.node_id),
1014 });
1015 }
1016 }
1017
1018 Ok(ops)
1019 }
1020
1021 fn derived_delete_ops(record: &VantaMemoryRecord) -> Result<Vec<BackendWriteOp>> {
1022 let mut ops = Vec::new();
1023 ops.push(BackendWriteOp::Delete {
1024 partition: BackendPartition::NamespaceIndex,
1025 key: namespace_index_key(&record.namespace, &record.key),
1026 });
1027
1028 for (field, value) in &record.metadata {
1029 for val in value.to_index_values() {
1030 ops.push(BackendWriteOp::Delete {
1031 partition: BackendPartition::PayloadIndex,
1032 key: payload_index_key(&record.namespace, field, &val, &record.key)?,
1033 });
1034 }
1035 }
1036
1037 Ok(ops)
1038 }
1039
1040 fn load_text_term_stats(
1041 engine: &StorageEngine,
1042 namespace: &str,
1043 token: &str,
1044 ) -> Result<Option<crate::text_index::TextTermStats>> {
1045 let cache_key = (namespace.to_string(), token.to_string());
1046 {
1047 let cache = engine.text_stats_cache.read();
1048 if let Some(stats) = cache.get(&cache_key) {
1049 return Ok(Some(stats.clone()));
1050 }
1051 }
1052
1053 let skey = crate::text_index::term_stats_key(namespace, token);
1054 let Some(bytes) = engine.get_from_partition(BackendPartition::TextIndex, &skey)? else {
1055 return Ok(None);
1056 };
1057 let stats = crate::text_index::decode_term_stats(&bytes)?;
1058
1059 {
1060 let mut cache = engine.text_stats_cache.write();
1061 cache.insert(cache_key, stats.clone());
1062 }
1063 Ok(Some(stats))
1064 }
1065
1066 fn load_text_namespace_stats(
1067 engine: &StorageEngine,
1068 namespace: &str,
1069 ) -> Result<Option<crate::text_index::TextNamespaceStats>> {
1070 {
1071 let cache = engine.text_ns_cache.read();
1072 if let Some(stats) = cache.get(namespace) {
1073 return Ok(Some(stats.clone()));
1074 }
1075 }
1076
1077 let skey = crate::text_index::namespace_stats_key(namespace);
1078 let Some(bytes) = engine.get_from_partition(BackendPartition::TextIndex, &skey)? else {
1079 return Ok(None);
1080 };
1081 let stats = crate::text_index::decode_namespace_stats(&bytes)?;
1082
1083 {
1084 let mut cache = engine.text_ns_cache.write();
1085 cache.insert(namespace.to_string(), stats.clone());
1086 }
1087 Ok(Some(stats))
1088 }
1089
1090 fn load_text_doc_stats(
1091 engine: &StorageEngine,
1092 namespace: &str,
1093 key: &str,
1094 ) -> Result<Option<crate::text_index::TextDocStats>> {
1095 let Some(bytes) = engine.get_from_partition(
1096 BackendPartition::TextIndex,
1097 &crate::text_index::doc_stats_key(namespace, key),
1098 )?
1099 else {
1100 return Ok(None);
1101 };
1102 crate::text_index::decode_doc_stats(&bytes).map(Some)
1103 }
1104
1105 fn apply_u64_delta(value: u64, delta: i64) -> u64 {
1106 if delta >= 0 {
1107 value.saturating_add(delta as u64)
1108 } else {
1109 value.saturating_sub(delta.unsigned_abs())
1110 }
1111 }
1112
1113 fn checked_stats_value(value: i128, label: &str) -> Result<u64> {
1114 if value < 0 {
1115 return Err(VantaError::Execution(format!(
1116 "text index {label} would become negative"
1117 )));
1118 }
1119 u64::try_from(value).map_err(|_| {
1120 VantaError::Execution(format!("text index {label} exceeds supported range"))
1121 })
1122 }
1123
1124 fn text_index_ops_for_replace(
1125 engine: &StorageEngine,
1126 previous: Option<&VantaMemoryRecord>,
1127 current: Option<&VantaMemoryRecord>,
1128 ) -> Result<(Vec<BackendWriteOp>, TextIndexMutationReport)> {
1129 let mut ops = Vec::new();
1130 let mut report = TextIndexMutationReport::default();
1131 let mut term_deltas: BTreeMap<(String, String), i64> = BTreeMap::new();
1132 let mut namespace_deltas: BTreeMap<String, (i64, i64)> = BTreeMap::new();
1133
1134 if let Some(previous) = previous {
1135 let terms = crate::text_index::record_terms(&previous.payload);
1136 ops.extend(crate::text_index::posting_delete_ops(
1137 &previous.namespace,
1138 &previous.key,
1139 &previous.payload,
1140 ));
1141 ops.push(crate::text_index::doc_stats_delete_op(
1142 &previous.namespace,
1143 &previous.key,
1144 ));
1145 report.doc_stats_delta -= 1;
1146
1147 for token in terms.token_counts.keys() {
1148 *term_deltas
1149 .entry((previous.namespace.clone(), token.clone()))
1150 .or_default() -= 1;
1151 }
1152 let namespace_delta = namespace_deltas
1153 .entry(previous.namespace.clone())
1154 .or_insert((0, 0));
1155 namespace_delta.0 -= 1;
1156 namespace_delta.1 -= i64::from(terms.doc_len);
1157 }
1158
1159 if let Some(current) = current {
1160 let terms = crate::text_index::record_terms(¤t.payload);
1161 let posting_ops = crate::text_index::posting_put_ops(
1162 ¤t.namespace,
1163 ¤t.key,
1164 ¤t.payload,
1165 current.node_id,
1166 )?;
1167 report.postings_written = posting_ops.len() as u64;
1168 ops.extend(posting_ops);
1169 ops.push(crate::text_index::doc_stats_put_op(
1170 ¤t.namespace,
1171 ¤t.key,
1172 ¤t.payload,
1173 current.node_id,
1174 )?);
1175 report.doc_stats_delta += 1;
1176
1177 for token in terms.token_counts.keys() {
1178 *term_deltas
1179 .entry((current.namespace.clone(), token.clone()))
1180 .or_default() += 1;
1181 }
1182 let namespace_delta = namespace_deltas
1183 .entry(current.namespace.clone())
1184 .or_insert((0, 0));
1185 namespace_delta.0 += 1;
1186 namespace_delta.1 += i64::from(terms.doc_len);
1187 }
1188
1189 for ((namespace, token), delta) in term_deltas {
1190 if delta == 0 {
1191 continue;
1192 }
1193
1194 let existing = Self::load_text_term_stats(engine, &namespace, &token)?
1195 .map(|stats| stats.df)
1196 .unwrap_or(0);
1197 let next = Self::checked_stats_value(existing as i128 + delta as i128, "df")?;
1198 match (existing == 0, next == 0) {
1199 (true, false) => report.term_stats_delta += 1,
1200 (false, true) => report.term_stats_delta -= 1,
1201 _ => {}
1202 }
1203 if next == 0 {
1204 ops.push(crate::text_index::term_stats_delete_op(&namespace, &token));
1205 } else {
1206 ops.push(crate::text_index::term_stats_put_op(
1207 &namespace, &token, next,
1208 )?);
1209 }
1210 }
1211
1212 for (namespace, (doc_delta, len_delta)) in namespace_deltas {
1213 if doc_delta == 0 && len_delta == 0 {
1214 continue;
1215 }
1216
1217 let existing = Self::load_text_namespace_stats(engine, &namespace)?.unwrap_or(
1218 crate::text_index::TextNamespaceStats {
1219 doc_count: 0,
1220 total_doc_len: 0,
1221 },
1222 );
1223 let next_doc_count = Self::checked_stats_value(
1224 existing.doc_count as i128 + doc_delta as i128,
1225 "doc_count",
1226 )?;
1227 let next_total_doc_len = Self::checked_stats_value(
1228 existing.total_doc_len as i128 + len_delta as i128,
1229 "total_doc_len",
1230 )?;
1231
1232 match (existing.doc_count == 0, next_doc_count == 0) {
1233 (true, false) => report.namespace_stats_delta += 1,
1234 (false, true) => report.namespace_stats_delta -= 1,
1235 _ => {}
1236 }
1237
1238 if next_doc_count == 0 {
1239 ops.push(crate::text_index::namespace_stats_delete_op(&namespace));
1240 } else {
1241 ops.push(crate::text_index::namespace_stats_put_op(
1242 &namespace,
1243 &crate::text_index::TextNamespaceStats {
1244 doc_count: next_doc_count,
1245 total_doc_len: next_total_doc_len,
1246 },
1247 )?);
1248 }
1249 }
1250
1251 Ok((ops, report))
1252 }
1253
1254 fn replace_derived_indexes(
1255 &self,
1256 engine: &StorageEngine,
1257 previous: Option<&VantaMemoryRecord>,
1258 current: Option<&VantaMemoryRecord>,
1259 ) -> Result<()> {
1260 let mut ops = Vec::new();
1261 if let Some(previous) = previous {
1262 ops.extend(Self::derived_delete_ops(previous)?);
1263 }
1264 if let Some(current) = current {
1265 ops.extend(Self::derived_put_ops(current)?);
1266 }
1267 let (text_ops, text_report) = Self::text_index_ops_for_replace(engine, previous, current)?;
1268 ops.extend(text_ops);
1269 if ops.is_empty() {
1270 return Ok(());
1271 }
1272 engine.write_backend_batch(ops.clone())?;
1273
1274 for op in &ops {
1276 match op {
1277 BackendWriteOp::Put {
1278 partition: BackendPartition::TextIndex,
1279 key,
1280 value,
1281 } => {
1282 if crate::text_index::is_term_stats_key(key) {
1283 if let Some((ns, token)) = Self::parse_term_stats_key(key) {
1284 if let Ok(stats) = crate::text_index::decode_term_stats(value) {
1285 let mut cache = engine.text_stats_cache.write();
1286 cache.insert((ns, token), stats);
1287 }
1288 }
1289 } else if crate::text_index::is_namespace_stats_key(key) {
1290 if let Some(ns) = Self::parse_namespace_stats_key(key) {
1291 if let Ok(stats) = crate::text_index::decode_namespace_stats(value) {
1292 let mut cache = engine.text_ns_cache.write();
1293 cache.insert(ns, stats);
1294 }
1295 }
1296 }
1297 }
1298 BackendWriteOp::Delete {
1299 partition: BackendPartition::TextIndex,
1300 key,
1301 } => {
1302 if crate::text_index::is_term_stats_key(key) {
1303 if let Some((ns, token)) = Self::parse_term_stats_key(key) {
1304 let mut cache = engine.text_stats_cache.write();
1305 cache.remove(&(ns, token));
1306 }
1307 } else if crate::text_index::is_namespace_stats_key(key) {
1308 if let Some(ns) = Self::parse_namespace_stats_key(key) {
1309 let mut cache = engine.text_ns_cache.write();
1310 cache.remove(&ns);
1311 }
1312 }
1313 }
1314 _ => {}
1315 }
1316 }
1317
1318 Self::adjust_derived_index_state_after_replace(engine, previous, current)?;
1319 Self::adjust_text_index_state_after_replace(engine, previous, current, text_report)?;
1320 crate::metrics::record_text_postings_written(text_report.postings_written);
1321 Ok(())
1322 }
1323
1324 fn parse_term_stats_key(key: &[u8]) -> Option<(String, String)> {
1325 const INTERNAL_PREFIX: &[u8] = b"\xffvanta_text_v3\0";
1326 const TERM_STATS_TAG: &[u8] = b"term\0";
1327 let remainder = key
1328 .strip_prefix(INTERNAL_PREFIX)?
1329 .strip_prefix(TERM_STATS_TAG)?;
1330 let pos = remainder.iter().position(|&b| b == 0)?;
1331 let ns = String::from_utf8(remainder[..pos].to_vec()).ok()?;
1332 let token = String::from_utf8(remainder[pos + 1..].to_vec()).ok()?;
1333 Some((ns, token))
1334 }
1335
1336 fn parse_namespace_stats_key(key: &[u8]) -> Option<String> {
1337 const INTERNAL_PREFIX: &[u8] = b"\xffvanta_text_v3\0";
1338 const NAMESPACE_STATS_TAG: &[u8] = b"ns\0";
1339 let remainder = key
1340 .strip_prefix(INTERNAL_PREFIX)?
1341 .strip_prefix(NAMESPACE_STATS_TAG)?;
1342 String::from_utf8(remainder.to_vec()).ok()
1343 }
1344
1345 fn adjust_derived_index_state_after_replace(
1346 engine: &StorageEngine,
1347 previous: Option<&VantaMemoryRecord>,
1348 current: Option<&VantaMemoryRecord>,
1349 ) -> Result<()> {
1350 let Some(mut state) = Self::load_derived_index_state(engine)? else {
1351 return Ok(());
1352 };
1353 if state.schema_version != DERIVED_INDEX_SCHEMA_VERSION {
1354 return Ok(());
1355 }
1356
1357 match (previous, current) {
1358 (None, Some(current)) => {
1359 state.record_count = state.record_count.saturating_add(1);
1360 state.namespace_entries = state.namespace_entries.saturating_add(1);
1361 state.payload_entries = state
1362 .payload_entries
1363 .saturating_add(current.metadata.len() as u64);
1364 }
1365 (Some(previous), None) => {
1366 state.record_count = state.record_count.saturating_sub(1);
1367 state.namespace_entries = state.namespace_entries.saturating_sub(1);
1368 state.payload_entries = state
1369 .payload_entries
1370 .saturating_sub(previous.metadata.len() as u64);
1371 }
1372 (Some(previous), Some(current)) => {
1373 state.payload_entries = state
1374 .payload_entries
1375 .saturating_sub(previous.metadata.len() as u64)
1376 .saturating_add(current.metadata.len() as u64);
1377 }
1378 (None, None) => {}
1379 }
1380
1381 Self::write_derived_index_state(engine, &state)
1382 }
1383
1384 fn adjust_text_index_state_after_replace(
1385 engine: &StorageEngine,
1386 previous: Option<&VantaMemoryRecord>,
1387 current: Option<&VantaMemoryRecord>,
1388 report: TextIndexMutationReport,
1389 ) -> Result<()> {
1390 let Some(mut state) = Self::load_text_index_state(engine)? else {
1391 return Ok(());
1392 };
1393 if !Self::text_index_state_matches_spec(&state) {
1394 return Ok(());
1395 }
1396
1397 match (previous, current) {
1398 (None, Some(current)) => {
1399 state.record_count = state.record_count.saturating_add(1);
1400 state.posting_entries = state
1401 .posting_entries
1402 .saturating_add(crate::text_index::posting_count(¤t.payload));
1403 state.doc_stats_entries =
1404 Self::apply_u64_delta(state.doc_stats_entries, report.doc_stats_delta);
1405 }
1406 (Some(previous), None) => {
1407 state.record_count = state.record_count.saturating_sub(1);
1408 state.posting_entries = state
1409 .posting_entries
1410 .saturating_sub(crate::text_index::posting_count(&previous.payload));
1411 state.doc_stats_entries =
1412 Self::apply_u64_delta(state.doc_stats_entries, report.doc_stats_delta);
1413 }
1414 (Some(previous), Some(current)) => {
1415 state.posting_entries = state
1416 .posting_entries
1417 .saturating_sub(crate::text_index::posting_count(&previous.payload))
1418 .saturating_add(crate::text_index::posting_count(¤t.payload));
1419 state.doc_stats_entries =
1420 Self::apply_u64_delta(state.doc_stats_entries, report.doc_stats_delta);
1421 }
1422 (None, None) => {}
1423 }
1424 state.term_stats_entries =
1425 Self::apply_u64_delta(state.term_stats_entries, report.term_stats_delta);
1426 state.namespace_stats_entries =
1427 Self::apply_u64_delta(state.namespace_stats_entries, report.namespace_stats_delta);
1428
1429 Self::write_text_index_state(engine, &state)
1430 }
1431
1432 fn ensure_derived_indexes_current(&self) -> Result<()> {
1433 let engine = self.engine_handle()?;
1434 let state = match Self::load_derived_index_state(&engine) {
1435 Ok(state) => state,
1436 Err(_) => {
1437 self.rebuild_derived_indexes_with_report()?;
1438 return Ok(());
1439 }
1440 };
1441
1442 let canonical_records = Self::count_memory_records(&engine)?;
1443 let (namespace_entries, payload_entries) = Self::current_derived_index_counts(&engine)?;
1444
1445 let has_state = state.is_some();
1446 let needs_rebuild = match &state {
1447 Some(state) => {
1448 state.schema_version != DERIVED_INDEX_SCHEMA_VERSION
1449 || state.record_count != canonical_records
1450 || state.namespace_entries != namespace_entries
1451 || state.payload_entries != payload_entries
1452 || namespace_entries < canonical_records
1453 }
1454 None => canonical_records > 0 || namespace_entries > 0 || payload_entries > 0,
1455 };
1456
1457 if needs_rebuild {
1458 self.rebuild_derived_indexes_with_report()?;
1459 } else if !has_state {
1460 Self::write_derived_index_state(
1461 &engine,
1462 &DerivedIndexState {
1463 schema_version: DERIVED_INDEX_SCHEMA_VERSION,
1464 rebuilt_at_ms: now_ms(),
1465 record_count: canonical_records,
1466 namespace_entries,
1467 payload_entries,
1468 },
1469 )?;
1470 }
1471
1472 Ok(())
1473 }
1474
1475 fn ensure_text_index_current(&self) -> Result<()> {
1476 let engine = self.engine_handle()?;
1477 let state = match Self::load_text_index_state(&engine) {
1478 Ok(state) => state,
1479 Err(_) => {
1480 crate::metrics::record_text_index_repair();
1481 self.rebuild_text_index_with_report()?;
1482 return Ok(());
1483 }
1484 };
1485
1486 let expected = Self::expected_text_index_counts(&engine)?;
1487 let current = Self::current_text_index_counts(&engine)?;
1488
1489 let has_state = state.is_some();
1490 let needs_rebuild = match &state {
1491 Some(state) => {
1492 !Self::text_index_state_matches_spec(state)
1493 || state.record_count != expected.record_count
1494 || state.posting_entries != current.posting_entries
1495 || state.posting_entries != expected.posting_entries
1496 || state.doc_stats_entries != current.doc_stats_entries
1497 || state.doc_stats_entries != expected.doc_stats_entries
1498 || state.term_stats_entries != current.term_stats_entries
1499 || state.term_stats_entries != expected.term_stats_entries
1500 || state.namespace_stats_entries != current.namespace_stats_entries
1501 || state.namespace_stats_entries != expected.namespace_stats_entries
1502 || current.posting_entries != expected.posting_entries
1503 || current.doc_stats_entries != expected.doc_stats_entries
1504 || current.term_stats_entries != expected.term_stats_entries
1505 || current.namespace_stats_entries != expected.namespace_stats_entries
1506 || current.unknown_entries != 0
1507 }
1508 None => {
1509 expected.record_count > 0
1510 || current.posting_entries > 0
1511 || current.doc_stats_entries > 0
1512 || current.term_stats_entries > 0
1513 || current.namespace_stats_entries > 0
1514 || current.unknown_entries > 0
1515 }
1516 };
1517
1518 if needs_rebuild {
1519 crate::metrics::record_text_index_repair();
1520 self.rebuild_text_index_with_report()?;
1521 } else if !has_state {
1522 Self::write_text_index_state(&engine, &Self::fresh_text_index_state(expected))?;
1523 }
1524
1525 Ok(())
1526 }
1527
1528 fn rebuild_derived_indexes_with_report(&self) -> Result<DerivedIndexRebuildReport> {
1529 let started = Instant::now();
1530 let engine = self.engine_handle()?;
1531 let mut ops = Vec::new();
1532 let mut record_count = 0u64;
1533 let mut namespace_entries = 0u64;
1534 let mut payload_entries = 0u64;
1535
1536 for (key, _value) in engine.scan_partition(BackendPartition::NamespaceIndex)? {
1537 ops.push(BackendWriteOp::Delete {
1538 partition: BackendPartition::NamespaceIndex,
1539 key,
1540 });
1541 }
1542 for (key, _value) in engine.scan_partition(BackendPartition::PayloadIndex)? {
1543 ops.push(BackendWriteOp::Delete {
1544 partition: BackendPartition::PayloadIndex,
1545 key,
1546 });
1547 }
1548 for node in engine.scan_nodes()? {
1549 if let Some(record) = memory_record_from_node(node) {
1550 record_count += 1;
1551 namespace_entries += 1;
1552 payload_entries += record.metadata.len() as u64;
1553 ops.extend(Self::derived_put_ops(&record)?);
1554 }
1555 }
1556
1557 if !ops.is_empty() {
1558 engine.write_backend_batch(ops)?;
1559 }
1560
1561 Self::write_derived_index_state(
1562 &engine,
1563 &DerivedIndexState {
1564 schema_version: DERIVED_INDEX_SCHEMA_VERSION,
1565 rebuilt_at_ms: now_ms(),
1566 record_count,
1567 namespace_entries,
1568 payload_entries,
1569 },
1570 )?;
1571
1572 let report = DerivedIndexRebuildReport {
1573 record_count,
1574 namespace_entries,
1575 payload_entries,
1576 duration_ms: started.elapsed().as_millis() as u64,
1577 };
1578 crate::metrics::record_derived_rebuild(report.duration_ms);
1579 Ok(report)
1580 }
1581
1582 fn rebuild_derived_indexes(&self) -> Result<()> {
1583 self.rebuild_derived_indexes_with_report().map(|_| ())
1584 }
1585
1586 fn rebuild_text_index_with_report(&self) -> Result<TextIndexRebuildReport> {
1587 let started = Instant::now();
1588 let engine = self.engine_handle()?;
1589
1590 {
1592 let mut cache = engine.text_stats_cache.write();
1593 cache.clear();
1594 }
1595 {
1596 let mut cache = engine.text_ns_cache.write();
1597 cache.clear();
1598 }
1599
1600 let mut ops = Vec::new();
1601 let mut counts = TextIndexCounts::default();
1602 let mut term_stats: BTreeMap<(String, String), u64> = BTreeMap::new();
1603 let mut namespace_stats: BTreeMap<String, crate::text_index::TextNamespaceStats> =
1604 BTreeMap::new();
1605
1606 for (key, _value) in engine.scan_partition(BackendPartition::TextIndex)? {
1607 ops.push(BackendWriteOp::Delete {
1608 partition: BackendPartition::TextIndex,
1609 key,
1610 });
1611 }
1612
1613 for node in engine.scan_nodes()? {
1614 if let Some(record) = memory_record_from_node(node) {
1615 counts.record_count += 1;
1616 let posting_ops = crate::text_index::posting_put_ops(
1617 &record.namespace,
1618 &record.key,
1619 &record.payload,
1620 record.node_id,
1621 )?;
1622 counts.posting_entries += posting_ops.len() as u64;
1623 ops.extend(posting_ops);
1624 ops.push(crate::text_index::doc_stats_put_op(
1625 &record.namespace,
1626 &record.key,
1627 &record.payload,
1628 record.node_id,
1629 )?);
1630 counts.doc_stats_entries += 1;
1631
1632 let terms = crate::text_index::record_terms(&record.payload);
1633 for token in terms.token_counts.keys() {
1634 *term_stats
1635 .entry((record.namespace.clone(), token.clone()))
1636 .or_default() += 1;
1637 }
1638 let namespace = namespace_stats.entry(record.namespace.clone()).or_insert(
1639 crate::text_index::TextNamespaceStats {
1640 doc_count: 0,
1641 total_doc_len: 0,
1642 },
1643 );
1644 namespace.doc_count += 1;
1645 namespace.total_doc_len += u64::from(terms.doc_len);
1646 }
1647 }
1648
1649 for ((namespace, token), df) in &term_stats {
1650 ops.push(crate::text_index::term_stats_put_op(namespace, token, *df)?);
1651 }
1652 for (namespace, stats) in &namespace_stats {
1653 ops.push(crate::text_index::namespace_stats_put_op(namespace, stats)?);
1654 }
1655 counts.term_stats_entries = term_stats.len() as u64;
1656 counts.namespace_stats_entries = namespace_stats.len() as u64;
1657
1658 if !ops.is_empty() {
1659 engine.write_backend_batch(ops)?;
1660 }
1661
1662 Self::write_text_index_state(&engine, &Self::fresh_text_index_state(counts))?;
1663
1664 let report = TextIndexRebuildReport {
1665 record_count: counts.record_count,
1666 posting_entries: counts.posting_entries,
1667 doc_stats_entries: counts.doc_stats_entries,
1668 term_stats_entries: counts.term_stats_entries,
1669 namespace_stats_entries: counts.namespace_stats_entries,
1670 duration_ms: started.elapsed().as_millis() as u64,
1671 };
1672 crate::metrics::record_text_index_rebuild(report.duration_ms, report.posting_entries);
1673 Ok(report)
1674 }
1675
1676 fn rebuild_text_index(&self) -> Result<()> {
1677 self.rebuild_text_index_with_report().map(|_| ())
1678 }
1679
1680 fn expected_text_index_entries(
1681 engine: &StorageEngine,
1682 namespace_filter: Option<&str>,
1683 ) -> Result<ExpectedTextIndexEntries> {
1684 let mut audit = ExpectedTextIndexEntries::default();
1685 let mut term_stats: BTreeMap<(String, String), u64> = BTreeMap::new();
1686 let mut namespace_stats: BTreeMap<String, crate::text_index::TextNamespaceStats> =
1687 BTreeMap::new();
1688
1689 for node in engine.scan_nodes()? {
1690 audit.records_scanned += 1;
1691 if let Some(record) = memory_record_from_node(node) {
1692 if matches!(namespace_filter, Some(namespace) if record.namespace != namespace) {
1693 continue;
1694 }
1695 audit.counts.record_count += 1;
1696 audit.namespaces.insert(record.namespace.clone());
1697 let terms = crate::text_index::record_terms(&record.payload);
1698 for (token, tf) in &terms.token_counts {
1699 audit.entries.insert(
1700 crate::text_index::posting_key(&record.namespace, token, &record.key),
1701 crate::text_index::posting_value(
1702 record.node_id,
1703 *tf,
1704 terms
1705 .token_positions
1706 .get(token)
1707 .map(Vec::as_slice)
1708 .unwrap_or(&[]),
1709 )?,
1710 );
1711 audit.counts.posting_entries += 1;
1712 *term_stats
1713 .entry((record.namespace.clone(), token.clone()))
1714 .or_default() += 1;
1715 }
1716 audit.entries.insert(
1717 crate::text_index::doc_stats_key(&record.namespace, &record.key),
1718 crate::text_index::doc_stats_value(record.node_id, terms.doc_len)?,
1719 );
1720 audit.counts.doc_stats_entries += 1;
1721 let namespace = namespace_stats.entry(record.namespace.clone()).or_insert(
1722 crate::text_index::TextNamespaceStats {
1723 doc_count: 0,
1724 total_doc_len: 0,
1725 },
1726 );
1727 namespace.doc_count += 1;
1728 namespace.total_doc_len += u64::from(terms.doc_len);
1729 }
1730 }
1731
1732 for ((namespace, token), df) in term_stats {
1733 audit.entries.insert(
1734 crate::text_index::term_stats_key(&namespace, &token),
1735 crate::text_index::term_stats_value(df)?,
1736 );
1737 }
1738 for (namespace, stats) in namespace_stats {
1739 audit.entries.insert(
1740 crate::text_index::namespace_stats_key(&namespace),
1741 crate::text_index::namespace_stats_value(stats.doc_count, stats.total_doc_len)?,
1742 );
1743 }
1744
1745 audit.counts.term_stats_entries = audit
1746 .entries
1747 .keys()
1748 .filter(|key| crate::text_index::is_term_stats_key(key))
1749 .count() as u64;
1750 audit.counts.namespace_stats_entries = audit
1751 .entries
1752 .keys()
1753 .filter(|key| crate::text_index::is_namespace_stats_key(key))
1754 .count() as u64;
1755
1756 Ok(audit)
1757 }
1758
1759 fn text_index_value_readable(key: &[u8], value: &[u8]) -> bool {
1760 if !crate::text_index::is_internal_key(key) {
1761 return crate::text_index::decode_posting(value).is_ok();
1762 }
1763
1764 if crate::text_index::is_doc_stats_key(key) {
1765 crate::text_index::decode_doc_stats(value).is_ok()
1766 } else if crate::text_index::is_term_stats_key(key) {
1767 crate::text_index::decode_term_stats(value).is_ok()
1768 } else if crate::text_index::is_namespace_stats_key(key) {
1769 crate::text_index::decode_namespace_stats(value).is_ok()
1770 } else {
1771 false
1772 }
1773 }
1774
1775 fn text_index_state_audit_status(
1776 engine: &StorageEngine,
1777 expected_counts: TextIndexCounts,
1778 namespace_filter: Option<&str>,
1779 ) -> (bool, String) {
1780 let state = match Self::load_text_index_state(engine) {
1781 Ok(Some(state)) => state,
1782 Ok(None) => return (false, "missing".to_string()),
1783 Err(err) => return (false, format!("decode_error: {err}")),
1784 };
1785
1786 if !Self::text_index_state_matches_spec(&state) {
1787 return (false, "incompatible".to_string());
1788 }
1789
1790 if namespace_filter.is_none()
1791 && (state.record_count != expected_counts.record_count
1792 || state.posting_entries != expected_counts.posting_entries
1793 || state.doc_stats_entries != expected_counts.doc_stats_entries
1794 || state.term_stats_entries != expected_counts.term_stats_entries
1795 || state.namespace_stats_entries != expected_counts.namespace_stats_entries)
1796 {
1797 return (false, "count_mismatch".to_string());
1798 }
1799
1800 (true, "current".to_string())
1801 }
1802
1803 fn build_text_index_audit_report_deep(
1804 engine: &StorageEngine,
1805 namespace_filter: Option<&str>,
1806 ) -> Result<VantaTextIndexAuditReport> {
1807 let started = Instant::now();
1808 let spec = crate::text_index::TextIndexSpec::default();
1809 let expected = Self::expected_text_index_entries(engine, namespace_filter)?;
1810 let actual: BTreeMap<Vec<u8>, Vec<u8>> = engine
1811 .scan_partition(BackendPartition::TextIndex)?
1812 .into_iter()
1813 .filter(|(key, _value)| {
1814 namespace_filter
1815 .map(|namespace| {
1816 crate::text_index::text_index_key_belongs_to_namespace(key, namespace)
1817 })
1818 .unwrap_or(true)
1819 })
1820 .collect();
1821
1822 let mut missing_entries = 0u64;
1823 let mut unexpected_entries = 0u64;
1824 let mut value_mismatches = 0u64;
1825 let mut unreadable_entries = 0u64;
1826 let mut position_errors = 0u64;
1827 let mut tf_errors = 0u64;
1828 let mut df_errors = 0u64;
1829 let mut doc_len_errors = 0u64;
1830 let mut logical_corruptions = 0u64;
1831
1832 for (key, value) in &expected.entries {
1833 match actual.get(key) {
1834 Some(actual_value) if actual_value == value => {}
1835 Some(actual_value) => {
1836 value_mismatches += 1;
1837 if !Self::text_index_value_readable(key, actual_value) {
1838 unreadable_entries += 1;
1839 } else if crate::text_index::is_doc_stats_key(key) {
1840 if let (Ok(expected_stats), Ok(actual_stats)) = (
1841 crate::text_index::decode_doc_stats(value),
1842 crate::text_index::decode_doc_stats(actual_value),
1843 ) {
1844 if expected_stats.doc_len != actual_stats.doc_len {
1845 doc_len_errors += 1;
1846 } else {
1847 logical_corruptions += 1;
1848 }
1849 }
1850 } else if crate::text_index::is_term_stats_key(key) {
1851 if let (Ok(expected_stats), Ok(actual_stats)) = (
1852 crate::text_index::decode_term_stats(value),
1853 crate::text_index::decode_term_stats(actual_value),
1854 ) {
1855 if expected_stats.df != actual_stats.df {
1856 df_errors += 1;
1857 } else {
1858 logical_corruptions += 1;
1859 }
1860 }
1861 } else if !crate::text_index::is_internal_key(key) {
1862 if let (Ok(expected_posting), Ok(actual_posting)) = (
1863 crate::text_index::decode_posting(value),
1864 crate::text_index::decode_posting(actual_value),
1865 ) {
1866 if expected_posting.tf != actual_posting.tf {
1867 tf_errors += 1;
1868 }
1869 if expected_posting.positions != actual_posting.positions {
1870 position_errors += 1;
1871 }
1872 if expected_posting.tf == actual_posting.tf
1873 && expected_posting.positions == actual_posting.positions
1874 {
1875 logical_corruptions += 1;
1876 }
1877 }
1878 } else {
1879 logical_corruptions += 1;
1880 }
1881 }
1882 None => missing_entries += 1,
1883 }
1884 }
1885 for key in actual.keys() {
1886 if !expected.entries.contains_key(key) {
1887 unexpected_entries += 1;
1888 if let Some(value) = actual.get(key) {
1889 if !Self::text_index_value_readable(key, value) {
1890 unreadable_entries += 1;
1891 }
1892 }
1893 }
1894 }
1895
1896 let (state_valid, state_status) =
1897 Self::text_index_state_audit_status(engine, expected.counts, namespace_filter);
1898 let state_mismatches = u64::from(!state_valid);
1899 let mismatches = missing_entries + unexpected_entries + value_mismatches + state_mismatches;
1900 let passed = mismatches == 0;
1901 let mut namespaces_audited: Vec<String> = expected.namespaces.into_iter().collect();
1902 if namespaces_audited.is_empty() {
1903 if let Some(namespace) = namespace_filter {
1904 namespaces_audited.push(namespace.to_string());
1905 }
1906 }
1907
1908 let report = VantaTextIndexAuditReport {
1909 schema_version: spec.schema_version,
1910 tokenizer: spec.tokenizer.name.to_string(),
1911 tokenizer_version: spec.tokenizer.version,
1912 key_format: spec.key_format.to_string(),
1913 namespace_filter: namespace_filter.map(ToOwned::to_owned),
1914 namespaces_audited,
1915 records_scanned: expected.records_scanned,
1916 expected_entries: expected.entries.len() as u64,
1917 actual_entries: actual.len() as u64,
1918 missing_entries,
1919 unexpected_entries,
1920 value_mismatches,
1921 unreadable_entries,
1922 mismatches,
1923 deep_audit: true,
1924 position_errors,
1925 tf_errors,
1926 df_errors,
1927 doc_len_errors,
1928 logical_corruptions,
1929 state_valid,
1930 state_status,
1931 duration_ms: started.elapsed().as_millis() as u64,
1932 passed,
1933 status: if passed {
1934 "ok".to_string()
1935 } else {
1936 "repair_recommended".to_string()
1937 },
1938 };
1939 crate::metrics::record_text_consistency_audit(!report.passed);
1940 Ok(report)
1941 }
1942
1943 fn build_text_index_audit_report_shallow(
1944 engine: &StorageEngine,
1945 namespace_filter: Option<&str>,
1946 ) -> Result<VantaTextIndexAuditReport> {
1947 let started = Instant::now();
1948 let spec = crate::text_index::TextIndexSpec::default();
1949 let expected = Self::expected_text_index_entries(engine, namespace_filter)?;
1950
1951 let (state_valid, state_status) =
1952 Self::text_index_state_audit_status(engine, expected.counts, namespace_filter);
1953
1954 let actual: BTreeSet<Vec<u8>> = engine
1956 .scan_partition(BackendPartition::TextIndex)?
1957 .into_iter()
1958 .filter(|(key, _value)| {
1959 namespace_filter
1960 .map(|namespace| {
1961 crate::text_index::text_index_key_belongs_to_namespace(key, namespace)
1962 })
1963 .unwrap_or(true)
1964 })
1965 .map(|(key, _value)| key)
1966 .collect();
1967
1968 let actual_entries = actual.len() as u64;
1969 let expected_keys: BTreeSet<&Vec<u8>> = expected.entries.keys().collect();
1970 let missing_entries = expected_keys
1971 .iter()
1972 .filter(|key| !actual.contains(**key))
1973 .count() as u64;
1974 let unexpected_entries = actual
1975 .iter()
1976 .filter(|key| !expected.entries.contains_key(*key))
1977 .count() as u64;
1978 let mismatches = missing_entries + unexpected_entries;
1979
1980 let passed = state_valid && mismatches == 0;
1981
1982 let mut namespaces_audited: Vec<String> = expected.namespaces.into_iter().collect();
1983 if namespaces_audited.is_empty() {
1984 if let Some(namespace) = namespace_filter {
1985 namespaces_audited.push(namespace.to_string());
1986 }
1987 }
1988
1989 let report = VantaTextIndexAuditReport {
1990 schema_version: spec.schema_version,
1991 tokenizer: spec.tokenizer.name.to_string(),
1992 tokenizer_version: spec.tokenizer.version,
1993 key_format: spec.key_format.to_string(),
1994 namespace_filter: namespace_filter.map(ToOwned::to_owned),
1995 namespaces_audited,
1996 records_scanned: expected.records_scanned,
1997 expected_entries: expected.entries.len() as u64,
1998 actual_entries,
1999 missing_entries,
2000 unexpected_entries,
2001 value_mismatches: 0,
2002 unreadable_entries: 0,
2003 mismatches,
2004 deep_audit: false,
2005 position_errors: 0,
2006 tf_errors: 0,
2007 df_errors: 0,
2008 doc_len_errors: 0,
2009 logical_corruptions: 0,
2010 state_valid,
2011 state_status,
2012 duration_ms: started.elapsed().as_millis() as u64,
2013 passed,
2014 status: if passed {
2015 "ok".to_string()
2016 } else {
2017 "repair_recommended".to_string()
2018 },
2019 };
2020 crate::metrics::record_text_consistency_audit(!report.passed);
2021 Ok(report)
2022 }
2023
2024 fn indexed_ids_by_namespace(
2025 &self,
2026 engine: &StorageEngine,
2027 namespace: &str,
2028 ) -> Result<(Vec<u64>, bool)> {
2029 let prefix = namespace_index_prefix(namespace);
2030 let entries = engine.scan_partition_prefix(BackendPartition::NamespaceIndex, &prefix)?;
2031 let mut ids = Vec::new();
2032 let has_index_entries = Self::load_derived_index_state(engine)?.is_some();
2033 crate::metrics::record_derived_prefix_scan();
2034
2035 for (_key, value) in entries {
2036 if let Some(node_id) = decode_node_id(&value) {
2037 ids.push(node_id);
2038 }
2039 }
2040
2041 Ok((ids, has_index_entries))
2042 }
2043
2044 fn indexed_ids_by_filter(
2045 &self,
2046 engine: &StorageEngine,
2047 namespace: &str,
2048 field: &str,
2049 value: &VantaValue,
2050 ) -> Result<(Vec<u64>, bool)> {
2051 let prefix = payload_index_prefix(namespace, field, value)?;
2052 let entries = engine.scan_partition_prefix(BackendPartition::PayloadIndex, &prefix)?;
2053 let mut ids = Vec::new();
2054 let has_index_entries = Self::load_derived_index_state(engine)?.is_some();
2055 crate::metrics::record_derived_prefix_scan();
2056
2057 for (_key, value) in entries {
2058 if let Some(node_id) = decode_node_id(&value) {
2059 ids.push(node_id);
2060 }
2061 }
2062
2063 Ok((ids, has_index_entries))
2064 }
2065
2066 fn ensure_text_index_query_ready(engine: &StorageEngine) -> Result<TextIndexState> {
2067 let state = Self::load_text_index_state(engine).map_err(|_| {
2068 VantaError::Execution(
2069 "text_query requires a current BM25 text index; reopen writable or run rebuild_index"
2070 .to_string(),
2071 )
2072 })?;
2073 let Some(state) = state else {
2074 return Err(VantaError::Execution(
2075 "text_query requires a current BM25 text index; reopen writable or run rebuild_index"
2076 .to_string(),
2077 ));
2078 };
2079 if !Self::text_index_state_matches_spec(&state) {
2080 return Err(VantaError::Execution(
2081 "text_query requires text_index schema v3; reopen writable or run rebuild_index"
2082 .to_string(),
2083 ));
2084 }
2085 Ok(state)
2086 }
2087
2088 fn text_positions_match_phrases(
2089 term_positions: &BTreeMap<String, Vec<u32>>,
2090 phrases: &[Vec<String>],
2091 ) -> bool {
2092 phrases
2093 .iter()
2094 .all(|phrase| Self::text_positions_match_phrase(term_positions, phrase))
2095 }
2096
2097 fn text_positions_match_phrase(
2098 term_positions: &BTreeMap<String, Vec<u32>>,
2099 phrase: &[String],
2100 ) -> bool {
2101 let Some(first_token) = phrase.first() else {
2102 return true;
2103 };
2104 let Some(first_positions) = term_positions.get(first_token) else {
2105 return false;
2106 };
2107 if phrase.len() == 1 {
2108 return !first_positions.is_empty();
2109 }
2110
2111 first_positions.iter().any(|start| {
2112 phrase.iter().enumerate().skip(1).all(|(offset, token)| {
2113 let Some(positions) = term_positions.get(token) else {
2114 return false;
2115 };
2116 positions.contains(&start.saturating_add(offset as u32))
2117 })
2118 })
2119 }
2120
2121 fn lexical_search(
2122 &self,
2123 namespace: &str,
2124 query_text: &str,
2125 filters: &VantaMemoryMetadata,
2126 top_k: usize,
2127 ) -> Result<Vec<VantaMemorySearchHit>> {
2128 let started = Instant::now();
2129 let engine = self.engine_handle()?;
2130 Self::ensure_text_index_query_ready(&engine)?;
2131
2132 if top_k == 0 {
2133 crate::metrics::record_text_lexical_query(0, 0);
2134 return Ok(Vec::new());
2135 }
2136
2137 let query_plan = crate::text_index::query_plan(query_text);
2138 if query_plan.terms.is_empty() {
2139 crate::metrics::record_text_lexical_query(0, 0);
2140 return Ok(Vec::new());
2141 }
2142
2143 let Some(namespace_stats) = Self::load_text_namespace_stats(&engine, namespace)? else {
2144 crate::metrics::record_text_lexical_query(started.elapsed().as_millis() as u64, 0);
2145 return Ok(Vec::new());
2146 };
2147 if namespace_stats.doc_count == 0 {
2148 crate::metrics::record_text_lexical_query(started.elapsed().as_millis() as u64, 0);
2149 return Ok(Vec::new());
2150 }
2151
2152 let doc_count = namespace_stats.doc_count as f32;
2153 let avg_doc_len = if namespace_stats.total_doc_len == 0 {
2154 1.0
2155 } else {
2156 namespace_stats.total_doc_len as f32 / doc_count
2157 };
2158 let mut scores: BTreeMap<u64, f32> = BTreeMap::new();
2159 let mut candidate_positions: BTreeMap<u64, BTreeMap<String, Vec<u32>>> = BTreeMap::new();
2160 let mut doc_stats_cache: BTreeMap<String, crate::text_index::TextDocStats> =
2161 BTreeMap::new();
2162 let mut candidates_scored = 0u64;
2163
2164 for token in query_plan.terms {
2165 let Some(term_stats) = Self::load_text_term_stats(&engine, namespace, &token)? else {
2166 continue;
2167 };
2168 if term_stats.df == 0 {
2169 continue;
2170 }
2171
2172 let df = term_stats.df as f32;
2173 let idf = (1.0 + ((doc_count - df + 0.5) / (df + 0.5))).ln();
2174 let prefix = crate::text_index::posting_prefix(namespace, &token);
2175 for (posting_key, posting_value) in
2176 engine.scan_partition_prefix(BackendPartition::TextIndex, &prefix)?
2177 {
2178 if crate::text_index::is_internal_key(&posting_key) {
2179 continue;
2180 }
2181 let posting = crate::text_index::decode_posting(&posting_value).map_err(|err| {
2182 VantaError::Execution(format!(
2183 "text_query found an unreadable posting; run rebuild_index: {err}"
2184 ))
2185 })?;
2186 let Some(record_key) =
2187 crate::text_index::posting_record_key(namespace, &token, &posting_key)
2188 else {
2189 continue;
2190 };
2191 let doc_stats = if let Some(stats) = doc_stats_cache.get(&record_key) {
2192 stats.clone()
2193 } else {
2194 let Some(stats) = Self::load_text_doc_stats(&engine, namespace, &record_key)?
2195 else {
2196 return Err(VantaError::Execution(
2197 "text_query found posting without document stats; run rebuild_index"
2198 .to_string(),
2199 ));
2200 };
2201 doc_stats_cache.insert(record_key.clone(), stats.clone());
2202 stats
2203 };
2204 if doc_stats.node_id != posting.node_id {
2205 return Err(VantaError::Execution(
2206 "text_query found posting/doc stats mismatch; run rebuild_index"
2207 .to_string(),
2208 ));
2209 }
2210
2211 let tf = posting.tf as f32;
2212 let doc_len = doc_stats.doc_len as f32;
2213 let denominator = tf
2214 + crate::text_index::BM25_K1
2215 * (1.0 - crate::text_index::BM25_B
2216 + crate::text_index::BM25_B * (doc_len / avg_doc_len));
2217 let contribution = idf * ((tf * (crate::text_index::BM25_K1 + 1.0)) / denominator);
2218 scores
2219 .entry(posting.node_id)
2220 .and_modify(|score| *score += contribution)
2221 .or_insert(contribution);
2222 candidate_positions
2223 .entry(posting.node_id)
2224 .or_default()
2225 .insert(token.clone(), posting.positions);
2226 candidates_scored += 1;
2227 }
2228 }
2229
2230 let mut hits = Vec::new();
2231 for (node_id, score) in scores {
2232 let positions_match = candidate_positions
2233 .get(&node_id)
2234 .map(|positions| Self::text_positions_match_phrases(positions, &query_plan.phrases))
2235 .unwrap_or(query_plan.phrases.is_empty());
2236 if !positions_match {
2237 continue;
2238 }
2239 if let Some(node) = engine.get(node_id)? {
2240 if let Some(record) = memory_record_from_node(node) {
2241 if record.namespace == namespace && matches_memory_filters(&record, filters) {
2242 hits.push(VantaMemorySearchHit {
2243 record,
2244 score,
2245 explanation: None,
2246 });
2247 }
2248 }
2249 }
2250 }
2251
2252 hits.sort_by(|a, b| {
2253 b.score
2254 .partial_cmp(&a.score)
2255 .unwrap_or(std::cmp::Ordering::Equal)
2256 .then(a.record.key.cmp(&b.record.key))
2257 .then(a.record.node_id.cmp(&b.record.node_id))
2258 });
2259 hits.truncate(top_k);
2260 crate::metrics::record_text_lexical_query(
2261 started.elapsed().as_millis() as u64,
2262 candidates_scored,
2263 );
2264 Ok(hits)
2265 }
2266
2267 fn vector_memory_search(
2268 &self,
2269 namespace: &str,
2270 query_vector: &[f32],
2271 filters: &VantaMemoryMetadata,
2272 top_k: usize,
2273 distance_metric: DistanceMetric,
2274 ) -> Result<Vec<VantaMemorySearchHit>> {
2275 if query_vector.is_empty() || top_k == 0 {
2276 return Ok(Vec::new());
2277 }
2278
2279 let engine = self.engine_handle()?;
2280
2281 let budget = (top_k.saturating_mul(10)).min(500).max(top_k);
2287 let candidates = {
2288 let hnsw = engine.hnsw.load();
2289 let vs = engine.vector_store.read();
2290 hnsw.search_nearest(query_vector, None, None, u128::MAX, budget, Some(&*vs))
2291 };
2292
2293 let mut hits = Vec::with_capacity(top_k);
2295 for (node_id, raw_score) in candidates {
2296 if hits.len() >= top_k {
2297 break;
2298 }
2299 if let Some(node) = engine.get(node_id)? {
2300 if let Some(record) = memory_record_from_node(node) {
2301 if record.namespace == namespace && matches_memory_filters(&record, filters) {
2302 let score = if distance_metric == DistanceMetric::Euclidean {
2305 raw_score
2307 } else {
2308 raw_score
2309 };
2310 hits.push(VantaMemorySearchHit {
2311 score,
2312 record,
2313 explanation: None,
2314 });
2315 }
2316 }
2317 }
2318 }
2319
2320 if hits.is_empty() && !query_vector.is_empty() {
2323 for record in self.records_for_namespace(namespace, filters)? {
2324 let Some(vector) = record.vector.as_ref() else {
2325 continue;
2326 };
2327 if vector.len() != query_vector.len() {
2328 continue;
2329 }
2330 let score = match distance_metric {
2331 DistanceMetric::Cosine => cosine_sim_f32(query_vector, vector),
2332 DistanceMetric::Euclidean => {
2333 -crate::index::euclidean_distance_squared_f32(query_vector, vector)
2334 }
2335 };
2336 hits.push(VantaMemorySearchHit {
2337 score,
2338 record,
2339 explanation: None,
2340 });
2341 }
2342 Self::sort_memory_hits(&mut hits);
2343 hits.truncate(top_k);
2344 if distance_metric == DistanceMetric::Euclidean {
2345 for hit in hits.iter_mut() {
2346 hit.score = -(-hit.score).max(0.0).sqrt();
2347 }
2348 }
2349 }
2350
2351 Ok(hits)
2352 }
2353
2354 fn sort_memory_hits(hits: &mut [VantaMemorySearchHit]) {
2355 crate::planner::sort_hits(hits);
2356 }
2357
2358 fn hybrid_candidate_budget(top_k: usize) -> usize {
2359 crate::planner::hybrid_candidate_budget(top_k)
2360 }
2361
2362 fn hybrid_search(
2363 &self,
2364 namespace: &str,
2365 query_vector: &[f32],
2366 text_query: &str,
2367 filters: &VantaMemoryMetadata,
2368 top_k: usize,
2369 distance_metric: DistanceMetric,
2370 ) -> Result<Vec<VantaMemorySearchHit>> {
2371 let started = Instant::now();
2372 if top_k == 0 {
2373 crate::metrics::record_hybrid_query(0, 0);
2374 return Ok(Vec::new());
2375 }
2376
2377 let budget = Self::hybrid_candidate_budget(top_k);
2378 let lexical_hits = self.lexical_search(namespace, text_query, filters, budget)?;
2379 let vector_hits =
2380 self.vector_memory_search(namespace, query_vector, filters, budget, distance_metric)?;
2381 let mut hits = Self::fuse_rrf(lexical_hits, vector_hits);
2382 let candidates_fused = hits.len() as u64;
2383 hits.truncate(top_k);
2384 crate::metrics::record_hybrid_query(started.elapsed().as_millis() as u64, candidates_fused);
2385 Ok(hits)
2386 }
2387
2388 fn fuse_rrf(
2389 lexical_hits: Vec<VantaMemorySearchHit>,
2390 vector_hits: Vec<VantaMemorySearchHit>,
2391 ) -> Vec<VantaMemorySearchHit> {
2392 crate::planner::fuse_rrf(lexical_hits, vector_hits)
2393 }
2394
2395 fn records_for_namespace(
2396 &self,
2397 namespace: &str,
2398 filters: &VantaMemoryMetadata,
2399 ) -> Result<Vec<VantaMemoryRecord>> {
2400 let engine = self.engine_handle()?;
2401
2402 let (candidate_ids, has_index_entries) = if let Some((field, value)) = filters.iter().next()
2403 {
2404 self.indexed_ids_by_filter(&engine, namespace, field, value)?
2405 } else {
2406 self.indexed_ids_by_namespace(&engine, namespace)?
2407 };
2408
2409 let mut records = Vec::new();
2410 let mut seen = BTreeSet::new();
2411
2412 for node_id in candidate_ids {
2413 if !seen.insert(node_id) {
2414 continue;
2415 }
2416 if let Some(node) = engine.get(node_id)? {
2417 if let Some(record) = memory_record_from_node(node) {
2418 if record.namespace == namespace && matches_memory_filters(&record, filters) {
2419 records.push(record);
2420 }
2421 }
2422 }
2423 }
2424
2425 if records.is_empty() && !has_index_entries {
2426 crate::metrics::record_derived_full_scan_fallback();
2427 for node in engine.scan_nodes()? {
2428 if let Some(record) = memory_record_from_node(node) {
2429 if record.namespace == namespace && matches_memory_filters(&record, filters) {
2430 records.push(record);
2431 }
2432 }
2433 }
2434 }
2435
2436 records.sort_by(|a, b| a.key.cmp(&b.key).then(a.node_id.cmp(&b.node_id)));
2437 Ok(records)
2438 }
2439
2440 fn put_record_exact(&self, record: VantaMemoryRecord) -> Result<VantaMemoryRecord> {
2441 validate_namespace(&record.namespace)?;
2442 validate_key(&record.key)?;
2443 validate_metadata(&record.metadata)?;
2444
2445 let expected_node_id = memory_node_id(&record.namespace, &record.key);
2446 if record.node_id != expected_node_id {
2447 return Err(VantaError::Execution(format!(
2448 "node_id does not match deterministic namespace/key hash for namespace='{}' key='{}'",
2449 record.namespace, record.key
2450 )));
2451 }
2452
2453 let engine = self.engine_handle()?;
2454 let previous = match engine.get(record.node_id)? {
2455 Some(node) => match memory_record_from_node(node) {
2456 Some(previous)
2457 if previous.namespace == record.namespace && previous.key == record.key =>
2458 {
2459 Some(previous)
2460 }
2461 _ => {
2462 return Err(VantaError::Execution(format!(
2463 "node id collision for namespace='{}' key='{}'",
2464 record.namespace, record.key
2465 )));
2466 }
2467 },
2468 None => None,
2469 };
2470
2471 let node = memory_record_to_node(&record);
2472 engine.insert(&node)?;
2473 self.replace_derived_indexes(&engine, previous.as_ref(), Some(&record))?;
2474
2475 Ok(record)
2476 }
2477
2478 pub fn insert_node(&self, input: VantaNodeInput) -> Result<()> {
2479 let engine = self.engine_handle()?;
2480 let mut node = UnifiedNode::new(input.id);
2481
2482 if let Some(content) = input.content {
2483 node.set_field("content", FieldValue::String(content));
2484 }
2485
2486 for (key, value) in input.fields {
2487 node.set_field(key, value.into());
2488 }
2489
2490 if let Some(vector) = input.vector.filter(|vector| !vector.is_empty()) {
2491 node.vector = VectorRepresentations::Full(vector);
2492 node.flags.set(crate::node::NodeFlags::HAS_VECTOR);
2493 }
2494
2495 engine.insert(&node)
2496 }
2497
2498 pub fn get_node(&self, id: u64) -> Result<Option<VantaNodeRecord>> {
2499 self.engine_handle()?
2500 .get(id)
2501 .map(|node| node.map(Into::into))
2502 }
2503
2504 pub fn delete_node(&self, id: u64, reason: &str) -> Result<()> {
2505 self.engine_handle()?.delete(id, reason)
2506 }
2507
2508 pub fn put_batch(&self, inputs: Vec<VantaMemoryInput>) -> Result<Vec<VantaMemoryRecord>> {
2514 use rayon::prelude::*;
2515
2516 for input in &inputs {
2517 validate_namespace(&input.namespace)?;
2518 validate_key(&input.key)?;
2519 validate_metadata(&input.metadata)?;
2520 }
2521
2522 let results: Vec<Result<VantaMemoryRecord>> = inputs
2523 .into_par_iter()
2524 .map(|input| {
2525 let engine = self.engine_handle()?;
2526 let node_id = memory_node_id(&input.namespace, &input.key);
2527 let existing = match engine.get(node_id)? {
2528 Some(node) => match memory_record_from_node(node) {
2529 Some(record)
2530 if record.namespace == input.namespace && record.key == input.key =>
2531 {
2532 Some(record)
2533 }
2534 _ => {
2535 return Err(VantaError::Execution(format!(
2536 "node id collision for namespace='{}' key='{}'",
2537 input.namespace, input.key
2538 )));
2539 }
2540 },
2541 None => None,
2542 };
2543
2544 let timestamp = now_ms();
2545 let created_at_ms = existing
2546 .as_ref()
2547 .map(|record| record.created_at_ms)
2548 .unwrap_or(timestamp);
2549 let version = existing
2550 .as_ref()
2551 .map(|record| record.version.saturating_add(1))
2552 .unwrap_or(1);
2553 let expires_at_ms = input.ttl_ms.map(|ttl| timestamp.saturating_add(ttl));
2554
2555 let record = VantaMemoryRecord {
2556 namespace: input.namespace,
2557 key: input.key,
2558 payload: input.payload,
2559 metadata: input.metadata,
2560 created_at_ms,
2561 updated_at_ms: timestamp,
2562 version,
2563 node_id,
2564 vector: input.vector.filter(|v| !v.is_empty()),
2565 expires_at_ms,
2566 };
2567 let node = memory_record_to_node(&record);
2568 engine.insert(&node)?;
2569 self.replace_derived_indexes(&engine, existing.as_ref(), Some(&record))?;
2570 Ok(record)
2571 })
2572 .collect();
2573
2574 results.into_iter().collect()
2575 }
2576
2577 pub fn put(&self, input: VantaMemoryInput) -> Result<VantaMemoryRecord> {
2578 validate_namespace(&input.namespace)?;
2579 validate_key(&input.key)?;
2580 validate_metadata(&input.metadata)?;
2581
2582 let engine = self.engine_handle()?;
2583 let node_id = memory_node_id(&input.namespace, &input.key);
2584 let existing = match engine.get(node_id)? {
2585 Some(node) => match memory_record_from_node(node) {
2586 Some(record) if record.namespace == input.namespace && record.key == input.key => {
2587 Some(record)
2588 }
2589 _ => None,
2591 },
2592 None => None,
2593 };
2594
2595 let timestamp = now_ms();
2596 let created_at_ms = existing
2597 .as_ref()
2598 .map(|r| r.created_at_ms)
2599 .unwrap_or(timestamp);
2600 let version = existing
2601 .as_ref()
2602 .map(|r| r.version.saturating_add(1))
2603 .unwrap_or(1);
2604 let expires_at_ms = input.ttl_ms.map(|ttl| timestamp.saturating_add(ttl));
2605
2606 let record = VantaMemoryRecord {
2607 namespace: input.namespace,
2608 key: input.key,
2609 payload: input.payload,
2610 metadata: input.metadata,
2611 created_at_ms,
2612 updated_at_ms: timestamp,
2613 version,
2614 node_id,
2615 vector: input.vector.filter(|v| !v.is_empty()),
2616 expires_at_ms,
2617 };
2618 let node = memory_record_to_node(&record);
2619 engine.insert(&node)?;
2620 self.replace_derived_indexes(&engine, existing.as_ref(), Some(&record))?;
2621
2622 Ok(record)
2623 }
2624
2625 pub fn get(&self, namespace: &str, key: &str) -> Result<Option<VantaMemoryRecord>> {
2626 validate_namespace(namespace)?;
2627 validate_key(key)?;
2628
2629 let node_id = memory_node_id(namespace, key);
2630 let Some(node) = self.engine_handle()?.get(node_id)? else {
2631 return Ok(None);
2632 };
2633
2634 match memory_record_from_node(node) {
2635 Some(record) if record.namespace == namespace && record.key == key => Ok(Some(record)),
2636 Some(_record) => Err(VantaError::Execution(format!(
2637 "node id collision for namespace='{}' key='{}'",
2638 namespace, key
2639 ))),
2640 None => Ok(None),
2641 }
2642 }
2643
2644 pub fn delete(&self, namespace: &str, key: &str) -> Result<bool> {
2645 validate_namespace(namespace)?;
2646 validate_key(key)?;
2647
2648 let Some(existing) = self.get(namespace, key)? else {
2649 return Ok(false);
2650 };
2651
2652 let node_id = memory_node_id(namespace, key);
2653 let engine = self.engine_handle()?;
2654 engine.delete(node_id, "memory delete")?;
2655 self.replace_derived_indexes(&engine, Some(&existing), None)?;
2656 Ok(true)
2657 }
2658
2659 pub fn list_namespaces(&self) -> Result<Vec<String>> {
2660 let engine = self.engine_handle()?;
2661 let mut namespaces = BTreeSet::new();
2662 let entries = engine.scan_partition(BackendPartition::NamespaceIndex)?;
2663
2664 if entries.is_empty() {
2665 for node in engine.scan_nodes()? {
2666 if let Some(record) = memory_record_from_node(node) {
2667 namespaces.insert(record.namespace);
2668 }
2669 }
2670 } else {
2671 for (key, _value) in entries {
2672 if let Some(separator) = key.iter().position(|byte| *byte == 0) {
2673 if let Ok(namespace) = String::from_utf8(key[..separator].to_vec()) {
2674 namespaces.insert(namespace);
2675 }
2676 }
2677 }
2678 }
2679
2680 Ok(namespaces.into_iter().collect())
2681 }
2682
2683 pub fn list(
2684 &self,
2685 namespace: &str,
2686 options: VantaMemoryListOptions,
2687 ) -> Result<VantaMemoryListPage> {
2688 validate_namespace(namespace)?;
2689 validate_metadata(&options.filters)?;
2690
2691 let records = self.records_for_namespace(namespace, &options.filters)?;
2692
2693 let start = options.cursor.unwrap_or(0).min(records.len());
2694 let limit = options.limit.max(1);
2695 let end = start.saturating_add(limit).min(records.len());
2696 let next_cursor = (end < records.len()).then_some(end);
2697
2698 Ok(VantaMemoryListPage {
2699 records: records[start..end].to_vec(),
2700 next_cursor,
2701 })
2702 }
2703
2704 pub fn search(&self, request: VantaMemorySearchRequest) -> Result<Vec<VantaMemorySearchHit>> {
2705 validate_namespace(&request.namespace)?;
2706 validate_metadata(&request.filters)?;
2707
2708 let text_query = crate::planner::trimmed_text_query(&request);
2709 let has_vector = !request.query_vector.is_empty();
2710
2711 if request.top_k == 0 {
2712 return Ok(Vec::new());
2713 }
2714
2715 if request.explain {
2716 let engine = self.engine_handle()?;
2717 let (hits, text_ranks, vector_ranks) = match (text_query, has_vector) {
2718 (Some(text_query), true) => {
2719 let budget = Self::hybrid_candidate_budget(request.top_k);
2720 let lexical_hits = self.lexical_search(
2721 &request.namespace,
2722 text_query,
2723 &request.filters,
2724 budget,
2725 )?;
2726 let vector_hits = self.vector_memory_search(
2727 &request.namespace,
2728 &request.query_vector,
2729 &request.filters,
2730 budget,
2731 request.distance_metric,
2732 )?;
2733 let text_ranks = Self::debug_rank_map(&lexical_hits);
2734 let vector_ranks = Self::debug_rank_map(&vector_hits);
2735 let (mut hits, _report) =
2736 crate::planner::fuse_rrf_with_report(lexical_hits, vector_hits);
2737 hits.truncate(request.top_k);
2738 (hits, text_ranks, vector_ranks)
2739 }
2740 (Some(text_query), false) => {
2741 let hits = self.lexical_search(
2742 &request.namespace,
2743 text_query,
2744 &request.filters,
2745 request.top_k,
2746 )?;
2747 let text_ranks = Self::debug_rank_map(&hits);
2748 (hits, text_ranks, BTreeMap::new())
2749 }
2750 (None, true) => {
2751 let hits = self.vector_memory_search(
2752 &request.namespace,
2753 &request.query_vector,
2754 &request.filters,
2755 request.top_k,
2756 request.distance_metric,
2757 )?;
2758 let vector_ranks = Self::debug_rank_map(&hits);
2759 (hits, BTreeMap::new(), vector_ranks)
2760 }
2761 (None, false) => (Vec::new(), BTreeMap::new(), BTreeMap::new()),
2762 };
2763
2764 let explained_hits = hits
2765 .into_iter()
2766 .map(|mut hit| {
2767 let explanation = Self::debug_explain_hit(
2768 &engine,
2769 hit.clone(),
2770 text_query,
2771 &text_ranks,
2772 &vector_ranks,
2773 )?;
2774 hit.explanation = Some(explanation);
2775 Ok(hit)
2776 })
2777 .collect::<Result<Vec<_>>>()?;
2778
2779 return Ok(explained_hits);
2780 }
2781
2782 match (text_query, has_vector) {
2783 (Some(text_query), true) => {
2784 crate::metrics::record_planner_hybrid_query();
2785 self.hybrid_search(
2786 &request.namespace,
2787 &request.query_vector,
2788 text_query,
2789 &request.filters,
2790 request.top_k,
2791 request.distance_metric,
2792 )
2793 }
2794 (Some(text_query), false) => {
2795 crate::metrics::record_planner_text_only_query();
2796 self.lexical_search(
2797 &request.namespace,
2798 text_query,
2799 &request.filters,
2800 request.top_k,
2801 )
2802 }
2803 (None, true) => {
2804 crate::metrics::record_planner_vector_only_query();
2805 self.vector_memory_search(
2806 &request.namespace,
2807 &request.query_vector,
2808 &request.filters,
2809 request.top_k,
2810 request.distance_metric,
2811 )
2812 }
2813 (None, false) => Ok(Vec::new()),
2814 }
2815 }
2816
2817 pub fn rebuild_index(&self) -> Result<VantaIndexRebuildReport> {
2818 if self.config.read_only {
2819 return Err(VantaError::Execution(
2820 "rebuild_index is not available when VantaDB is opened read-only".to_string(),
2821 ));
2822 }
2823 let report = self.engine_handle()?.rebuild_vector_index()?;
2824 let derived = self.rebuild_derived_indexes_with_report()?;
2825 self.rebuild_text_index_with_report()?;
2826 let mut report: VantaIndexRebuildReport = report.into();
2827 report.derived_rebuild_ms = derived.duration_ms;
2828 Ok(report)
2829 }
2830
2831 pub fn compact_layout(&self) -> Result<u64> {
2840 if self.config.read_only {
2841 return Err(VantaError::Execution(
2842 "compact_layout is not available when VantaDB is opened read-only".to_string(),
2843 ));
2844 }
2845 self.engine_handle()?.compact_layout_bfs()
2846 }
2847
2848 pub fn export_namespace(
2849 &self,
2850 path: impl AsRef<Path>,
2851 namespace: &str,
2852 ) -> Result<VantaExportReport> {
2853 validate_namespace(namespace)?;
2854 let started = Instant::now();
2855 let records = self.records_for_namespace(namespace, &VantaMemoryMetadata::new())?;
2856 self.write_export_file(path.as_ref(), records, vec![namespace.to_string()], started)
2857 }
2858
2859 pub fn export_all(&self, path: impl AsRef<Path>) -> Result<VantaExportReport> {
2860 let started = Instant::now();
2861 let namespaces = self.list_namespaces()?;
2862 let mut records = Vec::new();
2863 for namespace in &namespaces {
2864 records.extend(self.records_for_namespace(namespace, &VantaMemoryMetadata::new())?);
2865 }
2866 self.write_export_file(path.as_ref(), records, namespaces, started)
2867 }
2868
2869 fn write_export_file(
2870 &self,
2871 path: &Path,
2872 records: Vec<VantaMemoryRecord>,
2873 namespaces: Vec<String>,
2874 started: Instant,
2875 ) -> Result<VantaExportReport> {
2876 if let Some(parent) = path.parent() {
2877 std::fs::create_dir_all(parent).map_err(VantaError::IoError)?;
2878 }
2879
2880 let file = File::create(path).map_err(VantaError::IoError)?;
2881 let mut writer = BufWriter::new(file);
2882 let records_exported = records.len() as u64;
2883
2884 for record in records {
2885 let line = export_line_from_record(record);
2886 serde_json::to_writer(&mut writer, &line)
2887 .map_err(|err| VantaError::SerializationError(err.to_string()))?;
2888 writer.write_all(b"\n").map_err(VantaError::IoError)?;
2889 }
2890 writer.flush().map_err(VantaError::IoError)?;
2891 crate::metrics::record_export(records_exported);
2892
2893 Ok(VantaExportReport {
2894 records_exported,
2895 namespaces,
2896 path: path.to_string_lossy().into_owned(),
2897 duration_ms: started.elapsed().as_millis() as u64,
2898 })
2899 }
2900
2901 pub fn import_records(&self, records: Vec<VantaMemoryRecord>) -> Result<VantaImportReport> {
2902 if self.config.read_only {
2903 return Err(VantaError::Execution(
2904 "import_records is not available when VantaDB is opened read-only".to_string(),
2905 ));
2906 }
2907 let started = Instant::now();
2908 let mut report = VantaImportReport {
2909 inserted: 0,
2910 updated: 0,
2911 skipped: 0,
2912 errors: 0,
2913 duration_ms: 0,
2914 };
2915
2916 for record in records {
2917 let existed = matches!(self.get(&record.namespace, &record.key), Ok(Some(_)));
2918 match self.put_record_exact(record) {
2919 Ok(_) if existed => report.updated += 1,
2920 Ok(_) => report.inserted += 1,
2921 Err(_) => report.errors += 1,
2922 }
2923 }
2924
2925 self.rebuild_derived_indexes()?;
2926 self.rebuild_text_index()?;
2927 report.duration_ms = started.elapsed().as_millis() as u64;
2928 crate::metrics::record_import(report.inserted + report.updated, report.errors);
2929 Ok(report)
2930 }
2931
2932 pub fn import_file(&self, path: impl AsRef<Path>) -> Result<VantaImportReport> {
2933 if self.config.read_only {
2934 return Err(VantaError::Execution(
2935 "import_file is not available when VantaDB is opened read-only".to_string(),
2936 ));
2937 }
2938 let started = Instant::now();
2939 let file = File::open(path.as_ref()).map_err(VantaError::IoError)?;
2940 let reader = BufReader::new(file);
2941 let mut records = Vec::new();
2942 let mut skipped = 0u64;
2943 let mut errors = 0u64;
2944
2945 for line in reader.lines() {
2946 let line = line.map_err(VantaError::IoError)?;
2947 if line.trim().is_empty() {
2948 skipped += 1;
2949 continue;
2950 }
2951
2952 match serde_json::from_str::<VantaMemoryExportLine>(&line)
2953 .map_err(|err| VantaError::SerializationError(err.to_string()))
2954 .and_then(record_from_export_line)
2955 {
2956 Ok(record) => records.push(record),
2957 Err(_) => errors += 1,
2958 }
2959 }
2960
2961 let mut report = self.import_records(records)?;
2962 report.skipped += skipped;
2963 report.errors += errors;
2964 if errors > 0 {
2965 crate::metrics::record_import(0, errors);
2966 }
2967 report.duration_ms = started.elapsed().as_millis() as u64;
2968 Ok(report)
2969 }
2970
2971 pub fn audit_text_index(&self, namespace: Option<&str>) -> Result<VantaTextIndexAuditReport> {
2978 if let Some(namespace) = namespace {
2979 validate_namespace(namespace)?;
2980 }
2981 let engine = self.engine_handle()?;
2982 Self::build_text_index_audit_report_shallow(&engine, namespace)
2983 }
2984
2985 pub fn audit_text_index_deep(
2990 &self,
2991 namespace: Option<&str>,
2992 ) -> Result<VantaTextIndexAuditReport> {
2993 if let Some(namespace) = namespace {
2994 validate_namespace(namespace)?;
2995 }
2996 let engine = self.engine_handle()?;
2997 Self::build_text_index_audit_report_deep(&engine, namespace)
2998 }
2999
3000 pub fn repair_text_index(&self) -> Result<VantaTextIndexRepairReport> {
3003 if self.config.read_only {
3004 return Err(VantaError::Execution(
3005 "repair_text_index is not available when VantaDB is opened read-only".to_string(),
3006 ));
3007 }
3008 crate::metrics::record_text_index_repair();
3009 let report = self.rebuild_text_index_with_report()?;
3010 Ok(VantaTextIndexRepairReport {
3011 record_count: report.record_count,
3012 posting_entries: report.posting_entries,
3013 doc_stats_entries: report.doc_stats_entries,
3014 term_stats_entries: report.term_stats_entries,
3015 namespace_stats_entries: report.namespace_stats_entries,
3016 duration_ms: report.duration_ms,
3017 success: true,
3018 })
3019 }
3020
3021 pub fn operational_metrics(&self) -> VantaOperationalMetrics {
3022 if let Ok(engine) = self.engine_handle() {
3023 let stats = engine.get_memory_stats();
3024 crate::metrics::record_memory_breakdown(
3025 stats.node_count,
3026 stats.logical_bytes,
3027 stats.physical_rss,
3028 stats.cache_entries as u64,
3029 0,
3030 );
3031 }
3032 crate::metrics::operational_metrics_snapshot().into()
3033 }
3034
3035 pub fn search_vector(&self, vector: &[f32], top_k: usize) -> Result<Vec<VantaSearchHit>> {
3039 if vector.is_empty() || top_k == 0 {
3040 return Ok(Vec::new());
3041 }
3042 let engine = self.engine_handle()?;
3043 let results = {
3044 let hnsw = engine.hnsw.load();
3045 let vs = engine.vector_store.read();
3046 hnsw.search_nearest(
3047 vector,
3048 None, None, u128::MAX, top_k,
3052 Some(&*vs), )
3054 };
3055 Ok(results
3056 .into_iter()
3057 .map(|(node_id, distance)| VantaSearchHit { node_id, distance })
3058 .collect())
3059 }
3060
3061 pub fn flush(&self) -> Result<()> {
3066 if self.config.read_only {
3067 return Err(VantaError::Execution(
3068 "flush is not available when VantaDB is opened read-only".to_string(),
3069 ));
3070 }
3071 self.engine_handle()?.flush()
3072 }
3073
3074 pub fn compact_wal(&self) -> Result<()> {
3080 if self.config.read_only {
3081 return Err(VantaError::Execution(
3082 "compact_wal is not available when VantaDB is opened read-only".to_string(),
3083 ));
3084 }
3085 self.engine_handle()?.compact_wal()
3086 }
3087
3088 pub fn purge_expired(&self) -> Result<u64> {
3092 if self.config.read_only {
3093 return Err(VantaError::Execution(
3094 "purge_expired is not available when VantaDB is opened read-only".to_string(),
3095 ));
3096 }
3097 let engine = self.engine_handle()?;
3098 let now = now_ms();
3099 let mut to_delete: Vec<(String, String, u64)> = Vec::new();
3100
3101 for node in engine.scan_nodes()? {
3103 if !node.is_alive() {
3104 continue;
3105 }
3106 let namespace = match node.get_field(FIELD_NAMESPACE) {
3107 Some(crate::node::FieldValue::String(ns)) => ns.clone(),
3108 _ => continue,
3109 };
3110 let key = match node.get_field(FIELD_KEY) {
3111 Some(crate::node::FieldValue::String(k)) => k.clone(),
3112 _ => continue,
3113 };
3114 let expires = match node.get_field(FIELD_EXPIRES_AT_MS) {
3115 Some(crate::node::FieldValue::Int(ms)) if *ms > 0 => *ms as u64,
3116 _ => continue,
3117 };
3118 if now > expires {
3119 to_delete.push((namespace, key, node.id));
3120 }
3121 }
3122
3123 let count = to_delete.len() as u64;
3124 for (_, _, node_id) in &to_delete {
3125 engine.delete(*node_id, "purge_expired")?;
3126 self.replace_derived_indexes(&engine, None, None)?;
3127 }
3128
3129 Ok(count)
3130 }
3131
3132 pub fn capabilities(&self) -> VantaCapabilities {
3134 VantaCapabilities {
3135 runtime_profile: VantaRuntimeProfile::Performance,
3136 persistence: true,
3137 vector_search: true,
3138 iql_queries: true,
3139 read_only: self.config.read_only,
3140 }
3141 }
3142
3143 pub fn add_edge(
3145 &self,
3146 source_id: u64,
3147 target_id: u64,
3148 label: &str,
3149 weight: Option<f32>,
3150 ) -> Result<()> {
3151 let engine = self.engine_handle()?;
3152 let mut node = engine
3153 .get(source_id)?
3154 .ok_or(VantaError::NodeNotFound(source_id))?;
3155 node.edges.push(crate::node::Edge {
3156 target: target_id,
3157 label: label.to_string(),
3158 weight: weight.unwrap_or(1.0),
3159 });
3160 engine.insert(&node)
3161 }
3162
3163 pub fn close(&self) -> Result<()> {
3165 if let Err(e) = self.flush() {
3166 tracing::warn!("flush failed: {e}");
3167 }
3168 let mut guard = self.engine.write();
3169 *guard = None;
3170 Ok(())
3171 }
3172
3173 pub fn query(&self, query: &str) -> Result<VantaQueryResult> {
3175 let engine = self.engine_handle()?;
3176 let executor = Executor::new(&engine);
3177 let result = executor.execute_hybrid(query)?;
3178 Ok(result.into())
3179 }
3180
3181 pub fn generate_snippet(
3191 &self,
3192 payload: &str,
3193 text_query: &str,
3194 with_highlighting: bool,
3195 ) -> Option<String> {
3196 Self::generate_snippet_with_highlighting(payload, text_query, with_highlighting)
3197 }
3198
3199 #[cfg(debug_assertions)]
3200 #[doc(hidden)]
3201 pub fn debug_memory_breakdown(&self) -> serde_json::Value {
3202 let metrics = self.operational_metrics();
3203 serde_json::json!({
3204 "process_rss_bytes": metrics.process_rss_bytes,
3205 "process_virtual_bytes": metrics.process_virtual_bytes,
3206 "hnsw_nodes_count": metrics.hnsw_nodes_count,
3207 "hnsw_logical_bytes": metrics.hnsw_logical_bytes,
3208 "mmap_resident_bytes": metrics.mmap_resident_bytes,
3209 "volatile_cache_entries": metrics.volatile_cache_entries,
3210 "volatile_cache_cap_bytes": metrics.volatile_cache_cap_bytes,
3211 })
3212 }
3213
3214 #[cfg(debug_assertions)]
3215 #[doc(hidden)]
3216 pub fn debug_corrupt_derived_index_state_for_tests(&self) -> Result<()> {
3217 let engine = self.engine_handle()?;
3218 engine.put_to_partition(
3219 BackendPartition::InternalMetadata,
3220 DERIVED_INDEX_STATE_KEY,
3221 b"corrupt-derived-index-state",
3222 )
3223 }
3224
3225 #[cfg(debug_assertions)]
3226 #[doc(hidden)]
3227 pub fn debug_clear_derived_indexes_for_tests(&self) -> Result<()> {
3228 let engine = self.engine_handle()?;
3229 let mut ops = Vec::new();
3230 for (key, _value) in engine.scan_partition(BackendPartition::NamespaceIndex)? {
3231 ops.push(BackendWriteOp::Delete {
3232 partition: BackendPartition::NamespaceIndex,
3233 key,
3234 });
3235 }
3236 for (key, _value) in engine.scan_partition(BackendPartition::PayloadIndex)? {
3237 ops.push(BackendWriteOp::Delete {
3238 partition: BackendPartition::PayloadIndex,
3239 key,
3240 });
3241 }
3242 engine.write_backend_batch(ops)
3243 }
3244
3245 #[cfg(debug_assertions)]
3246 #[doc(hidden)]
3247 pub fn debug_corrupt_text_index_state_for_tests(&self) -> Result<()> {
3248 let engine = self.engine_handle()?;
3249 engine.put_to_partition(
3250 BackendPartition::InternalMetadata,
3251 TEXT_INDEX_STATE_KEY,
3252 b"corrupt-text-index-state",
3253 )
3254 }
3255
3256 #[cfg(debug_assertions)]
3257 #[doc(hidden)]
3258 pub fn debug_clear_text_index_for_tests(&self) -> Result<()> {
3259 let engine = self.engine_handle()?;
3260 let mut ops = Vec::new();
3261 for (key, _value) in engine.scan_partition(BackendPartition::TextIndex)? {
3262 ops.push(BackendWriteOp::Delete {
3263 partition: BackendPartition::TextIndex,
3264 key,
3265 });
3266 }
3267 engine.write_backend_batch(ops)
3268 }
3269
3270 #[cfg(debug_assertions)]
3271 #[doc(hidden)]
3272 pub fn debug_corrupt_text_index_posting_tf_for_tests(
3273 &self,
3274 namespace: &str,
3275 token: &str,
3276 key: &str,
3277 new_tf: u32,
3278 ) -> Result<()> {
3279 let engine = self.engine_handle()?;
3280 let pkey = crate::text_index::posting_key(namespace, token, key);
3281 let Some(bytes) = engine.get_from_partition(BackendPartition::TextIndex, &pkey)? else {
3282 return Err(VantaError::Execution("posting not found".to_string()));
3283 };
3284 let posting = crate::text_index::decode_posting(&bytes)?;
3285 let val = crate::text_index::posting_value(posting.node_id, new_tf, &posting.positions)?;
3286 engine.put_to_partition(BackendPartition::TextIndex, &pkey, &val)
3287 }
3288
3289 #[cfg(debug_assertions)]
3290 #[doc(hidden)]
3291 pub fn debug_corrupt_text_index_posting_positions_for_tests(
3292 &self,
3293 namespace: &str,
3294 token: &str,
3295 key: &str,
3296 new_positions: Vec<u32>,
3297 ) -> Result<()> {
3298 let engine = self.engine_handle()?;
3299 let pkey = crate::text_index::posting_key(namespace, token, key);
3300 let Some(bytes) = engine.get_from_partition(BackendPartition::TextIndex, &pkey)? else {
3301 return Err(VantaError::Execution("posting not found".to_string()));
3302 };
3303 let posting = crate::text_index::decode_posting(&bytes)?;
3304 let val = crate::text_index::posting_value(posting.node_id, posting.tf, &new_positions)?;
3305 engine.put_to_partition(BackendPartition::TextIndex, &pkey, &val)
3306 }
3307
3308 #[cfg(debug_assertions)]
3309 #[doc(hidden)]
3310 pub fn debug_corrupt_text_index_term_stats_for_tests(
3311 &self,
3312 namespace: &str,
3313 token: &str,
3314 new_df: u64,
3315 ) -> Result<()> {
3316 let engine = self.engine_handle()?;
3317 let skey = crate::text_index::term_stats_key(namespace, token);
3318 let val = crate::text_index::term_stats_value(new_df)?;
3319 engine.put_to_partition(BackendPartition::TextIndex, &skey, &val)
3320 }
3321
3322 #[cfg(debug_assertions)]
3323 #[doc(hidden)]
3324 pub fn debug_corrupt_text_index_doc_stats_for_tests(
3325 &self,
3326 namespace: &str,
3327 key: &str,
3328 new_doc_len: u32,
3329 ) -> Result<()> {
3330 let engine = self.engine_handle()?;
3331 let dkey = crate::text_index::doc_stats_key(namespace, key);
3332 let Some(bytes) = engine.get_from_partition(BackendPartition::TextIndex, &dkey)? else {
3333 return Err(VantaError::Execution("doc stats not found".to_string()));
3334 };
3335 let stats = crate::text_index::decode_doc_stats(&bytes)?;
3336 let val = crate::text_index::doc_stats_value(stats.node_id, new_doc_len)?;
3337 engine.put_to_partition(BackendPartition::TextIndex, &dkey, &val)
3338 }
3339
3340 #[cfg(debug_assertions)]
3341 #[doc(hidden)]
3342 pub fn debug_text_index_posting_keys_for_tests(&self) -> Result<Vec<Vec<u8>>> {
3343 let engine = self.engine_handle()?;
3344 let mut keys: Vec<Vec<u8>> = engine
3345 .scan_partition(BackendPartition::TextIndex)?
3346 .into_iter()
3347 .map(|(key, _value)| key)
3348 .filter(|key| !crate::text_index::is_internal_key(key))
3349 .collect();
3350 keys.sort();
3351 Ok(keys)
3352 }
3353
3354 #[cfg(debug_assertions)]
3355 #[doc(hidden)]
3356 pub fn debug_text_index_posting_for_tests(
3357 &self,
3358 namespace: &str,
3359 token: &str,
3360 key: &str,
3361 ) -> Result<Option<(u64, u32)>> {
3362 let engine = self.engine_handle()?;
3363 let Some(bytes) = engine.get_from_partition(
3364 BackendPartition::TextIndex,
3365 &crate::text_index::posting_key(namespace, token, key),
3366 )?
3367 else {
3368 return Ok(None);
3369 };
3370 let posting = crate::text_index::decode_posting(&bytes)?;
3371 Ok(Some((posting.node_id, posting.tf)))
3372 }
3373
3374 #[cfg(debug_assertions)]
3375 #[doc(hidden)]
3376 pub fn debug_text_index_audit_for_tests(&self) -> Result<VantaTextIndexAuditReport> {
3377 self.audit_text_index_deep(None)
3378 }
3379
3380 #[cfg(debug_assertions)]
3381 #[doc(hidden)]
3382 pub fn debug_memory_search_plan_for_tests(
3383 &self,
3384 request: VantaMemorySearchRequest,
3385 ) -> Result<VantaMemorySearchDebugReport> {
3386 validate_namespace(&request.namespace)?;
3387 validate_metadata(&request.filters)?;
3388
3389 let text_query = crate::planner::trimmed_text_query(&request);
3390 let has_vector = !request.query_vector.is_empty();
3391 if request.top_k == 0 {
3392 return Ok(VantaMemorySearchDebugReport {
3393 route: "empty".to_string(),
3394 budget: 0,
3395 text_candidates: 0,
3396 vector_candidates: 0,
3397 fused_candidates: 0,
3398 top_identities: Vec::new(),
3399 });
3400 }
3401
3402 match (text_query, has_vector) {
3403 (Some(text_query), true) => {
3404 let budget = Self::hybrid_candidate_budget(request.top_k);
3405 let lexical_hits =
3406 self.lexical_search(&request.namespace, text_query, &request.filters, budget)?;
3407 let vector_hits = self.vector_memory_search(
3408 &request.namespace,
3409 &request.query_vector,
3410 &request.filters,
3411 budget,
3412 request.distance_metric,
3413 )?;
3414 let text_candidates = lexical_hits.len();
3415 let vector_candidates = vector_hits.len();
3416 let mut fused_hits = Self::fuse_rrf(lexical_hits, vector_hits);
3417 let fused_candidates = fused_hits.len();
3418 fused_hits.truncate(request.top_k);
3419 Ok(VantaMemorySearchDebugReport {
3420 route: "hybrid".to_string(),
3421 budget,
3422 text_candidates,
3423 vector_candidates,
3424 fused_candidates,
3425 top_identities: Self::debug_hit_identities(&fused_hits),
3426 })
3427 }
3428 (Some(text_query), false) => {
3429 let hits = self.lexical_search(
3430 &request.namespace,
3431 text_query,
3432 &request.filters,
3433 request.top_k,
3434 )?;
3435 Ok(VantaMemorySearchDebugReport {
3436 route: "text-only".to_string(),
3437 budget: request.top_k,
3438 text_candidates: hits.len(),
3439 vector_candidates: 0,
3440 fused_candidates: hits.len(),
3441 top_identities: Self::debug_hit_identities(&hits),
3442 })
3443 }
3444 (None, true) => {
3445 let hits = self.vector_memory_search(
3446 &request.namespace,
3447 &request.query_vector,
3448 &request.filters,
3449 request.top_k,
3450 request.distance_metric,
3451 )?;
3452 Ok(VantaMemorySearchDebugReport {
3453 route: "vector-only".to_string(),
3454 budget: request.top_k,
3455 text_candidates: 0,
3456 vector_candidates: hits.len(),
3457 fused_candidates: hits.len(),
3458 top_identities: Self::debug_hit_identities(&hits),
3459 })
3460 }
3461 (None, false) => Ok(VantaMemorySearchDebugReport {
3462 route: "empty".to_string(),
3463 budget: 0,
3464 text_candidates: 0,
3465 vector_candidates: 0,
3466 fused_candidates: 0,
3467 top_identities: Vec::new(),
3468 }),
3469 }
3470 }
3471
3472 pub fn explain_memory_search(
3473 &self,
3474 request: VantaMemorySearchRequest,
3475 ) -> Result<VantaSearchExplanation> {
3476 validate_namespace(&request.namespace)?;
3477 validate_metadata(&request.filters)?;
3478
3479 let text_query = request
3480 .text_query
3481 .as_deref()
3482 .map(str::trim)
3483 .filter(|text| !text.is_empty());
3484 let has_vector = !request.query_vector.is_empty();
3485 if request.top_k == 0 {
3486 return Ok(VantaSearchExplanation {
3487 route: "empty".to_string(),
3488 hits: Vec::new(),
3489 fusion_report: None,
3490 });
3491 }
3492
3493 let engine = self.engine_handle()?;
3494 #[allow(clippy::type_complexity)]
3495 let (route, hits, text_ranks, vector_ranks, fusion_report): (
3496 String,
3497 Vec<VantaMemorySearchHit>,
3498 std::collections::BTreeMap<(String, String), usize>,
3499 std::collections::BTreeMap<(String, String), usize>,
3500 Option<VantaHybridFusionReport>,
3501 ) = match (text_query, has_vector) {
3502 (Some(text_query), true) => {
3503 let budget = Self::hybrid_candidate_budget(request.top_k);
3504 let lexical_hits =
3505 self.lexical_search(&request.namespace, text_query, &request.filters, budget)?;
3506 let vector_hits = self.vector_memory_search(
3507 &request.namespace,
3508 &request.query_vector,
3509 &request.filters,
3510 budget,
3511 request.distance_metric,
3512 )?;
3513 let text_ranks = Self::debug_rank_map(&lexical_hits);
3514 let vector_ranks = Self::debug_rank_map(&vector_hits);
3515 let (mut hits, report) =
3516 crate::planner::fuse_rrf_with_report(lexical_hits, vector_hits);
3517 hits.truncate(request.top_k);
3518 (
3519 "hybrid".to_string(),
3520 hits,
3521 text_ranks,
3522 vector_ranks,
3523 Some(report),
3524 )
3525 }
3526 (Some(text_query), false) => {
3527 let hits = self.lexical_search(
3528 &request.namespace,
3529 text_query,
3530 &request.filters,
3531 request.top_k,
3532 )?;
3533 let text_ranks = Self::debug_rank_map(&hits);
3534 (
3535 "text-only".to_string(),
3536 hits,
3537 text_ranks,
3538 BTreeMap::new(),
3539 None,
3540 )
3541 }
3542 (None, true) => {
3543 let hits = self.vector_memory_search(
3544 &request.namespace,
3545 &request.query_vector,
3546 &request.filters,
3547 request.top_k,
3548 request.distance_metric,
3549 )?;
3550 let vector_ranks = Self::debug_rank_map(&hits);
3551 (
3552 "vector-only".to_string(),
3553 hits,
3554 BTreeMap::new(),
3555 vector_ranks,
3556 None,
3557 )
3558 }
3559 (None, false) => {
3560 return Ok(VantaSearchExplanation {
3561 route: "empty".to_string(),
3562 hits: Vec::new(),
3563 fusion_report: None,
3564 });
3565 }
3566 };
3567
3568 let explained_hits = hits
3569 .into_iter()
3570 .map(|hit| {
3571 Self::debug_explain_hit(&engine, hit, text_query, &text_ranks, &vector_ranks)
3572 })
3573 .collect::<Result<Vec<_>>>()?;
3574
3575 Ok(VantaSearchExplanation {
3576 route,
3577 hits: explained_hits,
3578 fusion_report,
3579 })
3580 }
3581
3582 #[cfg(debug_assertions)]
3583 fn debug_hit_identities(hits: &[VantaMemorySearchHit]) -> Vec<String> {
3584 hits.iter()
3585 .map(|hit| format!("{}\0{}", hit.record.namespace, hit.record.key))
3586 .collect()
3587 }
3588
3589 fn debug_rank_map(hits: &[VantaMemorySearchHit]) -> BTreeMap<(String, String), usize> {
3590 hits.iter()
3591 .enumerate()
3592 .map(|(index, hit)| {
3593 (
3594 (hit.record.namespace.clone(), hit.record.key.clone()),
3595 index + 1,
3596 )
3597 })
3598 .collect()
3599 }
3600
3601 fn debug_explain_hit(
3602 engine: &StorageEngine,
3603 hit: VantaMemorySearchHit,
3604 text_query: Option<&str>,
3605 text_ranks: &BTreeMap<(String, String), usize>,
3606 vector_ranks: &BTreeMap<(String, String), usize>,
3607 ) -> Result<VantaSearchExplanationHit> {
3608 let identity_tuple = (hit.record.namespace.clone(), hit.record.key.clone());
3609 let identity = format!("{}\0{}", hit.record.namespace, hit.record.key);
3610 let bm25_terms = if let Some(text_query) = text_query {
3611 Self::debug_bm25_terms_for_record(engine, &hit.record, text_query)?
3612 } else {
3613 Vec::new()
3614 };
3615 let matched_tokens = bm25_terms
3616 .iter()
3617 .map(|term| term.token.clone())
3618 .collect::<Vec<_>>();
3619 let matched_phrases = if let Some(text_query) = text_query {
3620 Self::debug_matched_phrases_for_record(engine, &hit.record, text_query)?
3621 } else {
3622 Vec::new()
3623 };
3624 let snippet = text_query.and_then(|query| Self::debug_snippet(&hit.record.payload, query));
3625
3626 Ok(VantaSearchExplanationHit {
3627 identity,
3628 score: hit.score,
3629 snippet,
3630 matched_tokens,
3631 matched_phrases,
3632 bm25_terms,
3633 rrf_text_rank: text_ranks.get(&identity_tuple).copied(),
3634 rrf_vector_rank: vector_ranks.get(&identity_tuple).copied(),
3635 })
3636 }
3637
3638 fn debug_bm25_terms_for_record(
3639 engine: &StorageEngine,
3640 record: &VantaMemoryRecord,
3641 text_query: &str,
3642 ) -> Result<Vec<VantaBm25TermContribution>> {
3643 let query_plan = crate::text_index::query_plan(text_query);
3644 if query_plan.terms.is_empty() {
3645 return Ok(Vec::new());
3646 }
3647 let Some(namespace_stats) = Self::load_text_namespace_stats(engine, &record.namespace)?
3648 else {
3649 return Ok(Vec::new());
3650 };
3651 let Some(doc_stats) = Self::load_text_doc_stats(engine, &record.namespace, &record.key)?
3652 else {
3653 return Ok(Vec::new());
3654 };
3655 if namespace_stats.doc_count == 0 {
3656 return Ok(Vec::new());
3657 }
3658
3659 let doc_count = namespace_stats.doc_count as f32;
3660 let avg_doc_len = if namespace_stats.total_doc_len == 0 {
3661 1.0
3662 } else {
3663 namespace_stats.total_doc_len as f32 / doc_count
3664 };
3665 let doc_len = doc_stats.doc_len as f32;
3666 let mut terms = Vec::new();
3667
3668 for token in query_plan.terms {
3669 let Some(term_stats) = Self::load_text_term_stats(engine, &record.namespace, &token)?
3670 else {
3671 continue;
3672 };
3673 let Some(posting_value) = engine.get_from_partition(
3674 BackendPartition::TextIndex,
3675 &crate::text_index::posting_key(&record.namespace, &token, &record.key),
3676 )?
3677 else {
3678 continue;
3679 };
3680 let posting = crate::text_index::decode_posting(&posting_value)?;
3681 let df = term_stats.df as f32;
3682 let idf = (1.0 + ((doc_count - df + 0.5) / (df + 0.5))).ln();
3683 let tf = posting.tf as f32;
3684 let denominator = tf
3685 + crate::text_index::BM25_K1
3686 * (1.0 - crate::text_index::BM25_B
3687 + crate::text_index::BM25_B * (doc_len / avg_doc_len));
3688 let contribution = idf * ((tf * (crate::text_index::BM25_K1 + 1.0)) / denominator);
3689 terms.push(VantaBm25TermContribution {
3690 token,
3691 tf: posting.tf,
3692 df: term_stats.df,
3693 doc_len: doc_stats.doc_len,
3694 contribution,
3695 });
3696 }
3697
3698 Ok(terms)
3699 }
3700
3701 fn debug_matched_phrases_for_record(
3702 engine: &StorageEngine,
3703 record: &VantaMemoryRecord,
3704 text_query: &str,
3705 ) -> Result<Vec<String>> {
3706 let query_plan = crate::text_index::query_plan(text_query);
3707 if query_plan.phrases.is_empty() {
3708 return Ok(Vec::new());
3709 }
3710
3711 let mut term_positions = BTreeMap::new();
3712 for token in query_plan.terms {
3713 if let Some(value) = engine.get_from_partition(
3714 BackendPartition::TextIndex,
3715 &crate::text_index::posting_key(&record.namespace, &token, &record.key),
3716 )? {
3717 let posting = crate::text_index::decode_posting(&value)?;
3718 term_positions.insert(token, posting.positions);
3719 }
3720 }
3721
3722 Ok(query_plan
3723 .phrases
3724 .into_iter()
3725 .filter(|phrase| Self::text_positions_match_phrase(&term_positions, phrase))
3726 .map(|phrase| phrase.join(" "))
3727 .collect())
3728 }
3729
3730 fn debug_snippet(payload: &str, text_query: &str) -> Option<String> {
3731 Self::generate_snippet_with_highlighting(payload, text_query, false)
3732 }
3733
3734 fn generate_snippet_with_highlighting(
3744 payload: &str,
3745 text_query: &str,
3746 with_highlighting: bool,
3747 ) -> Option<String> {
3748 let query_plan = crate::text_index::query_plan(text_query);
3749 let first_token = query_plan.terms.iter().next()?;
3750
3751 if payload.len() <= 120 {
3752 if with_highlighting {
3753 return Some(Self::highlight_terms(payload, &query_plan.terms));
3754 }
3755 return Some(payload.to_string());
3756 }
3757
3758 let lower_payload = payload.to_ascii_lowercase();
3759 let match_at = lower_payload.find(first_token).unwrap_or(0);
3760 let mut start = match_at.saturating_sub(48);
3761 let mut end = match_at
3762 .saturating_add(first_token.len())
3763 .saturating_add(72)
3764 .min(payload.len());
3765 while start > 0 && !payload.is_char_boundary(start) {
3766 start -= 1;
3767 }
3768 while end < payload.len() && !payload.is_char_boundary(end) {
3769 end += 1;
3770 }
3771
3772 let snippet_text = payload[start..end].trim();
3773
3774 if with_highlighting {
3775 let highlighted = Self::highlight_terms(snippet_text, &query_plan.terms);
3776 let mut snippet = String::new();
3777 if start > 0 {
3778 snippet.push_str("...");
3779 }
3780 snippet.push_str(&highlighted);
3781 if end < payload.len() {
3782 snippet.push_str("...");
3783 }
3784 Some(snippet)
3785 } else {
3786 let mut snippet = String::new();
3787 if start > 0 {
3788 snippet.push_str("...");
3789 }
3790 snippet.push_str(snippet_text);
3791 if end < payload.len() {
3792 snippet.push_str("...");
3793 }
3794 Some(snippet)
3795 }
3796 }
3797
3798 fn highlight_terms(text: &str, terms: &std::collections::BTreeSet<String>) -> String {
3807 let mut result = String::new();
3808 let mut i = 0;
3809 let chars: Vec<char> = text.chars().collect();
3810
3811 while i < chars.len() {
3812 let mut matched = false;
3813
3814 for term in terms {
3815 let term_chars: Vec<char> = term.chars().collect();
3816 if i + term_chars.len() <= chars.len() {
3817 let slice: String = chars[i..i + term_chars.len()].iter().collect();
3818 if slice.eq_ignore_ascii_case(term) {
3819 result.push_str("<strong>");
3820 result.push_str(&slice);
3821 result.push_str("</strong>");
3822 i += term_chars.len();
3823 matched = true;
3824 break;
3825 }
3826 }
3827 }
3828
3829 if !matched {
3830 result.push(chars[i]);
3831 i += 1;
3832 }
3833 }
3834
3835 result
3836 }
3837
3838 pub fn graph_bfs(&self, roots: &[u64], max_depth: usize) -> Result<Vec<u64>> {
3839 let engine = self.engine_handle()?;
3840 let traverser = crate::graph::GraphTraverser::new(&engine);
3841 traverser.bfs_traverse(roots, max_depth)
3842 }
3843
3844 pub fn graph_dfs(&self, roots: &[u64], max_depth: usize) -> Result<Vec<u64>> {
3845 let engine = self.engine_handle()?;
3846 let traverser = crate::graph::GraphTraverser::new(&engine);
3847 traverser.dfs_traverse(roots, max_depth)
3848 }
3849
3850 pub fn graph_topological_sort(&self, roots: &[u64]) -> Result<Vec<u64>> {
3851 let engine = self.engine_handle()?;
3852 let traverser = crate::graph::GraphTraverser::new(&engine);
3853 traverser.topological_sort(roots)
3854 }
3855
3856 pub fn graph_is_dag(&self, roots: &[u64]) -> Result<bool> {
3857 let engine = self.engine_handle()?;
3858 let traverser = crate::graph::GraphTraverser::new(&engine);
3859 traverser.is_dag(roots)
3860 }
3861}
3862
3863impl From<IndexRebuildReport> for VantaIndexRebuildReport {
3864 fn from(report: IndexRebuildReport) -> Self {
3865 Self {
3866 scanned_nodes: report.scanned_nodes,
3867 indexed_vectors: report.indexed_vectors,
3868 skipped_tombstones: report.skipped_tombstones,
3869 duration_ms: report.duration_ms,
3870 derived_rebuild_ms: 0,
3871 index_path: report.index_path.to_string_lossy().into_owned(),
3872 success: report.success,
3873 }
3874 }
3875}
3876
3877impl From<crate::metrics::OperationalMetricsSnapshot> for VantaOperationalMetrics {
3878 fn from(metrics: crate::metrics::OperationalMetricsSnapshot) -> Self {
3879 Self {
3880 startup_ms: metrics.startup_ms,
3881 wal_replay_ms: metrics.wal_replay_ms,
3882 wal_records_replayed: metrics.wal_records_replayed,
3883 ann_rebuild_ms: metrics.ann_rebuild_ms,
3884 ann_rebuild_scanned_nodes: metrics.ann_rebuild_scanned_nodes,
3885 derived_rebuild_ms: metrics.derived_rebuild_ms,
3886 text_index_rebuild_ms: metrics.text_index_rebuild_ms,
3887 text_postings_written: metrics.text_postings_written,
3888 text_index_repairs: metrics.text_index_repairs,
3889 text_lexical_queries: metrics.text_lexical_queries,
3890 text_lexical_query_ms: metrics.text_lexical_query_ms,
3891 text_candidates_scored: metrics.text_candidates_scored,
3892 text_consistency_audits: metrics.text_consistency_audits,
3893 text_consistency_audit_failures: metrics.text_consistency_audit_failures,
3894 hybrid_query_ms: metrics.hybrid_query_ms,
3895 hybrid_candidates_fused: metrics.hybrid_candidates_fused,
3896 planner_hybrid_queries: metrics.planner_hybrid_queries,
3897 planner_text_only_queries: metrics.planner_text_only_queries,
3898 planner_vector_only_queries: metrics.planner_vector_only_queries,
3899 records_exported: metrics.records_exported,
3900 records_imported: metrics.records_imported,
3901 import_errors: metrics.import_errors,
3902 derived_prefix_scans: metrics.derived_prefix_scans,
3903 derived_full_scan_fallbacks: metrics.derived_full_scan_fallbacks,
3904 process_rss_bytes: metrics.memory.process_rss_bytes,
3905 process_virtual_bytes: metrics.memory.process_virtual_bytes,
3906 hnsw_nodes_count: metrics.memory.hnsw_nodes_count,
3907 hnsw_logical_bytes: metrics.memory.hnsw_logical_bytes,
3908 mmap_resident_bytes: metrics.memory.mmap_resident_bytes,
3909 volatile_cache_entries: metrics.memory.volatile_cache_entries,
3910 volatile_cache_cap_bytes: metrics.memory.volatile_cache_cap_bytes,
3911 }
3912 }
3913}
3914
3915impl From<VantaValue> for FieldValue {
3916 fn from(value: VantaValue) -> Self {
3917 match value {
3918 VantaValue::String(value) => FieldValue::String(value),
3919 VantaValue::Int(value) => FieldValue::Int(value),
3920 VantaValue::Float(value) => FieldValue::Float(value),
3921 VantaValue::Bool(value) => FieldValue::Bool(value),
3922 VantaValue::DateTime(value) => FieldValue::DateTime(value),
3923 VantaValue::ListString(value) => FieldValue::ListString(value),
3924 VantaValue::ListInt(value) => FieldValue::ListInt(value),
3925 VantaValue::ListFloat(value) => FieldValue::ListFloat(value),
3926 VantaValue::ListBool(value) => FieldValue::ListBool(value),
3927 VantaValue::ListDateTime(value) => FieldValue::ListDateTime(value),
3928 VantaValue::Null => FieldValue::Null,
3929 }
3930 }
3931}
3932
3933impl From<FieldValue> for VantaValue {
3934 fn from(value: FieldValue) -> Self {
3935 match value {
3936 FieldValue::String(value) => VantaValue::String(value),
3937 FieldValue::Int(value) => VantaValue::Int(value),
3938 FieldValue::Float(value) => VantaValue::Float(value),
3939 FieldValue::Bool(value) => VantaValue::Bool(value),
3940 FieldValue::DateTime(value) => VantaValue::DateTime(value),
3941 FieldValue::ListString(value) => VantaValue::ListString(value),
3942 FieldValue::ListInt(value) => VantaValue::ListInt(value),
3943 FieldValue::ListFloat(value) => VantaValue::ListFloat(value),
3944 FieldValue::ListBool(value) => VantaValue::ListBool(value),
3945 FieldValue::ListDateTime(value) => VantaValue::ListDateTime(value),
3946 FieldValue::Null => VantaValue::Null,
3947 }
3948 }
3949}
3950
3951impl From<ExecutionResult> for VantaQueryResult {
3952 fn from(result: ExecutionResult) -> Self {
3953 match result {
3954 ExecutionResult::Read(nodes) => {
3955 VantaQueryResult::Read(nodes.into_iter().map(Into::into).collect())
3956 }
3957 ExecutionResult::Write {
3958 affected_nodes,
3959 message,
3960 node_id,
3961 } => VantaQueryResult::Write {
3962 affected_nodes,
3963 message,
3964 node_id,
3965 },
3966 ExecutionResult::StaleContext(node_id) => VantaQueryResult::StaleContext { node_id },
3967 }
3968 }
3969}
3970
3971impl From<UnifiedNode> for VantaNodeRecord {
3972 fn from(node: UnifiedNode) -> Self {
3973 let is_alive = node.is_alive();
3974 let (vector, vector_dimensions) = match node.vector {
3975 VectorRepresentations::Full(vector) => {
3976 let dims = vector.len();
3977 (Some(vector), dims)
3978 }
3979 VectorRepresentations::None => (None, 0),
3980 other => (None, other.dimensions()),
3981 };
3982
3983 let tier = match node.tier {
3984 crate::node::NodeTier::Hot => VantaStorageTier::Hot,
3985 crate::node::NodeTier::Cold => VantaStorageTier::Cold,
3986 };
3987
3988 let fields = node
3989 .relational
3990 .into_iter()
3991 .map(|(key, value)| (key, value.into()))
3992 .collect();
3993
3994 let edges = node
3995 .edges
3996 .into_iter()
3997 .map(|edge| VantaEdgeRecord {
3998 target: edge.target,
3999 label: edge.label,
4000 weight: edge.weight,
4001 })
4002 .collect();
4003
4004 Self {
4005 id: node.id,
4006 fields,
4007 vector,
4008 vector_dimensions,
4009 edges,
4010 confidence_score: node.confidence_score,
4011 importance: node.importance,
4012 hits: node.hits,
4013 last_accessed: node.last_accessed,
4014 epoch: node.epoch,
4015 tier,
4016 is_alive,
4017 }
4018 }
4019}