1use std::collections::HashMap;
4
5use async_trait::async_trait;
6use lc_core::http::{HttpClient, RequestOptions};
7use serde::Deserialize;
8use serde_json::Value;
9
10use crate::{
11 Document, Embeddings, FilterOp, MetadataFilter, SearchResult, VectorStore, VectorStoreError,
12};
13
14const PINECONE_METADATA_MAX_BYTES: usize = 40 * 1024;
17
18pub struct PineconeStore {
20 api_key: String,
21 host: String,
22 http: HttpClient,
23}
24
25impl PineconeStore {
26 pub fn new(api_key: impl Into<String>, host: impl Into<String>) -> Self {
30 let http = HttpClient::api()
31 .build()
32 .expect("build pinecone http client");
33 Self {
34 api_key: api_key.into(),
35 host: host.into(),
36 http,
37 }
38 }
39
40 fn request_options(&self) -> RequestOptions {
42 let mut opts = RequestOptions::new();
43 if let Ok(v) = reqwest::header::HeaderValue::from_str(&self.api_key) {
44 opts = opts.header(reqwest::header::HeaderName::from_static("api-key"), v);
45 }
46 opts
47 }
48
49 fn map_http_error(prefix: &str, err: lc_core::http::HttpError) -> VectorStoreError {
52 let msg = match err {
53 lc_core::http::HttpError::Status { status, body } => {
54 format!("HTTP {status}: {body}")
55 }
56 other => other.to_string(),
57 };
58 VectorStoreError::ConnectionError(format!("{prefix}: {msg}"))
59 }
60
61 pub fn build_upsert_body(
70 docs: &[Document],
71 vectors: &[Vec<f32>],
72 ids: &[String],
73 ) -> Result<serde_json::Value, VectorStoreError> {
74 if docs.len() != vectors.len() || docs.len() != ids.len() {
75 return Err(VectorStoreError::StorageError(format!(
76 "Pinecone upsert shape mismatch: {} docs, {} vectors, {} ids",
77 docs.len(),
78 vectors.len(),
79 ids.len()
80 )));
81 }
82 let mut vectors_json: Vec<serde_json::Value> = Vec::with_capacity(docs.len());
83 for (i, (doc, id)) in docs.iter().zip(ids.iter()).enumerate() {
84 let mut metadata = doc.metadata.clone();
87 if !metadata.contains_key("content") {
88 metadata.insert("content".to_string(), Value::from(doc.content.clone()));
89 }
90 let metadata_bytes = serde_json::to_vec(&metadata)
91 .map_err(|e| {
92 VectorStoreError::StorageError(format!(
93 "failed to serialize Pinecone metadata: {e}"
94 ))
95 })?
96 .len();
97 if metadata_bytes > PINECONE_METADATA_MAX_BYTES {
98 return Err(VectorStoreError::StorageError(format!(
99 "vector metadata exceeds the 40 KiB limit ({metadata_bytes} bytes); \
100 refusing to upsert id {id}"
101 )));
102 }
103 vectors_json.push(serde_json::json!({
104 "id": id,
105 "values": &vectors[i],
106 "metadata": metadata,
107 }));
108 }
109 Ok(serde_json::json!({ "vectors": vectors_json }))
110 }
111
112 pub fn build_query_body(query_vec: &[f32], top_k: usize) -> serde_json::Value {
114 serde_json::json!({
115 "vector": query_vec,
116 "topK": top_k,
117 "includeMetadata": true,
118 })
119 }
120
121 pub fn build_query_body_filtered(
127 query_vec: &[f32],
128 top_k: usize,
129 filter: &MetadataFilter,
130 ) -> serde_json::Value {
131 let mut body = Self::build_query_body(query_vec, top_k);
132 body["filter"] = filter_to_pinecone(filter);
133 body
134 }
135
136 pub async fn upsert(
138 &self,
139 docs: &[Document],
140 embeddings: &dyn Embeddings,
141 ) -> Result<(), VectorStoreError> {
142 let texts: Vec<&str> = docs.iter().map(|d| d.content.as_str()).collect();
143 let vectors = embeddings
144 .embed_documents(&texts)
145 .await
146 .map_err(|e| VectorStoreError::EmbeddingError(e.to_string()))?;
147 let ids: Vec<String> = docs
148 .iter()
149 .map(|d| d.id.clone().unwrap_or_else(|| uuid::Uuid::new_v4().to_string()))
150 .collect();
151 let body = Self::build_upsert_body(docs, &vectors, &ids)?;
152 let url = format!("{}/vectors/upsert", self.host);
153 self.http
154 .post_json_with(&url, &body, self.request_options())
155 .await
156 .map_err(|e| Self::map_http_error("Pinecone upsert error", e))?;
157 Ok(())
158 }
159
160 pub async fn query(
162 &self,
163 query_vec: Vec<f32>,
164 top_k: usize,
165 ) -> Result<Vec<Document>, VectorStoreError> {
166 let body = Self::build_query_body(&query_vec, top_k);
167 let url = format!("{}/query", self.host);
168 let resp = self
169 .http
170 .post_json_with(&url, &body, self.request_options())
171 .await
172 .map_err(|e| Self::map_http_error("Pinecone query error", e))?;
173 let query_resp: QueryResponse = serde_json::from_str(&resp.body).map_err(|e| {
174 VectorStoreError::StorageError(format!("failed to parse Pinecone query: {e}"))
175 })?;
176 let result = query_resp
177 .matches
178 .into_iter()
179 .map(|m| {
180 let content = m
181 .metadata
182 .as_ref()
183 .and_then(|md| md.get("content").and_then(|v| v.as_str()))
184 .unwrap_or_default()
185 .to_string();
186 Document {
187 content,
188 metadata: m.metadata.unwrap_or_default(),
189 id: Some(m.id),
190 }
191 })
192 .collect();
193 Ok(result)
194 }
195
196 pub async fn describe_index_stats(&self) -> Result<PineconeIndexStats, VectorStoreError> {
200 let url = format!("{}/describe_index_stats", self.host);
201 let resp = self
202 .http
203 .post_json_with(&url, &serde_json::json!({}), self.request_options())
204 .await
205 .map_err(|e| Self::map_http_error("Pinecone describe_index_stats error", e))?;
206 serde_json::from_str(&resp.body).map_err(|e| {
207 VectorStoreError::StorageError(format!(
208 "failed to parse describe_index_stats response: {e}"
209 ))
210 })
211 }
212
213 pub async fn delete(&self, ids: &[String]) -> Result<(), VectorStoreError> {
215 let url = format!("{}/vectors/delete", self.host);
216 let body = serde_json::json!({ "ids": ids });
217 self.http
218 .post_json_with(&url, &body, self.request_options())
219 .await
220 .map_err(|e| Self::map_http_error("Pinecone delete error", e))?;
221 Ok(())
222 }
223
224 async fn query_impl(
226 &self,
227 body: serde_json::Value,
228 ) -> Result<Vec<SearchResult>, VectorStoreError> {
229 let url = format!("{}/query", self.host);
230 let resp = self
231 .http
232 .post_json_with(&url, &body, self.request_options())
233 .await
234 .map_err(|e| Self::map_http_error("Pinecone query HTTP error", e))?;
235
236 let query_resp: QueryResponse = serde_json::from_str(&resp.body).map_err(|e| {
237 VectorStoreError::StorageError(format!("Pinecone query parse error: {e}"))
238 })?;
239
240 Ok(query_resp
241 .matches
242 .into_iter()
243 .map(|m| {
244 let content = m
245 .metadata
246 .as_ref()
247 .and_then(|md| md.get("content").and_then(|v| v.as_str()))
248 .unwrap_or_default()
249 .to_string();
250 let doc = Document {
251 content,
252 metadata: m.metadata.unwrap_or_default(),
253 id: Some(m.id.clone()),
254 };
255 SearchResult {
256 document: doc,
257 score: m.score as f32,
258 }
259 })
260 .collect())
261 }
262}
263
264pub fn filter_to_pinecone(filter: &MetadataFilter) -> serde_json::Value {
271 fn op_str(op: FilterOp) -> &'static str {
272 match op {
273 FilterOp::Eq => "$eq",
274 FilterOp::Ne => "$ne",
275 FilterOp::Gt => "$gt",
276 FilterOp::Gte => "$gte",
277 FilterOp::Lt => "$lt",
278 FilterOp::Lte => "$lte",
279 FilterOp::In => "$in",
280 FilterOp::Nin => "$nin",
281 }
282 }
283 match filter {
284 MetadataFilter::Field { key, op, value } => {
285 serde_json::json!({ key.clone(): { op_str(*op): value.clone() } })
286 }
287 MetadataFilter::And(filters) => {
288 let items: Vec<serde_json::Value> = filters.iter().map(filter_to_pinecone).collect();
289 serde_json::json!({ "$and": items })
290 }
291 MetadataFilter::Or(filters) => {
292 let items: Vec<serde_json::Value> = filters.iter().map(filter_to_pinecone).collect();
293 serde_json::json!({ "$or": items })
294 }
295 }
296}
297
298#[async_trait]
299impl VectorStore for PineconeStore {
300 async fn add_documents(
301 &self,
302 documents: Vec<Document>,
303 embeddings: Vec<Vec<f32>>,
304 ) -> Result<Vec<String>, VectorStoreError> {
305 let ids: Vec<String> = documents
306 .iter()
307 .map(|d| {
308 d.id.clone()
309 .unwrap_or_else(|| uuid::Uuid::new_v4().to_string())
310 })
311 .collect();
312
313 let body = Self::build_upsert_body(&documents, &embeddings, &ids)?;
319 let url = format!("{}/vectors/upsert", self.host);
320 self.http
321 .post_json_with(&url, &body, self.request_options())
322 .await
323 .map_err(|e| Self::map_http_error("Pinecone upsert HTTP error", e))?;
324
325 Ok(ids)
326 }
327
328 async fn similarity_search(
329 &self,
330 query_embedding: &[f32],
331 k: usize,
332 ) -> Result<Vec<SearchResult>, VectorStoreError> {
333 let body = Self::build_query_body(query_embedding, k);
334 self.query_impl(body).await
335 }
336
337 async fn similarity_search_with_filter(
339 &self,
340 query_embedding: &[f32],
341 k: usize,
342 filter: Option<&MetadataFilter>,
343 ) -> Result<Vec<SearchResult>, VectorStoreError> {
344 let body = match filter {
345 Some(f) => Self::build_query_body_filtered(query_embedding, k, f),
346 None => Self::build_query_body(query_embedding, k),
347 };
348 self.query_impl(body).await
349 }
350
351 async fn get_document(&self, _id: &str) -> Result<Option<Document>, VectorStoreError> {
352 Err(VectorStoreError::StorageError(
355 "Pinecone does not support direct document fetch by ID via HTTP API".to_string(),
356 ))
357 }
358
359 async fn get_embedding(&self, _id: &str) -> Result<Option<Vec<f32>>, VectorStoreError> {
360 Err(VectorStoreError::StorageError(
361 "Pinecone does not support direct embedding fetch by ID via HTTP API".to_string(),
362 ))
363 }
364
365 async fn delete_document(&self, id: &str) -> Result<(), VectorStoreError> {
366 self.delete(&[id.to_string()]).await
367 }
368
369 async fn count(&self) -> usize {
370 match self.describe_index_stats().await {
373 Ok(stats) => stats.total_vector_count,
374 Err(e) => {
375 log::warn!("Pinecone count failed, treating as 0: {}", e);
376 0
377 }
378 }
379 }
380
381 async fn clear(&self) -> Result<(), VectorStoreError> {
382 Err(VectorStoreError::StorageError(
383 "Pinecone does not support clearing all vectors via HTTP API. Delete by namespace or IDs instead.".to_string()
384 ))
385 }
386}
387
388#[derive(Deserialize)]
389struct QueryResponse {
390 matches: Vec<QueryMatch>,
391}
392
393#[derive(Deserialize)]
394struct QueryMatch {
395 id: String,
396 score: f64,
397 metadata: Option<HashMap<String, serde_json::Value>>,
398}
399
400#[derive(Deserialize)]
405pub struct PineconeIndexStats {
406 #[serde(default)]
408 pub total_vector_count: usize,
409}
410
411#[cfg(test)]
412mod tests {
413 use super::*;
414 use tokio::io::{AsyncReadExt, AsyncWriteExt};
415 use tokio::net::TcpListener;
416
417 async fn spawn_loopback(
422 handler: impl Fn(&str, &str) -> (u16, String) + Send + Sync + 'static,
423 ) -> String {
424 use std::sync::Arc;
425 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
426 let addr = listener.local_addr().unwrap();
427 let handler = Arc::new(handler);
428 tokio::spawn(async move {
429 loop {
430 let Ok((mut socket, _)) = listener.accept().await else {
431 break;
432 };
433 let handler = handler.clone();
434 tokio::spawn(async move {
435 let mut raw = Vec::new();
436 let mut buf = [0u8; 4096];
437 let (head_end, len) = loop {
438 if let Some(pos) = raw.windows(4).position(|w| w == b"\r\n\r\n") {
439 let text = String::from_utf8_lossy(&raw).to_string();
440 let len = text
441 .lines()
442 .find_map(|l| {
443 let lower = l.to_ascii_lowercase();
444 lower
445 .strip_prefix("content-length:")
446 .map(|v| v.trim().to_string())
447 })
448 .and_then(|v| v.parse::<usize>().ok())
449 .unwrap_or(0);
450 break (pos + 4, len);
451 }
452 let n = socket.read(&mut buf).await.unwrap_or(0);
453 if n == 0 {
454 break (raw.len(), 0);
455 }
456 raw.extend_from_slice(&buf[..n]);
457 };
458 while raw.len() < head_end + len {
459 let n = socket.read(&mut buf).await.unwrap_or(0);
460 if n == 0 {
461 break;
462 }
463 raw.extend_from_slice(&buf[..n]);
464 }
465 let text = String::from_utf8_lossy(&raw).to_string();
466 let path = text.split_whitespace().nth(1).unwrap_or("/").to_string();
467 let (status, resp_body) = handler(&path, &text);
468 let reason = if status == 200 { "OK" } else { "Error" };
469 let head = format!(
470 "HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
471 resp_body.len()
472 );
473 let _ = socket.write_all(head.as_bytes()).await;
474 let _ = socket.write_all(resp_body.as_bytes()).await;
475 let _ = socket.shutdown().await;
476 });
477 }
478 });
479 format!("http://{}", addr)
480 }
481
482 fn doc(id: &str, content: &str) -> Document {
483 Document {
484 content: content.to_string(),
485 metadata: HashMap::new(),
486 id: Some(id.to_string()),
487 }
488 }
489
490 #[test]
491 fn test_build_upsert_body() {
492 let docs = vec![doc("1", "hello"), doc("2", "world")];
493 let vectors = vec![vec![1.0, 2.0], vec![3.0, 4.0]];
494 let ids = vec!["1".to_string(), "2".to_string()];
495 let body = PineconeStore::build_upsert_body(&docs, &vectors, &ids).unwrap();
496 let vectors_arr = body.get("vectors").unwrap().as_array().unwrap();
497 assert_eq!(vectors_arr.len(), 2);
498 assert_eq!(vectors_arr[0]["id"], "1");
499 assert_eq!(vectors_arr[0]["values"][0], 1.0);
500 assert_eq!(vectors_arr[0]["metadata"]["content"], "hello");
502 }
503
504 #[test]
509 fn test_build_upsert_body_uses_supplied_ids() {
510 let mut d = doc("", "x");
511 d.id = None;
512 let ids = vec![uuid::Uuid::new_v4().to_string()];
513 let body = PineconeStore::build_upsert_body(&[d], &[vec![0.1]], &ids).unwrap();
514 assert_eq!(body["vectors"][0]["id"], ids[0]);
515 }
516
517 #[test]
518 fn test_build_upsert_body_rejects_shape_mismatch() {
519 let docs = vec![doc("1", "a")];
520 let ids = vec!["1".to_string(), "2".to_string()]; let err = PineconeStore::build_upsert_body(&docs, &[vec![0.1]], &ids).unwrap_err();
522 assert!(err.to_string().contains("shape mismatch"), "message: {err}");
523 }
524
525 #[test]
528 fn test_build_upsert_body_respects_caller_content_key() {
529 let mut d = doc("1", "doc text");
530 d.metadata
531 .insert("content".to_string(), serde_json::json!("custom"));
532 let body = PineconeStore::build_upsert_body(&[d], &[vec![1.0]], &["1".to_string()]).unwrap();
533 assert_eq!(body["vectors"][0]["metadata"]["content"], "custom");
534 }
535
536 #[test]
539 fn test_build_upsert_body_rejects_oversized_metadata() {
540 let mut d = doc("1", "text");
541 d.metadata.insert(
542 "blob".to_string(),
543 serde_json::Value::String("x".repeat(40 * 1024 + 1)),
544 );
545 let err =
546 PineconeStore::build_upsert_body(&[d], &[vec![1.0]], &["1".to_string()]).unwrap_err();
547 assert!(err.to_string().contains("40 KiB"), "message: {err}");
548 }
549
550 #[test]
551 fn test_build_query_body() {
552 let body = PineconeStore::build_query_body(&[1.0, 2.0, 3.0], 5);
553 assert_eq!(body["topK"], 5);
554 assert_eq!(body["includeMetadata"], true);
555 assert_eq!(body["vector"][2], 3.0);
556 }
557
558 #[test]
559 fn test_new() {
560 let store = PineconeStore::new("key", "https://index.svc.env.pinecone.io");
561 assert_eq!(store.host, "https://index.svc.env.pinecone.io");
562 }
563
564 #[test]
566 fn test_filter_to_pinecone_field_ops() {
567 assert_eq!(
568 filter_to_pinecone(&MetadataFilter::field("lang", FilterOp::Eq, "rust")),
569 serde_json::json!({ "lang": { "$eq": "rust" } })
570 );
571 assert_eq!(
572 filter_to_pinecone(&MetadataFilter::field("year", FilterOp::Gte, 2020)),
573 serde_json::json!({ "year": { "$gte": 2020 } })
574 );
575 assert_eq!(
576 filter_to_pinecone(&MetadataFilter::field("tags", FilterOp::Nin, vec!["blog"])),
577 serde_json::json!({ "tags": { "$nin": ["blog"] } })
578 );
579 }
580
581 #[test]
583 fn test_filter_to_pinecone_and_or() {
584 let f = MetadataFilter::and(vec![
585 MetadataFilter::field("lang", FilterOp::Eq, "rust"),
586 MetadataFilter::or(vec![
587 MetadataFilter::field("year", FilterOp::Gte, 2020),
588 MetadataFilter::field("tags", FilterOp::In, vec!["ml"]),
589 ]),
590 ]);
591 assert_eq!(
592 filter_to_pinecone(&f),
593 serde_json::json!({
594 "$and": [
595 { "lang": { "$eq": "rust" } },
596 { "$or": [
597 { "year": { "$gte": 2020 } },
598 { "tags": { "$in": ["ml"] } }
599 ]}
600 ]
601 })
602 );
603 }
604
605 #[test]
607 fn test_build_query_body_filtered() {
608 let f = MetadataFilter::field("lang", FilterOp::Eq, "rust");
609 let body = PineconeStore::build_query_body_filtered(&[1.0, 2.0], 5, &f);
610 assert_eq!(body["topK"], 5);
611 assert_eq!(body["includeMetadata"], true);
612 assert_eq!(
613 body["filter"],
614 serde_json::json!({ "lang": { "$eq": "rust" } })
615 );
616 }
617
618 #[tokio::test]
622 async fn test_query_recovers_metadata_content_roundtrip() {
623 let base = spawn_loopback(|_path, _request| {
624 (
625 200,
626 serde_json::json!({
627 "matches": [{
628 "id": "abc",
629 "score": 1.0,
630 "metadata": { "content": "hello world" }
631 }]
632 })
633 .to_string(),
634 )
635 })
636 .await;
637 let store = PineconeStore::new("key", &base);
638 let docs = store.query(vec![1.0, 2.0], 3).await.unwrap();
639 assert_eq!(docs.len(), 1);
640 assert_eq!(docs[0].content, "hello world");
641 assert_eq!(docs[0].id.as_deref(), Some("abc"));
642 }
643
644 #[tokio::test]
647 async fn test_http_sends_api_key_header() {
648 let seen = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
649 let seen_h = seen.clone();
650 let base = spawn_loopback(move |_path, request| {
651 let auth: String = request
652 .lines()
653 .find(|l| l.to_ascii_lowercase().starts_with("api-key:"))
654 .unwrap_or("")
655 .to_string();
656 let mut guard = seen_h.lock().unwrap();
657 *guard = auth;
658 (200, serde_json::json!({ "matches": [] }).to_string())
659 })
660 .await;
661 let store = PineconeStore::new("pinecone-key", &base);
662 let _ = store.query(vec![1.0], 1).await.unwrap();
663 let auth = &*seen.lock().unwrap();
664 assert!(
665 auth.to_ascii_lowercase().contains("api-key: pinecone-key"),
666 "expected Api-Key header, got: {auth:?}"
667 );
668 }
669
670 #[tokio::test]
673 async fn test_non_2xx_surfaces_body_text() {
674 let base =
675 spawn_loopback(|_path, _request| (400, r#"{"message":"quota exceeded"}"#.to_string()))
676 .await;
677 let store = PineconeStore::new("key", &base);
678 let err = store.query(vec![1.0], 1).await.unwrap_err();
679 let msg = err.to_string();
680 assert!(msg.contains("400"), "message: {msg}");
681 assert!(msg.contains("quota exceeded"), "message: {msg}");
682 }
683}