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

zeph_memory/
db_vector_store.rs
