zeph_memory/
db_vector_store.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! `SQLite` BLOB vector store — offline fallback implementation.
5//!
6//! Stores dense vectors as raw `f32` BLOBs in a `SQLite` table and performs cosine
7//! similarity in memory.  Suitable for offline use and CI environments without a
8//! running Qdrant instance.  Not optimised for large collections.
9
10use std::collections::HashMap;
11use std::sync::atomic::AtomicBool;
12#[allow(unused_imports)]
13use zeph_db::sql;
14
15use zeph_db::{ActiveDialect, DbPool};
16
17use crate::vector_store::{
18    BoxFuture, FieldValue, ScoredVectorPoint, ScrollResult, ScrollWithIdsResult, VectorFilter,
19    VectorPoint, VectorStore, VectorStoreError,
20};
21
22/// Database-backed in-process vector store.
23///
24/// Stores vectors as BLOBs in `SQLite` and performs cosine similarity in memory.
25/// For production-scale workloads, prefer the Qdrant-backed store.
26pub struct DbVectorStore {
27    pool: DbPool,
28}
29
30/// Backward-compatible alias.
31pub type SqliteVectorStore = DbVectorStore;
32
33impl DbVectorStore {
34    /// Create a new `DbVectorStore` from an existing connection pool.
35    ///
36    /// The pool must come from a database that has run the `zeph-db` migrations
37    /// (which create the `vector_store` table).
38    #[must_use]
39    pub fn new(pool: DbPool) -> Self {
40        Self { pool }
41    }
42}
43
44use zeph_common::math::cosine_similarity;
45
46fn matches_filter(payload: &HashMap<String, serde_json::Value>, filter: &VectorFilter) -> bool {
47    for cond in &filter.must {
48        let Some(val) = payload.get(&cond.field) else {
49            return false;
50        };
51        let matches = match &cond.value {
52            FieldValue::Integer(i) => val.as_i64().is_some_and(|v| v == *i),
53            FieldValue::Text(t) => val.as_str().is_some_and(|v| v == t.as_str()),
54        };
55        if !matches {
56            return false;
57        }
58    }
59    for cond in &filter.must_not {
60        let Some(val) = payload.get(&cond.field) else {
61            continue;
62        };
63        let matches = match &cond.value {
64            FieldValue::Integer(i) => val.as_i64().is_some_and(|v| v == *i),
65            FieldValue::Text(t) => val.as_str().is_some_and(|v| v == t.as_str()),
66        };
67        if matches {
68            return false;
69        }
70    }
71    true
72}
73
74impl VectorStore for DbVectorStore {
75    fn ensure_collection(
76        &self,
77        collection: &str,
78        _vector_size: u64,
79    ) -> BoxFuture<'_, Result<(), VectorStoreError>> {
80        let collection = collection.to_owned();
81        Box::pin(async move {
82            let sql = zeph_db::rewrite_placeholders(&format!(
83                "{} INTO vector_collections (name) VALUES (?){}",
84                <ActiveDialect as zeph_db::dialect::Dialect>::INSERT_IGNORE,
85                <ActiveDialect as zeph_db::dialect::Dialect>::CONFLICT_NOTHING,
86            ));
87            zeph_db::query(sqlx::AssertSqlSafe(sql))
88                .bind(&collection)
89                .execute(&self.pool)
90                .await
91                .map_err(|e| VectorStoreError::Collection(e.to_string()))?;
92            Ok(())
93        })
94    }
95
96    fn collection_exists(&self, collection: &str) -> BoxFuture<'_, Result<bool, VectorStoreError>> {
97        let collection = collection.to_owned();
98        Box::pin(async move {
99            let row: (i64,) = zeph_db::query_as(sql!(
100                "SELECT COUNT(*) FROM vector_collections WHERE name = ?"
101            ))
102            .bind(&collection)
103            .fetch_one(&self.pool)
104            .await
105            .map_err(|e| VectorStoreError::Connection(e.to_string()))?;
106            Ok(row.0 > 0)
107        })
108    }
109
110    fn delete_collection(&self, collection: &str) -> BoxFuture<'_, Result<(), VectorStoreError>> {
111        let collection = collection.to_owned();
112        Box::pin(async move {
113            zeph_db::query(sql!("DELETE FROM vector_points WHERE collection = ?"))
114                .bind(&collection)
115                .execute(&self.pool)
116                .await
117                .map_err(|e| VectorStoreError::Delete(e.to_string()))?;
118            zeph_db::query(sql!("DELETE FROM vector_collections WHERE name = ?"))
119                .bind(&collection)
120                .execute(&self.pool)
121                .await
122                .map_err(|e| VectorStoreError::Delete(e.to_string()))?;
123            Ok(())
124        })
125    }
126
127    fn upsert(
128        &self,
129        collection: &str,
130        points: Vec<VectorPoint>,
131    ) -> BoxFuture<'_, Result<(), VectorStoreError>> {
132        let collection = collection.to_owned();
133        #[cfg(feature = "profiling")]
134        let span = tracing::info_span!(
135            "memory.vector_store",
136            operation = "upsert",
137            collection = %collection
138        );
139        let fut = Box::pin(async move {
140            for point in points {
141                let vector_bytes: Vec<u8> =
142                    point.vector.iter().flat_map(|f| f.to_le_bytes()).collect();
143                let payload_json = serde_json::to_string(&point.payload)
144                    .map_err(|e| VectorStoreError::Serialization(e.to_string()))?;
145                zeph_db::query(
146                    sql!("INSERT INTO vector_points (id, collection, vector, payload) VALUES (?, ?, ?, ?) \
147                     ON CONFLICT(collection, id) DO UPDATE SET vector = excluded.vector, payload = excluded.payload"),
148                )
149                .bind(&point.id)
150                .bind(&collection)
151                .bind(&vector_bytes)
152                .bind(&payload_json)
153                .execute(&self.pool)
154                .await
155                .map_err(|e| VectorStoreError::Upsert(e.to_string()))?;
156            }
157            Ok(())
158        });
159        #[cfg(feature = "profiling")]
160        return Box::pin(tracing::Instrument::instrument(fut, span));
161        #[cfg(not(feature = "profiling"))]
162        fut
163    }
164
165    fn search_clamp_diagnostics(&self) -> (&'static str, &'static AtomicBool) {
166        static CLAMP_WARNED: AtomicBool = AtomicBool::new(false);
167        ("DbVectorStore::search", &CLAMP_WARNED)
168    }
169
170    fn search_clamped(
171        &self,
172        collection: &str,
173        vector: Vec<f32>,
174        limit: u64,
175        filter: Option<VectorFilter>,
176    ) -> BoxFuture<'_, Result<Vec<ScoredVectorPoint>, VectorStoreError>> {
177        let collection = collection.to_owned();
178        #[cfg(feature = "profiling")]
179        let span = tracing::info_span!(
180            "memory.vector_store",
181            operation = "search",
182            collection = %collection
183        );
184        let fut = Box::pin(async move {
185            let rows: Vec<(String, Vec<u8>, String)> = zeph_db::query_as(sql!(
186                "SELECT id, vector, payload FROM vector_points WHERE collection = ?"
187            ))
188            .bind(&collection)
189            .fetch_all(&self.pool)
190            .await
191            .map_err(|e| VectorStoreError::Search(e.to_string()))?;
192
193            let limit_usize = usize::try_from(limit).unwrap_or(usize::MAX);
194            let mut scored: Vec<ScoredVectorPoint> = rows
195                .into_iter()
196                .filter_map(|(id, blob, payload_str)| {
197                    if blob.len() % 4 != 0 {
198                        return None;
199                    }
200                    let stored: Vec<f32> = blob
201                        .chunks_exact(4)
202                        .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
203                        .collect();
204                    let payload: HashMap<String, serde_json::Value> =
205                        serde_json::from_str(&payload_str).unwrap_or_default();
206
207                    if filter
208                        .as_ref()
209                        .is_some_and(|f| !matches_filter(&payload, f))
210                    {
211                        return None;
212                    }
213
214                    let score = cosine_similarity(&vector, &stored);
215                    Some(ScoredVectorPoint { id, score, payload })
216                })
217                .collect();
218
219            scored.sort_by(|a, b| {
220                b.score
221                    .partial_cmp(&a.score)
222                    .unwrap_or(std::cmp::Ordering::Equal)
223            });
224            scored.truncate(limit_usize);
225            Ok(scored)
226        });
227        #[cfg(feature = "profiling")]
228        return Box::pin(tracing::Instrument::instrument(fut, span));
229        #[cfg(not(feature = "profiling"))]
230        fut
231    }
232
233    fn delete_by_ids(
234        &self,
235        collection: &str,
236        ids: Vec<String>,
237    ) -> BoxFuture<'_, Result<(), VectorStoreError>> {
238        let collection = collection.to_owned();
239        #[cfg(feature = "profiling")]
240        let span = tracing::info_span!(
241            "memory.vector_store",
242            operation = "delete",
243            collection = %collection
244        );
245        let fut = Box::pin(async move {
246            for id in ids {
247                zeph_db::query(sql!(
248                    "DELETE FROM vector_points WHERE collection = ? AND id = ?"
249                ))
250                .bind(&collection)
251                .bind(&id)
252                .execute(&self.pool)
253                .await
254                .map_err(|e| VectorStoreError::Delete(e.to_string()))?;
255            }
256            Ok(())
257        });
258        #[cfg(feature = "profiling")]
259        return Box::pin(tracing::Instrument::instrument(fut, span));
260        #[cfg(not(feature = "profiling"))]
261        fut
262    }
263
264    fn scroll_all(
265        &self,
266        collection: &str,
267        key_field: &str,
268    ) -> BoxFuture<'_, Result<ScrollResult, VectorStoreError>> {
269        let collection = collection.to_owned();
270        let key_field = key_field.to_owned();
271        Box::pin(async move {
272            let rows: Vec<(String, String)> = zeph_db::query_as(sql!(
273                "SELECT id, payload FROM vector_points WHERE collection = ?"
274            ))
275            .bind(&collection)
276            .fetch_all(&self.pool)
277            .await
278            .map_err(|e| VectorStoreError::Scroll(e.to_string()))?;
279
280            let mut result = ScrollResult::new();
281            for (id, payload_str) in rows {
282                let payload: HashMap<String, serde_json::Value> =
283                    serde_json::from_str(&payload_str).unwrap_or_default();
284                if let Some(val) = payload.get(&key_field) {
285                    let mut map = HashMap::new();
286                    map.insert(
287                        key_field.clone(),
288                        val.as_str().unwrap_or_default().to_owned(),
289                    );
290                    result.insert(id, map);
291                }
292            }
293            Ok(result)
294        })
295    }
296
297    fn scroll_all_with_point_ids(
298        &self,
299        collection: &str,
300        key_field: &str,
301    ) -> BoxFuture<'_, Result<ScrollWithIdsResult, VectorStoreError>> {
302        let collection = collection.to_owned();
303        let key_field = key_field.to_owned();
304        Box::pin(async move {
305            let rows: Vec<(String, String)> = zeph_db::query_as(sql!(
306                "SELECT id, payload FROM vector_points WHERE collection = ?"
307            ))
308            .bind(&collection)
309            .fetch_all(&self.pool)
310            .await
311            .map_err(|e| VectorStoreError::Scroll(e.to_string()))?;
312
313            let mut result = Vec::new();
314            for (point_id, payload_str) in rows {
315                let payload: HashMap<String, serde_json::Value> =
316                    serde_json::from_str(&payload_str).unwrap_or_default();
317                let Some(key_val) = payload.get(&key_field).and_then(|v| v.as_str()) else {
318                    continue;
319                };
320                let mut fields = HashMap::new();
321                for (k, v) in &payload {
322                    if let Some(s) = v.as_str() {
323                        fields.insert(k.clone(), s.to_owned());
324                    }
325                }
326                // Ensure the key_field value is always present in the fields map.
327                fields.insert(key_field.clone(), key_val.to_owned());
328                result.push((point_id, fields));
329            }
330            Ok(result)
331        })
332    }
333
334    fn health_check(&self) -> BoxFuture<'_, Result<bool, VectorStoreError>> {
335        Box::pin(async move {
336            zeph_db::query_scalar::<_, i32>(sql!("SELECT 1"))
337                .fetch_one(&self.pool)
338                .await
339                .map(|_| true)
340                .map_err(|e| VectorStoreError::Collection(e.to_string()))
341        })
342    }
343}
344
345#[cfg(test)]
346mod tests {
347    use super::*;
348    use crate::store::SqliteStore;
349    use crate::vector_store::FieldCondition;
350
351    async fn setup() -> (DbVectorStore, SqliteStore) {
352        let store = SqliteStore::new(":memory:").await.unwrap();
353        let pool = store.pool().clone();
354        let vs = DbVectorStore::new(pool);
355        (vs, store)
356    }
357
358    #[tokio::test]
359    async fn ensure_and_exists() {
360        let (vs, _) = setup().await;
361        assert!(!vs.collection_exists("col1").await.unwrap());
362        vs.ensure_collection("col1", 4).await.unwrap();
363        assert!(vs.collection_exists("col1").await.unwrap());
364        // idempotent
365        vs.ensure_collection("col1", 4).await.unwrap();
366        assert!(vs.collection_exists("col1").await.unwrap());
367    }
368
369    #[tokio::test]
370    async fn delete_collection() {
371        let (vs, _) = setup().await;
372        vs.ensure_collection("col1", 4).await.unwrap();
373        vs.upsert(
374            "col1",
375            vec![VectorPoint {
376                id: "p1".into(),
377                vector: vec![1.0, 0.0, 0.0, 0.0],
378                payload: HashMap::new(),
379            }],
380        )
381        .await
382        .unwrap();
383        vs.delete_collection("col1").await.unwrap();
384        assert!(!vs.collection_exists("col1").await.unwrap());
385    }
386
387    #[tokio::test]
388    async fn upsert_and_search() {
389        let (vs, _) = setup().await;
390        vs.ensure_collection("c", 4).await.unwrap();
391        vs.upsert(
392            "c",
393            vec![
394                VectorPoint {
395                    id: "a".into(),
396                    vector: vec![1.0, 0.0, 0.0, 0.0],
397                    payload: HashMap::from([("role".into(), serde_json::json!("user"))]),
398                },
399                VectorPoint {
400                    id: "b".into(),
401                    vector: vec![0.0, 1.0, 0.0, 0.0],
402                    payload: HashMap::from([("role".into(), serde_json::json!("assistant"))]),
403                },
404            ],
405        )
406        .await
407        .unwrap();
408
409        let results = vs
410            .search("c", vec![1.0, 0.0, 0.0, 0.0], 2, None)
411            .await
412            .unwrap();
413        assert_eq!(results.len(), 2);
414        assert_eq!(results[0].id, "a");
415        assert!((results[0].score - 1.0).abs() < 1e-5);
416    }
417
418    /// Issue #6616: `VectorStore::search` must clamp an oversized `limit` at the trait-impl
419    /// choke point itself, not only in the wrapper methods added for issue #6553 — a caller
420    /// reaching `DbVectorStore` directly must still get the same bound.
421    #[tokio::test]
422    #[tracing_test::traced_test]
423    async fn search_clamps_oversized_limit() {
424        let (vs, _) = setup().await;
425        vs.ensure_collection("c", 4).await.unwrap();
426        let points: Vec<VectorPoint> = (0..(crate::MAX_SEARCH_LIMIT + 10))
427            .map(|i| VectorPoint {
428                id: format!("p{i}"),
429                vector: vec![1.0, 0.0, 0.0, 0.0],
430                payload: HashMap::new(),
431            })
432            .collect();
433        vs.upsert("c", points).await.unwrap();
434
435        let results = vs
436            .search("c", vec![1.0, 0.0, 0.0, 0.0], u64::MAX, None)
437            .await
438            .unwrap();
439        assert_eq!(results.len(), crate::MAX_SEARCH_LIMIT);
440        assert!(
441            logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"),
442            "expected a one-shot warn when the trait-impl clamp actually reduces the requested limit"
443        );
444    }
445
446    #[tokio::test]
447    async fn search_with_filter() {
448        let (vs, _) = setup().await;
449        vs.ensure_collection("c", 4).await.unwrap();
450        vs.upsert(
451            "c",
452            vec![
453                VectorPoint {
454                    id: "a".into(),
455                    vector: vec![1.0, 0.0, 0.0, 0.0],
456                    payload: HashMap::from([("role".into(), serde_json::json!("user"))]),
457                },
458                VectorPoint {
459                    id: "b".into(),
460                    vector: vec![1.0, 0.0, 0.0, 0.0],
461                    payload: HashMap::from([("role".into(), serde_json::json!("assistant"))]),
462                },
463            ],
464        )
465        .await
466        .unwrap();
467
468        let filter = VectorFilter {
469            must: vec![FieldCondition {
470                field: "role".into(),
471                value: FieldValue::Text("user".into()),
472            }],
473            must_not: vec![],
474        };
475        let results = vs
476            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, Some(filter))
477            .await
478            .unwrap();
479        assert_eq!(results.len(), 1);
480        assert_eq!(results[0].id, "a");
481    }
482
483    #[tokio::test]
484    async fn delete_by_ids() {
485        let (vs, _) = setup().await;
486        vs.ensure_collection("c", 4).await.unwrap();
487        vs.upsert(
488            "c",
489            vec![
490                VectorPoint {
491                    id: "a".into(),
492                    vector: vec![1.0, 0.0, 0.0, 0.0],
493                    payload: HashMap::new(),
494                },
495                VectorPoint {
496                    id: "b".into(),
497                    vector: vec![0.0, 1.0, 0.0, 0.0],
498                    payload: HashMap::new(),
499                },
500            ],
501        )
502        .await
503        .unwrap();
504        vs.delete_by_ids("c", vec!["a".into()]).await.unwrap();
505        let results = vs
506            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, None)
507            .await
508            .unwrap();
509        assert_eq!(results.len(), 1);
510        assert_eq!(results[0].id, "b");
511    }
512
513    #[tokio::test]
514    async fn scroll_all() {
515        let (vs, _) = setup().await;
516        vs.ensure_collection("c", 4).await.unwrap();
517        vs.upsert(
518            "c",
519            vec![VectorPoint {
520                id: "p1".into(),
521                vector: vec![1.0, 0.0, 0.0, 0.0],
522                payload: HashMap::from([("text".into(), serde_json::json!("hello"))]),
523            }],
524        )
525        .await
526        .unwrap();
527        let result = vs.scroll_all("c", "text").await.unwrap();
528        assert_eq!(result.len(), 1);
529        assert_eq!(result["p1"]["text"], "hello");
530    }
531
532    #[tokio::test]
533    async fn upsert_updates_existing() {
534        let (vs, _) = setup().await;
535        vs.ensure_collection("c", 4).await.unwrap();
536        vs.upsert(
537            "c",
538            vec![VectorPoint {
539                id: "p1".into(),
540                vector: vec![1.0, 0.0, 0.0, 0.0],
541                payload: HashMap::from([("v".into(), serde_json::json!(1))]),
542            }],
543        )
544        .await
545        .unwrap();
546        vs.upsert(
547            "c",
548            vec![VectorPoint {
549                id: "p1".into(),
550                vector: vec![0.0, 1.0, 0.0, 0.0],
551                payload: HashMap::from([("v".into(), serde_json::json!(2))]),
552            }],
553        )
554        .await
555        .unwrap();
556        let results = vs
557            .search("c", vec![0.0, 1.0, 0.0, 0.0], 1, None)
558            .await
559            .unwrap();
560        assert_eq!(results.len(), 1);
561        assert!((results[0].score - 1.0).abs() < 1e-5);
562    }
563
564    #[test]
565    fn cosine_similarity_import_wired() {
566        // Smoke test: verifies the re-export binding is intact. Edge-case coverage is in math.rs.
567        assert!(!cosine_similarity(&[1.0, 0.0], &[0.0, 1.0]).is_nan());
568    }
569
570    #[tokio::test]
571    async fn search_with_must_not_filter() {
572        let (vs, _) = setup().await;
573        vs.ensure_collection("c", 4).await.unwrap();
574        vs.upsert(
575            "c",
576            vec![
577                VectorPoint {
578                    id: "a".into(),
579                    vector: vec![1.0, 0.0, 0.0, 0.0],
580                    payload: HashMap::from([("role".into(), serde_json::json!("user"))]),
581                },
582                VectorPoint {
583                    id: "b".into(),
584                    vector: vec![1.0, 0.0, 0.0, 0.0],
585                    payload: HashMap::from([("role".into(), serde_json::json!("system"))]),
586                },
587            ],
588        )
589        .await
590        .unwrap();
591
592        let filter = VectorFilter {
593            must: vec![],
594            must_not: vec![FieldCondition {
595                field: "role".into(),
596                value: FieldValue::Text("system".into()),
597            }],
598        };
599        let results = vs
600            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, Some(filter))
601            .await
602            .unwrap();
603        assert_eq!(results.len(), 1);
604        assert_eq!(results[0].id, "a");
605    }
606
607    #[tokio::test]
608    async fn search_with_integer_filter() {
609        let (vs, _) = setup().await;
610        vs.ensure_collection("c", 4).await.unwrap();
611        vs.upsert(
612            "c",
613            vec![
614                VectorPoint {
615                    id: "a".into(),
616                    vector: vec![1.0, 0.0, 0.0, 0.0],
617                    payload: HashMap::from([("conv_id".into(), serde_json::json!(1))]),
618                },
619                VectorPoint {
620                    id: "b".into(),
621                    vector: vec![1.0, 0.0, 0.0, 0.0],
622                    payload: HashMap::from([("conv_id".into(), serde_json::json!(2))]),
623                },
624            ],
625        )
626        .await
627        .unwrap();
628
629        let filter = VectorFilter {
630            must: vec![FieldCondition {
631                field: "conv_id".into(),
632                value: FieldValue::Integer(1),
633            }],
634            must_not: vec![],
635        };
636        let results = vs
637            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, Some(filter))
638            .await
639            .unwrap();
640        assert_eq!(results.len(), 1);
641        assert_eq!(results[0].id, "a");
642    }
643
644    #[tokio::test]
645    async fn search_empty_collection() {
646        let (vs, _) = setup().await;
647        vs.ensure_collection("c", 4).await.unwrap();
648        let results = vs
649            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, None)
650            .await
651            .unwrap();
652        assert!(results.is_empty());
653    }
654
655    #[tokio::test]
656    async fn search_with_must_not_integer_filter() {
657        let (vs, _) = setup().await;
658        vs.ensure_collection("c", 4).await.unwrap();
659        vs.upsert(
660            "c",
661            vec![
662                VectorPoint {
663                    id: "a".into(),
664                    vector: vec![1.0, 0.0, 0.0, 0.0],
665                    payload: HashMap::from([("conv_id".into(), serde_json::json!(1))]),
666                },
667                VectorPoint {
668                    id: "b".into(),
669                    vector: vec![1.0, 0.0, 0.0, 0.0],
670                    payload: HashMap::from([("conv_id".into(), serde_json::json!(2))]),
671                },
672            ],
673        )
674        .await
675        .unwrap();
676
677        let filter = VectorFilter {
678            must: vec![],
679            must_not: vec![FieldCondition {
680                field: "conv_id".into(),
681                value: FieldValue::Integer(1),
682            }],
683        };
684        let results = vs
685            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, Some(filter))
686            .await
687            .unwrap();
688        assert_eq!(results.len(), 1);
689        assert_eq!(results[0].id, "b");
690    }
691
692    #[tokio::test]
693    async fn search_with_combined_must_and_must_not() {
694        let (vs, _) = setup().await;
695        vs.ensure_collection("c", 4).await.unwrap();
696        vs.upsert(
697            "c",
698            vec![
699                VectorPoint {
700                    id: "a".into(),
701                    vector: vec![1.0, 0.0, 0.0, 0.0],
702                    payload: HashMap::from([
703                        ("role".into(), serde_json::json!("user")),
704                        ("conv_id".into(), serde_json::json!(1)),
705                    ]),
706                },
707                VectorPoint {
708                    id: "b".into(),
709                    vector: vec![1.0, 0.0, 0.0, 0.0],
710                    payload: HashMap::from([
711                        ("role".into(), serde_json::json!("user")),
712                        ("conv_id".into(), serde_json::json!(2)),
713                    ]),
714                },
715                VectorPoint {
716                    id: "c".into(),
717                    vector: vec![1.0, 0.0, 0.0, 0.0],
718                    payload: HashMap::from([
719                        ("role".into(), serde_json::json!("assistant")),
720                        ("conv_id".into(), serde_json::json!(1)),
721                    ]),
722                },
723            ],
724        )
725        .await
726        .unwrap();
727
728        let filter = VectorFilter {
729            must: vec![FieldCondition {
730                field: "role".into(),
731                value: FieldValue::Text("user".into()),
732            }],
733            must_not: vec![FieldCondition {
734                field: "conv_id".into(),
735                value: FieldValue::Integer(2),
736            }],
737        };
738        let results = vs
739            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, Some(filter))
740            .await
741            .unwrap();
742        // Only "a": role=user AND conv_id != 2
743        assert_eq!(results.len(), 1);
744        assert_eq!(results[0].id, "a");
745    }
746
747    #[tokio::test]
748    async fn scroll_all_missing_key_field() {
749        let (vs, _) = setup().await;
750        vs.ensure_collection("c", 4).await.unwrap();
751        vs.upsert(
752            "c",
753            vec![VectorPoint {
754                id: "p1".into(),
755                vector: vec![1.0, 0.0, 0.0, 0.0],
756                payload: HashMap::from([("other".into(), serde_json::json!("value"))]),
757            }],
758        )
759        .await
760        .unwrap();
761        // key_field "text" doesn't exist in payload → point excluded from result
762        let result = vs.scroll_all("c", "text").await.unwrap();
763        assert!(
764            result.is_empty(),
765            "points without the key field must not appear in scroll result"
766        );
767    }
768
769    #[tokio::test]
770    async fn delete_by_ids_empty_and_nonexistent() {
771        let (vs, _) = setup().await;
772        vs.ensure_collection("c", 4).await.unwrap();
773        vs.upsert(
774            "c",
775            vec![VectorPoint {
776                id: "a".into(),
777                vector: vec![1.0, 0.0, 0.0, 0.0],
778                payload: HashMap::new(),
779            }],
780        )
781        .await
782        .unwrap();
783
784        // Empty list: no-op, must succeed
785        vs.delete_by_ids("c", vec![]).await.unwrap();
786
787        // Non-existent id: must succeed (idempotent)
788        vs.delete_by_ids("c", vec!["nonexistent".into()])
789            .await
790            .unwrap();
791
792        // Original point still present
793        let results = vs
794            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, None)
795            .await
796            .unwrap();
797        assert_eq!(results.len(), 1);
798        assert_eq!(results[0].id, "a");
799    }
800
801    #[tokio::test]
802    async fn search_corrupt_blob_skipped() {
803        let (vs, store) = setup().await;
804        vs.ensure_collection("c", 4).await.unwrap();
805
806        // Insert a valid point first
807        vs.upsert(
808            "c",
809            vec![VectorPoint {
810                id: "valid".into(),
811                vector: vec![1.0, 0.0, 0.0, 0.0],
812                payload: HashMap::new(),
813            }],
814        )
815        .await
816        .unwrap();
817
818        // Insert raw invalid bytes directly into vector_points table
819        // 3 bytes cannot be cast to f32 (needs multiples of 4)
820        let corrupt_blob: Vec<u8> = vec![0xFF, 0xFE, 0xFD];
821        let payload_json = r"{}";
822        zeph_db::query(sql!(
823            "INSERT INTO vector_points (id, collection, vector, payload) VALUES (?, ?, ?, ?)"
824        ))
825        .bind("corrupt")
826        .bind("c")
827        .bind(&corrupt_blob)
828        .bind(payload_json)
829        .execute(store.pool())
830        .await
831        .unwrap();
832
833        // Search must not panic and must skip the corrupt point
834        let results = vs
835            .search("c", vec![1.0, 0.0, 0.0, 0.0], 10, None)
836            .await
837            .unwrap();
838        assert_eq!(results.len(), 1);
839        assert_eq!(results[0].id, "valid");
840    }
841
842    #[tokio::test]
843    async fn scroll_all_with_point_ids_basic() {
844        let (vs, _) = setup().await;
845        vs.ensure_collection("c", 4).await.unwrap();
846        vs.upsert(
847            "c",
848            vec![
849                VectorPoint {
850                    id: "p1".into(),
851                    vector: vec![1.0, 0.0, 0.0, 0.0],
852                    payload: HashMap::from([
853                        ("entity_id_str".into(), serde_json::json!("42")),
854                        ("name".into(), serde_json::json!("alice")),
855                    ]),
856                },
857                VectorPoint {
858                    id: "p2".into(),
859                    vector: vec![0.0, 1.0, 0.0, 0.0],
860                    payload: HashMap::from([
861                        ("entity_id_str".into(), serde_json::json!("99")),
862                        ("name".into(), serde_json::json!("bob")),
863                    ]),
864                },
865            ],
866        )
867        .await
868        .unwrap();
869
870        let result = vs
871            .scroll_all_with_point_ids("c", "entity_id_str")
872            .await
873            .unwrap();
874        assert_eq!(result.len(), 2);
875
876        // Collect into a sorted map for deterministic assertion
877        let mut by_id: std::collections::BTreeMap<
878            String,
879            std::collections::HashMap<String, String>,
880        > = result.into_iter().collect();
881        let p1 = by_id.remove("p1").expect("p1 missing");
882        let p2 = by_id.remove("p2").expect("p2 missing");
883        assert_eq!(p1.get("entity_id_str").map(String::as_str), Some("42"));
884        assert_eq!(p1.get("name").map(String::as_str), Some("alice"));
885        assert_eq!(p2.get("entity_id_str").map(String::as_str), Some("99"));
886        assert_eq!(p2.get("name").map(String::as_str), Some("bob"));
887    }
888
889    #[tokio::test]
890    async fn scroll_all_with_point_ids_skips_missing_key_field() {
891        let (vs, _) = setup().await;
892        vs.ensure_collection("c", 4).await.unwrap();
893        vs.upsert(
894            "c",
895            vec![
896                VectorPoint {
897                    id: "has-key".into(),
898                    vector: vec![1.0, 0.0, 0.0, 0.0],
899                    payload: HashMap::from([("entity_id_str".into(), serde_json::json!("7"))]),
900                },
901                VectorPoint {
902                    id: "no-key".into(),
903                    vector: vec![0.0, 1.0, 0.0, 0.0],
904                    payload: HashMap::from([("other".into(), serde_json::json!("value"))]),
905                },
906            ],
907        )
908        .await
909        .unwrap();
910
911        let result = vs
912            .scroll_all_with_point_ids("c", "entity_id_str")
913            .await
914            .unwrap();
915        // Only the point that has the key field must be returned
916        assert_eq!(result.len(), 1);
917        assert_eq!(result[0].0, "has-key");
918        assert_eq!(
919            result[0].1.get("entity_id_str").map(String::as_str),
920            Some("7")
921        );
922    }
923}
zeph_memory/db_vector_store.rs

zeph_memory/
db_vector_store.rs
