feat: add bulk upsert by external_id (#102)

README.md

@@ -172,6 +172,25 @@ ctx.add_many([
     },
 ])
 
+# Bulk insert-or-replace by external_id (idempotent re-ingestion). New
+# external_ids are inserted; existing ones are replaced (the previous row is
+# superseded), all in one storage operation. Returns one result per record.
+results = ctx.upsert_many([
+    {
+        "role": "source",
+        "content": "Chunk 1, revised",
+        "content_type": "text/markdown",
+        "external_id": "doc-77#chunk-1",
+    },
+    {
+        "role": "source",
+        "content": "Brand new chunk",
+        "content_type": "text/markdown",
+        "external_id": "doc-77#chunk-2",
+    },
+])
+print([(r["inserted"], r["replaced_id"]) for r in results])
+
 # Deferred embeddings: raw-first capture, enrich later.
 #
 # Bulk ingestion often needs to persist source chunks immediately and compute

crates/lance-context-api/src/lib.rs

@@ -39,6 +39,11 @@ pub trait ContextStoreApi {
         request: &UpsertRecordRequest,
     ) -> impl Future<Output = ContextResult<UpsertRecordResponse>> + Send;
 
+    fn upsert_many(
+        &mut self,
+        request: &UpsertRecordsRequest,
+    ) -> impl Future<Output = ContextResult<UpsertRecordsResponse>> + Send;
+
     fn update(
         &mut self,
         request: &UpdateRecordRequest,
@@ -225,6 +230,28 @@ pub struct UpsertRecordResponse {
     pub record: RecordDto,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UpsertRecordsRequest {
+    pub records: Vec<AddRecordRequest>,
+    #[serde(default = "default_upsert_key")]
+    pub key: String,
+}
+
+/// Per-record outcome of a batch upsert, in input order.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UpsertResultDto {
+    pub inserted: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub replaced_id: Option<String>,
+    pub record: RecordDto,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UpsertRecordsResponse {
+    pub version: u64,
+    pub results: Vec<UpsertResultDto>,
+}
+
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct RecordPatchDto {
     #[serde(default, skip_serializing_if = "Option::is_none")]

crates/lance-context-client/src/lib.rs

@@ -71,6 +71,19 @@ impl ContextStoreApi for RemoteContextStore {
         Ok(resp)
     }
 
+    async fn upsert_many(
+        &mut self,
+        request: &UpsertRecordsRequest,
+    ) -> ContextResult<UpsertRecordsResponse> {
+        let resp = self
+            .client
+            .upsert_records(&self.context_name, request)
+            .await
+            .map_err(to_ctx_err)?;
+        self.cached_version = resp.version;
+        Ok(resp)
+    }
+
     async fn update(
         &mut self,
         request: &UpdateRecordRequest,
@@ -339,6 +352,20 @@ impl ContextClient {
         Self::handle_response(resp).await
     }
 
+    pub async fn upsert_records(
+        &self,
+        name: &str,
+        req: &UpsertRecordsRequest,
+    ) -> Result<UpsertRecordsResponse, ClientError> {
+        let resp = self
+            .http
+            .put(self.url(&format!("/contexts/{}/records/batch", name)))
+            .json(req)
+            .send()
+            .await?;
+        Self::handle_response(resp).await
+    }
+
     pub async fn update_record(
         &self,
         name: &str,

crates/lance-context-core/src/api_impl.rs

@@ -7,7 +7,7 @@ use lance_context_api::{
     ContextError, ContextResult, ContextStoreApi, DeleteRecordResponse, RecordDto, RecordPatchDto,
     RelationshipDto, RetrieveRequest, RetrieveResultDto, SearchRequest, SearchResultDto,
     StateMetadataDto, UpdateRecordRequest, UpdateRecordResponse, UpsertRecordRequest,
-    UpsertRecordResponse,
+    UpsertRecordResponse, UpsertRecordsRequest, UpsertRecordsResponse, UpsertResultDto,
 };
 
 use crate::record::{
@@ -74,6 +74,57 @@ impl ContextStoreApi for ContextStore {
         })
     }
 
+    async fn upsert_many(
+        &mut self,
+        request: &UpsertRecordsRequest,
+    ) -> ContextResult<UpsertRecordsResponse> {
+        if request.key != "external_id" {
+            return Err(ContextError::InvalidRequest(format!(
+                "upsert key '{}' is not supported; use 'external_id'",
+                request.key
+            )));
+        }
+        if request.records.is_empty() {
+            return Err(ContextError::InvalidRequest(
+                "upsert_many requires at least one record".to_string(),
+            ));
+        }
+        for (index, record) in request.records.iter().enumerate() {
+            if record.external_id.as_deref().is_none_or(str::is_empty) {
+                return Err(ContextError::InvalidRequest(format!(
+                    "upsert_many requires record.external_id (records[{index}])"
+                )));
+            }
+        }
+
+        let core_records: Vec<ContextRecord> = request
+            .records
+            .iter()
+            .map(|r| {
+                record_from_add_request(r, Uuid::new_v4().to_string(), Uuid::new_v4().to_string())
+            })
+            .collect();
+
+        let results = ContextStore::upsert_many_by_external_id(self, core_records)
+            .await
+            .map_err(to_ctx_err)?;
+        let version = results
+            .last()
+            .map(|r| r.version)
+            .unwrap_or_else(|| ContextStore::version(self));
+        Ok(UpsertRecordsResponse {
+            version,
+            results: results
+                .into_iter()
+                .map(|r| UpsertResultDto {
+                    inserted: r.inserted,
+                    replaced_id: r.replaced_id,
+                    record: record_to_dto(r.record),
+                })
+                .collect(),
+        })
+    }
+
     async fn update(
         &mut self,
         request: &UpdateRecordRequest,