diff --git a/crates/dbx-core/src/db/elasticsearch_driver.rs b/crates/dbx-core/src/db/elasticsearch_driver.rs index f8bc9e5b2..4a91f8cf9 100644 --- a/crates/dbx-core/src/db/elasticsearch_driver.rs +++ b/crates/dbx-core/src/db/elasticsearch_driver.rs @@ -324,6 +324,22 @@ fn elasticsearch_document_path(index: &str, id: &str, routing: Option<&str>) -> elasticsearch_path_with_routing_refresh(base, routing) } +fn elasticsearch_update_document_path( + index: &str, + id: &str, + document_type: Option<&str>, + routing: Option<&str>, +) -> String { + let document_type = document_type.map(str::trim).filter(|value| !value.is_empty()).unwrap_or("_doc"); + let base = format!( + "/{}/{}/{}", + elasticsearch_path_segment(index), + elasticsearch_path_segment(document_type), + elasticsearch_path_segment(id) + ); + elasticsearch_path_with_routing_refresh(base, routing) +} + /// Auto-id index path: `POST /{index}/_doc` with optional custom routing. fn elasticsearch_auto_id_document_path(index: &str, routing: Option<&str>) -> String { let base = format!("/{}/_doc", elasticsearch_path_segment(index)); @@ -531,6 +547,8 @@ impl<'de> Deserialize<'de> for HitsTotal { struct SearchHit { #[serde(rename = "_id")] id: String, + #[serde(rename = "_type")] + document_type: Option, #[serde(rename = "_routing")] routing: Option, #[serde(rename = "_source")] @@ -622,6 +640,9 @@ fn search_response_to_document_result(result: SearchResponse) -> Result serde_json::Map::new(), }; doc.insert("_id".to_string(), serde_json::Value::String(hit.id)); + if let Some(document_type) = hit.document_type.filter(|value| value != "_doc") { + doc.insert("_type".to_string(), serde_json::Value::String(document_type)); + } if let Some(routing) = hit.routing { doc.insert("_routing".to_string(), serde_json::Value::String(routing)); } @@ -903,9 +924,9 @@ pub async fn update_document( doc_json: &str, routing: Option<&str>, ) -> Result { - let (doc, routing) = elasticsearch_document_body_and_routing_from_json(doc_json, routing)?; + let (doc, routing, document_type) = elasticsearch_update_document_body_and_metadata(doc_json, routing)?; - let path = elasticsearch_document_path(index, id, routing.as_deref()); + let path = elasticsearch_update_document_path(index, id, document_type.as_deref(), routing.as_deref()); let resp = client.put(&path).json(&doc).send().await.map_err(|e| format!("Elasticsearch request failed: {e}"))?; if !client.response_status(&resp).is_success() { @@ -916,6 +937,24 @@ pub async fn update_document( Ok(1) } +fn elasticsearch_update_document_body_and_metadata( + doc_json: &str, + routing: Option<&str>, +) -> Result<(serde_json::Value, Option, Option), String> { + let (mut doc, routing) = elasticsearch_document_body_and_routing_from_json(doc_json, routing)?; + let document_type = match &mut doc { + serde_json::Value::Object(map) => map.remove("_type").and_then(|value| match value { + serde_json::Value::String(value) => { + let trimmed = value.trim(); + (!trimmed.is_empty()).then(|| trimmed.to_string()) + } + _ => None, + }), + _ => None, + }; + Ok((doc, routing, document_type)) +} + fn elasticsearch_document_body_and_routing_from_json( doc_json: &str, routing: Option<&str>, @@ -2293,6 +2332,19 @@ mod tests { assert_eq!(super::elasticsearch_document_path("orders", "1", None), "/orders/_doc/1?refresh=true"); } + #[test] + fn builds_legacy_elasticsearch_update_path_from_document_type() { + assert_eq!( + super::elasticsearch_update_document_path("orders/2026", "a%b/c", Some("legacy/order"), Some("tenant/a&b")), + "/orders%2F2026/legacy%2Forder/a%25b%2Fc?routing=tenant%2Fa%26b&refresh=true" + ); + assert_eq!( + super::elasticsearch_update_document_path("orders", "1", Some("_doc"), None), + "/orders/_doc/1?refresh=true" + ); + assert_eq!(super::elasticsearch_update_document_path("orders", "1", None, None), "/orders/_doc/1?refresh=true"); + } + #[test] fn builds_elasticsearch_auto_id_document_path_with_routing() { assert_eq!( @@ -2646,6 +2698,25 @@ mod tests { assert_eq!(response.hits.hits[0].routing.as_deref(), Some("tenant-1")); } + #[test] + fn preserves_legacy_elasticsearch_document_type_metadata() { + let response: SearchResponse = serde_json::from_value(json!({ + "hits": { + "total": { "value": 2, "relation": "eq" }, + "hits": [ + { "_id": "legacy-1", "_type": "order", "_source": { "name": "Legacy" } }, + { "_id": "modern-1", "_type": "_doc", "_source": { "name": "Modern" } } + ] + } + })) + .unwrap(); + + let result = super::search_response_to_document_result(response).unwrap(); + + assert_eq!(result.documents[0]["_type"], json!("order")); + assert!(result.documents[1].get("_type").is_none()); + } + #[test] fn preserves_long_literals_in_document_transport_json() { let response: SearchResponse = serde_json::from_str( @@ -3223,6 +3294,37 @@ mod tests { assert_eq!(result.rows[0][name_idx], json!("Notebook")); } + #[tokio::test] + async fn update_document_uses_legacy_type_path_without_storing_metadata() { + use tokio::io::AsyncWriteExt; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut socket).await; + assert!(request.starts_with("PUT /orders/order/abc?routing=tenant-1&refresh=true ")); + assert!(request.ends_with(r#"{"name":"Alice"}"#)); + assert!(!request.contains(r#""_type""#)); + assert!(!request.contains(r#""_routing""#)); + let response = + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}"; + socket.write_all(response.as_bytes()).await.unwrap(); + }); + + let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1)); + super::update_document( + &client, + "orders", + "abc", + r#"{"_id":"abc","_type":"order","_routing":"tenant-1","name":"Alice"}"#, + None, + ) + .await + .unwrap(); + server.await.unwrap(); + } + #[tokio::test] async fn execute_rest_search_preserves_full_json_response() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -3353,6 +3455,19 @@ mod tests { assert_eq!(routing.as_deref(), Some("tenant-1")); } + #[test] + fn update_document_body_extracts_legacy_type_metadata() { + let (doc, routing, document_type) = super::elasticsearch_update_document_body_and_metadata( + r#"{"_id":"abc","_type":"order","_routing":"tenant-1","name":"Alice"}"#, + None, + ) + .unwrap(); + + assert_eq!(doc, json!({ "name": "Alice" })); + assert_eq!(routing.as_deref(), Some("tenant-1")); + assert_eq!(document_type.as_deref(), Some("order")); + } + #[test] fn explicit_elasticsearch_routing_overrides_document_metadata() { let (doc, routing) = super::elasticsearch_document_body_and_routing_from_json(