use std::collections::{BTreeMap, BTreeSet};
use serde_json::{Value as JsonValue, json};
use sha2::{Digest, Sha256};
use crate::generation::ManifestStore;
use crate::proto::{
VectorHybridSearchRequest, VectorPoint, VectorPointMutation, VectorSearchRequest, VectorSet,
VectorUpsertRequest,
};
use crate::runtime::executor_utils::{
backend_transport_status, build_probe, capability_status, invalid_argument_fields, json_bool,
json_i32, json_into_struct, json_required_f32_vec, json_required_str, json_scalar_to_string,
json_to_struct, qdrant_status, store_option, store_option_i32, struct_to_json,
};
fn point_to_vector_point(mut point: JsonValue) -> VectorPoint {
let id = point
.get("id")
.map(json_scalar_to_string)
.unwrap_or_default();
let score = point
.get("score")
.and_then(JsonValue::as_f64)
.unwrap_or_default() as f32;
let payload = point
.get_mut("payload")
.map(JsonValue::take)
.and_then(json_into_struct);
let vector_value = point.get_mut("vector").map(JsonValue::take);
let (vector_name, vector) = match vector_value {
Some(JsonValue::Array(values)) => (String::new(), json_f32_array(&values)),
Some(JsonValue::Object(mut vectors)) => vectors
.iter_mut()
.next()
.map(|(name, values)| {
let name = name.clone();
let values = values.take();
let vector = values
.as_array()
.map_or_else(Vec::new, |values| json_f32_array(values));
(name, vector)
})
.unwrap_or_default(),
_ => (String::new(), Vec::new()),
};
VectorPoint {
id,
score,
payload,
vector,
vector_name,
}
}
fn json_f32_array(values: &[JsonValue]) -> Vec<f32> {
values
.iter()
.filter_map(JsonValue::as_f64)
.map(|value| value as f32)
.collect()
}
fn qdrant_point_id(id: &str) -> JsonValue {
if let Ok(n) = id.parse::<u64>() {
return json!(n);
}
if uuid::Uuid::parse_str(id).is_ok() {
return json!(id);
}
let digest = Sha256::digest(id.as_bytes());
json!(format!(
"{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-4{:01x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
digest[0],
digest[1],
digest[2],
digest[3],
digest[4],
digest[5],
digest[6] & 0x0f,
digest[7],
(digest[8] & 0x3f) | 0x80,
digest[9],
digest[10],
digest[11],
digest[12],
digest[13],
digest[14],
digest[15],
))
}
use crate::runtime::executors::{
BackendExecutor, BackendHealth, BackendProbe, MutationExecutor, ObjectExecutor, QueryExecutor,
ResourceAdminExecutor, SearchExecutor,
};
const QDRANT_DEFAULT_SEARCH_LIMIT: i32 = 10;
fn qdrant_invalid_field_status(
field: impl Into<String>,
description: impl Into<String>,
message: impl Into<String>,
) -> tonic::Status {
invalid_argument_fields(message, [(field.into(), description.into())])
}
fn invalid_qdrant_request_json_status(err: serde_json::Error) -> tonic::Status {
qdrant_invalid_field_status(
"request_json",
"must be valid JSON for Qdrant generic dispatch",
format!("invalid request json: {err}"),
)
}
fn invalid_qdrant_ensure_resource_spec_status(err: serde_json::Error) -> tonic::Status {
qdrant_invalid_field_status(
"spec_json",
"must be valid JSON for Qdrant ensure_resource",
format!("invalid qdrant ensure_resource spec: {err}"),
)
}
fn qdrant_required_field_status(field: &'static str, message: &'static str) -> tonic::Status {
qdrant_invalid_field_status(
field,
format!("{field} is required for this Qdrant operation"),
message,
)
}
fn unsupported_qdrant_operation_status(operation: &str) -> tonic::Status {
qdrant_invalid_field_status(
"operation",
"unsupported Qdrant mutation operation",
format!("unsupported Qdrant mutation operation '{operation}'"),
)
}
fn qdrant_collection_not_found_status(collection: &str) -> tonic::Status {
crate::runtime::executor_utils::schema_status(
tonic::Code::NotFound,
"qdrant",
"collection_check",
"qdrant_collection_not_found",
format!("Qdrant collection {collection} is missing"),
)
}
fn qdrant_internal_status(
operation: impl Into<String>,
message: impl Into<String>,
) -> tonic::Status {
crate::runtime::executor_utils::internal_status("qdrant", operation, message)
}
#[allow(clippy::result_large_err)]
fn validate_collection_name(name: &str) -> Result<(), tonic::Status> {
if name.is_empty() || name.len() > 255 {
return Err(qdrant_invalid_field_status(
"collection",
"must be 1-255 characters",
"Qdrant collection name must be 1–255 characters",
));
}
if !name
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-'))
{
return Err(qdrant_invalid_field_status(
"collection",
"may only contain ASCII letters, digits, hyphens, and underscores",
"Qdrant collection name may only contain ASCII letters, digits, hyphens, and underscores",
));
}
if name.starts_with('.') || name.starts_with('-') {
return Err(qdrant_invalid_field_status(
"collection",
"may not start with '.' or '-'",
"Qdrant collection name may not start with '.' or '-'",
));
}
Ok(())
}
fn encode_collection(name: &str) -> String {
name.chars()
.flat_map(|c| {
if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
vec![c]
} else {
format!("%{:02X}", c as u32).chars().collect()
}
})
.collect()
}
fn normalize_qdrant_distance(raw: &str) -> String {
let upper = raw.trim().to_ascii_uppercase();
let canonical = upper
.trim_start_matches("VECTOR_DISTANCE_")
.trim_start_matches("DISTANCE_");
match canonical {
"COSINE" => "Cosine",
"DOT" => "Dot",
"EUCLID" | "EUCLIDEAN" => "Euclid",
"MANHATTAN" => "Manhattan",
_ => "Cosine", }
.to_string()
}
fn configured_vector_names(store: &ManifestStore) -> Vec<String> {
serde_json::from_str::<Vec<String>>(&store_option(store, "vector_names_json"))
.unwrap_or_default()
.into_iter()
.map(|name| name.trim().to_string())
.filter(|name| !name.is_empty())
.collect::<BTreeSet<_>>()
.into_iter()
.collect()
}
fn query_result_points(mut payload: JsonValue) -> Vec<VectorPoint> {
let result = payload.get_mut("result").map(JsonValue::take);
let points = match result {
Some(JsonValue::Array(points)) => points,
Some(JsonValue::Object(mut result)) => result
.remove("points")
.and_then(|points| match points {
JsonValue::Array(points) => Some(points),
_ => None,
})
.unwrap_or_default(),
_ => Vec::new(),
};
points.into_iter().map(point_to_vector_point).collect()
}
#[derive(Debug, Clone)]
pub(crate) struct QdrantHttpClient {
pub(crate) base_url: String,
pub(crate) api_key: Option<String>,
pub(crate) http: reqwest::Client,
}
impl QdrantHttpClient {
pub(crate) async fn collection_exists(&self, collection: &str) -> Result<(), tonic::Status> {
validate_collection_name(collection)?;
let url = format!(
"{}/collections/{}",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.get(url))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection check", err))?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Err(qdrant_collection_not_found_status(collection));
}
qdrant_status(response.status())
}
pub(crate) async fn ensure_collection(
&self,
store: &ManifestStore,
) -> Result<(), tonic::Status> {
validate_collection_name(&store.resource_name)?;
let dimension = store_option_i32(store, "dimension").max(1);
let distance = normalize_qdrant_distance(&store_option(store, "distance"));
let output_dtype = store_option(store, "output_dtype")
.trim()
.to_ascii_uppercase();
let vector_names = configured_vector_names(store);
let quantization = match output_dtype.as_str() {
"INT8" | "UINT8" => json!({
"scalar": { "type": "int8", "quantile": 0.99, "always_ram": true }
}),
"BINARY" => json!({ "binary": { "always_ram": true } }),
_ => JsonValue::Null,
};
let url = format!(
"{}/collections/{}",
self.base_url,
encode_collection(&store.resource_name)
);
let vector_params = json!({
"size": dimension,
"distance": distance.as_str(),
});
let mut create = json!({
"vectors": vector_params,
"hnsw_config": { "m": 0, "payload_m": 16 }
});
if !quantization.is_null() {
create["quantization_config"] = quantization;
}
let response = self
.auth(self.http.put(url))
.json(&create)
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection create", err))?;
let status = response.status();
if status != reqwest::StatusCode::CONFLICT && !status.is_success() {
let body = response.text().await.unwrap_or_default();
let message = format!("Qdrant collection create failed: HTTP {status}: {body}");
if status.is_server_error() {
return Err(crate::runtime::executor_utils::retryable_status(
"Qdrant",
"collection create",
crate::runtime::executor_utils::HTTP_RETRYABLE_BACKOFF_MS,
message,
));
}
return Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection create",
"qdrant_collection_create_rejected",
message,
));
}
if !vector_names.is_empty() {
self.ensure_named_vectors(&store.resource_name, dimension, &distance, &vector_names)
.await?;
}
self.ensure_collection_policy(&store.resource_name).await?;
self.verify_collection_geometry(
&store.resource_name,
dimension,
&distance,
&output_dtype,
&vector_names,
)
.await?;
self.ensure_tenant_index(&store.resource_name).await
}
async fn ensure_collection_policy(&self, collection: &str) -> Result<(), tonic::Status> {
let url = format!(
"{}/collections/{}",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.patch(url))
.json(&json!({ "hnsw_config": { "m": 0, "payload_m": 16 } }))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection policy", err))?;
let status = response.status();
if status.is_success() {
return Ok(());
}
let body = response.text().await.unwrap_or_default();
Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection policy",
"qdrant_collection_policy_rejected",
format!(
"Qdrant collection '{collection}' rejected tenant HNSW policy: HTTP {status}: {body}"
),
))
}
async fn ensure_named_vectors(
&self,
collection: &str,
dimensions: i32,
distance: &str,
vector_names: &[String],
) -> Result<(), tonic::Status> {
let details_url = format!(
"{}/collections/{}",
self.base_url,
encode_collection(collection)
);
let details_response = self
.auth(self.http.get(details_url))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "named vector inspect", err))?;
qdrant_status(details_response.status())?;
let details = details_response.json::<JsonValue>().await.map_err(|err| {
backend_transport_status("Qdrant", "named vector inspect decode", err)
})?;
let existing = details
.pointer("/result/config/params/vectors")
.and_then(JsonValue::as_object);
for name in vector_names {
validate_collection_name(name).map_err(|_| {
qdrant_invalid_field_status(
"vector_name",
"may only contain ASCII letters, digits, hyphens, and underscores",
format!("invalid Qdrant vector name '{name}'"),
)
})?;
if existing.is_some_and(|vectors| vectors.contains_key(name)) {
continue;
}
let url = format!(
"{}/collections/{}/vectors/{}",
self.base_url,
encode_collection(collection),
encode_collection(name)
);
let response = self
.auth(self.http.put(url))
.json(&json!({
"dense": { "size": dimensions, "distance": distance }
}))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "named vector create", err))?;
let status = response.status();
if status == reqwest::StatusCode::CONFLICT || status.is_success() {
continue;
}
let body = response.text().await.unwrap_or_default();
return Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"named vector create",
"qdrant_named_vector_create_rejected",
format!(
"Qdrant named vector '{name}' creation failed for collection '{collection}': HTTP {status}: {body}"
),
));
}
Ok(())
}
async fn verify_collection_geometry(
&self,
collection: &str,
dimensions: i32,
distance: &str,
output_dtype: &str,
vector_names: &[String],
) -> Result<(), tonic::Status> {
let url = format!(
"{}/collections/{}",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.get(url))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection geometry", err))?;
qdrant_status(response.status())?;
let payload = response
.json::<JsonValue>()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection geometry decode", err))?;
let vectors = payload
.pointer("/result/config/params/vectors")
.ok_or_else(|| {
qdrant_internal_status(
"collection_geometry",
"Qdrant collection response omitted vector configuration",
)
})?;
let configs = if vector_names.is_empty() {
vec![("", vectors)]
} else {
vector_names
.iter()
.map(|name| {
vectors
.get(name)
.map(|config| (name.as_str(), config))
.ok_or_else(|| {
crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection geometry",
"qdrant_named_vector_missing",
format!(
"Qdrant collection '{collection}' does not define named vector '{name}'"
),
)
})
})
.collect::<Result<Vec<_>, _>>()?
};
for (name, config) in configs {
let actual_dimensions = config
.get("size")
.and_then(JsonValue::as_i64)
.unwrap_or_default();
let actual_distance = config
.get("distance")
.and_then(JsonValue::as_str)
.unwrap_or_default();
if actual_dimensions != i64::from(dimensions)
|| !actual_distance.eq_ignore_ascii_case(distance)
{
let vector_label = if name.is_empty() { "default" } else { name };
return Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection geometry",
"qdrant_vector_geometry_mismatch",
format!(
"Qdrant collection '{collection}' vector '{vector_label}' has geometry \
({actual_dimensions}, {actual_distance}), expected ({dimensions}, {distance})"
),
));
}
}
let quantization = payload.pointer("/result/config/quantization_config");
let quantization_matches = match output_dtype {
"INT8" | "UINT8" => quantization.and_then(|value| value.get("scalar")).is_some(),
"BINARY" => quantization.and_then(|value| value.get("binary")).is_some(),
_ => quantization.is_none_or(JsonValue::is_null),
};
if !quantization_matches {
return Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection geometry",
"qdrant_quantization_mismatch",
format!(
"Qdrant collection '{collection}' quantization does not match output dtype '{output_dtype}'"
),
));
}
let hnsw = payload
.pointer("/result/config/hnsw_config")
.ok_or_else(|| {
qdrant_internal_status(
"collection_geometry",
"Qdrant collection response omitted HNSW configuration",
)
})?;
if hnsw.get("m").and_then(JsonValue::as_i64) != Some(0)
|| hnsw.get("payload_m").and_then(JsonValue::as_i64) != Some(16)
{
return Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"collection geometry",
"qdrant_tenant_hnsw_mismatch",
format!("Qdrant collection '{collection}' must use hnsw m=0 and payload_m=16"),
));
}
Ok(())
}
pub(crate) async fn ensure_tenant_index(&self, collection: &str) -> Result<(), tonic::Status> {
validate_collection_name(collection)?;
let url = format!(
"{}/collections/{}/index?wait=true",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.put(url))
.json(&json!({
"field_name": "_tenant_id",
"field_schema": { "type": "keyword", "is_tenant": true }
}))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "tenant payload index", err))?;
let status = response.status();
if status == reqwest::StatusCode::CONFLICT || status.is_success() {
return Ok(());
}
let body = response.text().await.unwrap_or_default();
Err(crate::runtime::executor_utils::schema_status(
tonic::Code::FailedPrecondition,
"Qdrant",
"tenant payload index",
"qdrant_tenant_index_rejected",
format!("Qdrant tenant payload index failed: HTTP {status}: {body}"),
))
}
pub(crate) async fn swap_collection_alias(
&self,
alias: &str,
collection: &str,
) -> Result<(), tonic::Status> {
validate_collection_name(alias)?;
validate_collection_name(collection)?;
self.collection_exists(collection).await?;
let aliases_url = format!("{}/aliases", self.base_url);
let aliases_response = self
.auth(self.http.get(&aliases_url))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "list collection aliases", err))?;
let aliases_status = aliases_response.status();
if !aliases_status.is_success() {
return qdrant_status(aliases_status);
}
let aliases: JsonValue = aliases_response.json().await.map_err(|err| {
crate::runtime::executor_utils::schema_status(
tonic::Code::Internal,
"Qdrant",
"list collection aliases",
"qdrant_alias_response_invalid",
format!("invalid Qdrant alias response: {err}"),
)
})?;
let exists = aliases
.pointer("/result/aliases")
.and_then(JsonValue::as_array)
.is_some_and(|items| {
items
.iter()
.any(|item| item.get("alias_name").and_then(JsonValue::as_str) == Some(alias))
});
let url = format!("{}/collections/aliases", self.base_url);
let actions = if exists {
json!([
{ "delete_alias": { "alias_name": alias } },
{ "create_alias": { "collection_name": collection, "alias_name": alias } }
])
} else {
json!([{ "create_alias": { "collection_name": collection, "alias_name": alias } }])
};
let response = self
.auth(self.http.post(url))
.json(&json!({
"actions": actions
}))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "collection alias cutover", err))?;
qdrant_status(response.status())
}
pub(crate) async fn search(
&self,
request: &VectorSearchRequest,
filter: JsonValue,
) -> Result<VectorSet, tonic::Status> {
validate_collection_name(&request.collection)?;
let mut body = json!({
"query": request.vector,
"limit": if request.limit > 0 {
request.limit
} else {
QDRANT_DEFAULT_SEARCH_LIMIT
},
"with_payload": request.with_payload,
"with_vector": request.with_vector,
});
if !request.vector_name.trim().is_empty() {
body["using"] = json!(request.vector_name);
}
if !filter.is_null() {
body["filter"] = filter;
}
if request.score_threshold > 0.0 {
body["score_threshold"] = json!(request.score_threshold);
}
if request.quantization_rescore {
body["params"] = json!({ "quantization": { "rescore": true } });
}
let url = format!(
"{}/collections/{}/points/query",
self.base_url,
encode_collection(&request.collection)
);
let response = self
.auth(self.http.post(url))
.json(&body)
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "search", err))?;
qdrant_status(response.status())?;
let payload: JsonValue = response
.json()
.await
.map_err(|err| backend_transport_status("Qdrant", "response decode", err))?;
let points = query_result_points(payload);
Ok(VectorSet { points })
}
pub(crate) async fn upsert(&self, request: &VectorUpsertRequest) -> Result<(), tonic::Status> {
validate_collection_name(&request.collection)?;
let mut grouped = BTreeMap::<
String,
(
JsonValue,
JsonValue,
Option<Vec<f32>>,
serde_json::Map<String, JsonValue>,
),
>::new();
for point in &request.points {
let entry = grouped.entry(point.id.clone()).or_insert_with(|| {
(
qdrant_point_id(&point.id),
point
.payload
.as_ref()
.map(struct_to_json)
.unwrap_or(JsonValue::Null),
None,
serde_json::Map::new(),
)
});
if let Some(payload) = point.payload.as_ref() {
entry.1 = struct_to_json(payload);
}
if point.vector_name.trim().is_empty() {
if !entry.3.is_empty() {
return Err(qdrant_invalid_field_status(
"points.vector_name",
"a point cannot mix unnamed and named vectors",
format!("point '{}' mixes unnamed and named vectors", point.id),
));
}
entry.2 = Some(point.vector.clone());
} else {
if entry.2.is_some() {
return Err(qdrant_invalid_field_status(
"points.vector_name",
"a point cannot mix unnamed and named vectors",
format!("point '{}' mixes unnamed and named vectors", point.id),
));
}
entry
.3
.insert(point.vector_name.trim().to_string(), json!(point.vector));
}
}
let points = grouped
.into_values()
.map(|(id, payload, unnamed, named)| {
let vector = unnamed
.map(|vector| json!(vector))
.unwrap_or_else(|| JsonValue::Object(named));
json!({ "id": id, "vector": vector, "payload": payload })
})
.collect::<Vec<_>>();
let url = format!(
"{}/collections/{}/points?wait=true",
self.base_url,
encode_collection(&request.collection)
);
let response = self
.auth(self.http.put(url))
.json(&json!({ "points": points }))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "upsert", err))?;
qdrant_status(response.status())?;
Ok(())
}
pub(crate) async fn delete_points(
&self,
collection: &str,
point_ids: &[String],
) -> Result<(), tonic::Status> {
validate_collection_name(collection)?;
if point_ids.is_empty() {
return Ok(());
}
let url = format!(
"{}/collections/{}/points/delete?wait=true",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.post(url))
.json(&json!({
"points": point_ids.iter().map(|id| qdrant_point_id(id)).collect::<Vec<_>>()
}))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "delete", err))?;
qdrant_status(response.status())?;
Ok(())
}
pub(crate) async fn delete_by_filter(
&self,
collection: &str,
filter: JsonValue,
) -> Result<(), tonic::Status> {
validate_collection_name(collection)?;
let url = format!(
"{}/collections/{}/points/delete?wait=true",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.post(url))
.json(&json!({ "filter": filter }))
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "filtered delete", err))?;
qdrant_status(response.status())?;
Ok(())
}
pub(crate) async fn set_payload(
&self,
collection: &str,
payload: JsonValue,
point_ids: Option<Vec<String>>,
filter: Option<JsonValue>,
) -> Result<(), tonic::Status> {
validate_collection_name(collection)?;
let mut body = json!({ "payload": payload });
if let Some(ids) = point_ids {
body["points"] = json!(ids);
}
if let Some(filter) = filter {
body["filter"] = filter;
}
let url = format!(
"{}/collections/{}/points/payload?wait=true",
self.base_url,
encode_collection(collection)
);
let response = self
.auth(self.http.post(url))
.json(&body)
.send()
.await
.map_err(|err| backend_transport_status("Qdrant", "payload patch", err))?;
qdrant_status(response.status())?;
Ok(())
}
pub(crate) async fn hybrid_search(
&self,
request: &VectorHybridSearchRequest,
filter: JsonValue,
) -> Result<VectorSet, tonic::Status> {
validate_collection_name(&request.collection)?;
let limit = if request.limit > 0 {
request.limit as usize
} else {
QDRANT_DEFAULT_SEARCH_LIMIT as usize
};
let text_query = request.text_query.trim().to_lowercase();
let dense_weight = request.fusion_weights.first().copied().unwrap_or(0.7) as f64;
let sparse_weight = request.fusion_weights.get(1).copied().unwrap_or(0.3) as f64;
if !request.vector.is_empty() {
let prefetch_limit = if request.prefetch_limit > 0 {
request.prefetch_limit as usize
} else {
(limit * 4).max(50)
};
let mut dense_prefetch = json!({
"query": request.vector,
"limit": prefetch_limit
});
if !request.vector_name.trim().is_empty() {
dense_prefetch["using"] = json!(request.vector_name);
}
let mut prefetch = vec![dense_prefetch];
if !text_query.is_empty() {
prefetch.push(json!({
"query": request.vector,
"filter": {
"must": [{
"key": "_full_text",
"match": { "text": text_query }
}]
},
"limit": prefetch_limit
}));
}
let fusion = match request.fusion_strategy {
3 => "dbsf",
_ => "rrf",
};
let mut query_body = json!({
"prefetch": prefetch,
"query": { "fusion": fusion },
"limit": (limit * 2).max(20),
"with_payload": request.with_payload,
"with_vector": request.with_vector,
});
if request.quantization_rescore {
query_body["params"] = json!({ "quantization": { "rescore": true } });
}
if !filter.is_null() {
query_body["filter"] = filter.clone();
}
let url = format!(
"{}/collections/{}/points/query",
self.base_url,
encode_collection(&request.collection)
);
if let Ok(resp) = self
.auth(self.http.post(&url))
.json(&query_body)
.send()
.await
&& resp.status().is_success()
&& let Ok(payload) = resp.json::<JsonValue>().await
{
let points = query_result_points(payload);
let reranked = if text_query.is_empty() {
points.into_iter().take(limit).collect()
} else {
rerank_with_text(points, &text_query, dense_weight, sparse_weight, limit)
};
return Ok(VectorSet { points: reranked });
}
}
let fetch_limit = if text_query.is_empty() {
limit as i32
} else {
(limit * 4).max(40) as i32
};
let dense_req = VectorSearchRequest {
context: request.context.clone(),
collection: request.collection.clone(),
vector: request.vector.clone(),
filter: request.filter.clone(),
limit: fetch_limit,
score_threshold: 0.0,
with_payload: true,
with_vector: request.with_vector,
vector_name: request.vector_name.clone(),
quantization_rescore: request.quantization_rescore,
};
let result = self.search(&dense_req, filter).await?;
if text_query.is_empty() {
return Ok(result);
}
let reranked = rerank_with_text(
result.points,
&text_query,
dense_weight,
sparse_weight,
limit,
);
Ok(VectorSet { points: reranked })
}
pub(crate) fn auth(&self, builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
let Some(api_key) = &self.api_key else {
return builder;
};
match reqwest::header::HeaderValue::from_str(api_key) {
Ok(value) => builder.header("api-key", value),
Err(err) => {
tracing::error!(
"Qdrant 'api-key' header value is invalid ({err}); it likely contains \
whitespace/control characters (e.g. a trailing CRLF from a Windows .env). \
The request will fail — set UDB_QDRANT_API_KEY to a clean value"
);
builder.header("api-key", api_key)
}
}
}
}
fn rerank_with_text(
points: Vec<VectorPoint>,
text_query: &str,
dense_weight: f64,
sparse_weight: f64,
limit: usize,
) -> Vec<VectorPoint> {
if points.is_empty() {
return vec![];
}
let query_tokens: Vec<String> = text_query
.split_whitespace()
.map(|t| t.to_lowercase())
.collect();
if query_tokens.is_empty() {
return points.into_iter().take(limit).collect();
}
let total = dense_weight + sparse_weight;
let (dw, sw) = if total > 0.0 {
(dense_weight / total, sparse_weight / total)
} else {
(0.5, 0.5)
};
let max_dense = points
.iter()
.map(|p| p.score as f64)
.fold(0.0_f64, f64::max)
.max(1e-9);
let mut scored: Vec<(f32, VectorPoint)> = points
.into_iter()
.map(|p| {
let text_score = p
.payload
.as_ref()
.map(|pl| lexical_score(&payload_to_text(pl), &query_tokens))
.unwrap_or(0.0);
let norm_dense = p.score as f64 / max_dense;
let combined = (dw * norm_dense + sw * text_score) as f32;
(combined, p)
})
.collect();
scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
scored
.into_iter()
.take(limit)
.map(|(combined_score, mut p)| {
p.score = combined_score;
p
})
.collect()
}
fn payload_to_text(payload: &prost_types::Struct) -> String {
payload
.fields
.values()
.map(|v| match &v.kind {
Some(prost_types::value::Kind::StringValue(s)) => s.to_lowercase(),
Some(prost_types::value::Kind::NumberValue(n)) => n.to_string(),
_ => String::new(),
})
.filter(|s| !s.is_empty())
.collect::<Vec<_>>()
.join(" ")
}
fn lexical_score(text: &str, tokens: &[String]) -> f64 {
if text.is_empty() || tokens.is_empty() {
return 0.0;
}
let matched = tokens.iter().filter(|t| text.contains(t.as_str())).count();
matched as f64 / tokens.len() as f64
}
pub(crate) struct QdrantExecutor(pub(crate) QdrantHttpClient);
impl crate::runtime::backend_context::BackendContextEnforcer for QdrantExecutor {
fn backend_label(&self) -> &str {
"qdrant"
}
fn enforce(
&self,
ctx: &crate::runtime::backend_context::AppliedContext,
) -> crate::runtime::backend_context::ContextEffect {
crate::runtime::backend_context::enforce_with_mechanism(
ctx,
"_tenant_id / _project_id payload stamps; AND'd into must-filters",
)
}
}
impl BackendHealth for QdrantExecutor {
async fn ping(&self) -> Result<(), String> {
let _ = self.0.collection_exists("__ping__").await;
Ok(())
}
}
impl QueryExecutor for QdrantExecutor {
async fn query(&self, _request_json: &str) -> Result<String, tonic::Status> {
Err(capability_status(
"qdrant",
"query",
"generic_query",
"qdrant does not support generic query; use search",
))
}
}
impl SearchExecutor for QdrantExecutor {
async fn search(&self, request_json: &str) -> Result<String, tonic::Status> {
let spec: JsonValue =
serde_json::from_str(request_json).map_err(invalid_qdrant_request_json_status)?;
let collection = json_required_str(&spec, "collection")?;
let vector = json_required_f32_vec(&spec, "vector")?;
let filter = spec.get("filter").cloned().unwrap_or(JsonValue::Null);
let limit = json_i32(&spec, "limit").unwrap_or(QDRANT_DEFAULT_SEARCH_LIMIT);
let with_payload = json_bool(&spec, "with_payload").unwrap_or(true);
let result = if let Some(text_query) = spec
.get("text_query")
.and_then(JsonValue::as_str)
.filter(|value| !value.trim().is_empty())
{
let fusion_weights = spec
.get("fusion_weights")
.and_then(JsonValue::as_array)
.map(|values| {
values
.iter()
.filter_map(|value| value.as_f64().map(|number| number as f32))
.collect::<Vec<_>>()
})
.unwrap_or_default();
let request = VectorHybridSearchRequest {
context: None,
collection: collection.to_string(),
vector,
text_query: text_query.to_string(),
filter: json_to_struct(&filter),
limit,
fusion_weights,
with_payload,
with_vector: spec
.get("with_vector")
.and_then(JsonValue::as_bool)
.unwrap_or(false),
vector_name: spec
.get("vector_name")
.and_then(JsonValue::as_str)
.unwrap_or_default()
.to_string(),
fusion_strategy: spec
.get("fusion_strategy")
.and_then(JsonValue::as_i64)
.unwrap_or_default() as i32,
prefetch_limit: spec
.get("prefetch_limit")
.and_then(JsonValue::as_i64)
.unwrap_or_default() as i32,
quantization_rescore: spec
.get("quantization_rescore")
.and_then(JsonValue::as_bool)
.unwrap_or(false),
};
self.0.hybrid_search(&request, filter).await?
} else {
let request = VectorSearchRequest {
context: None,
collection: collection.to_string(),
vector,
filter: json_to_struct(&filter),
limit,
score_threshold: spec
.get("score_threshold")
.and_then(JsonValue::as_f64)
.unwrap_or_default() as f32,
with_payload,
with_vector: spec
.get("with_vector")
.and_then(JsonValue::as_bool)
.unwrap_or(false),
vector_name: spec
.get("vector_name")
.and_then(JsonValue::as_str)
.unwrap_or_default()
.to_string(),
quantization_rescore: spec
.get("quantization_rescore")
.and_then(JsonValue::as_bool)
.unwrap_or(false),
};
self.0.search(&request, filter).await?
};
let points = result
.points
.iter()
.map(|point| {
json!({
"id": point.id,
"score": point.score,
"payload": point.payload.as_ref().map(struct_to_json).unwrap_or(JsonValue::Null)
})
})
.collect::<Vec<_>>();
serde_json::to_string(&points)
.map_err(|e| qdrant_internal_status("search_result_encode", e.to_string()))
}
}
impl MutationExecutor for QdrantExecutor {
async fn mutate(&self, request_json: &str) -> Result<String, tonic::Status> {
let mut spec: JsonValue =
serde_json::from_str(request_json).map_err(invalid_qdrant_request_json_status)?;
let operation = spec
.get("operation")
.and_then(JsonValue::as_str)
.unwrap_or("upsert")
.to_string();
match operation.as_str() {
"upsert" | "upsert_points" => {
let collection = json_required_str(&spec, "collection")?.to_string();
let idempotency_key = spec
.get("idempotency_key")
.and_then(JsonValue::as_str)
.unwrap_or_default()
.to_string();
let point_specs = spec
.get_mut("points")
.and_then(JsonValue::as_array_mut)
.map(std::mem::take)
.ok_or_else(|| {
qdrant_required_field_status("points", "points must be an array")
})?;
let mut points = Vec::with_capacity(point_specs.len());
for mut point in point_specs {
points.push(VectorPointMutation {
id: json_required_str(&point, "id")?.to_string(),
vector: json_required_f32_vec(&point, "vector")?,
payload: point
.get_mut("payload")
.map(JsonValue::take)
.and_then(json_into_struct),
vector_name: point
.get("vector_name")
.and_then(JsonValue::as_str)
.unwrap_or_default()
.to_string(),
});
}
let request = VectorUpsertRequest {
context: None,
collection: collection.clone(),
points,
idempotency_key,
};
self.0.upsert(&request).await?;
Ok(json!({
"resource_uri": format!("vector://{collection}"),
"affected_rows": request.points.len()
})
.to_string())
}
"delete" | "delete_points" => {
let collection = json_required_str(&spec, "collection")?;
if let Some(filter) = spec.get("filter").cloned() {
self.0.delete_by_filter(collection, filter).await?;
return Ok(json!({
"resource_uri": format!("vector://{collection}"),
"affected_rows": 0,
"matched_by_filter": true
})
.to_string());
}
let point_ids = spec
.get("point_ids")
.or_else(|| spec.get("ids"))
.and_then(JsonValue::as_array)
.ok_or_else(|| {
qdrant_required_field_status(
"point_ids",
"point_ids must be an array when filter is absent",
)
})?
.iter()
.map(json_scalar_to_string)
.collect::<Vec<_>>();
self.0.delete_points(collection, &point_ids).await?;
Ok(json!({
"resource_uri": format!("vector://{collection}"),
"affected_rows": point_ids.len()
})
.to_string())
}
"set_payload" | "patch_payload" | "upsert_payload" => {
let collection = json_required_str(&spec, "collection")?;
let payload = spec.get("payload").cloned().ok_or_else(|| {
qdrant_required_field_status("payload", "payload is required")
})?;
let point_ids = spec
.get("point_ids")
.or_else(|| spec.get("ids"))
.and_then(|ids| {
ids.as_array().map(|values| {
values.iter().map(json_scalar_to_string).collect::<Vec<_>>()
})
});
let filter = spec.get("filter").cloned();
if point_ids.is_none() && filter.is_none() {
return Err(qdrant_required_field_status(
"point_ids",
"set_payload requires point_ids or filter",
));
}
self.0
.set_payload(collection, payload, point_ids, filter)
.await?;
Ok(json!({
"resource_uri": format!("vector://{collection}"),
"affected_rows": 0
})
.to_string())
}
other => Err(unsupported_qdrant_operation_status(other)),
}
}
}
impl ObjectExecutor for QdrantExecutor {
async fn get_object(&self, _request_json: &str) -> Result<Vec<u8>, tonic::Status> {
Err(capability_status(
"qdrant",
"get_object",
"object_store",
"qdrant is not an object store",
))
}
async fn put_object(
&self,
_request_json: &str,
_bytes: Vec<u8>,
) -> Result<String, tonic::Status> {
Err(capability_status(
"qdrant",
"put_object",
"object_store",
"qdrant is not an object store",
))
}
}
impl ResourceAdminExecutor for QdrantExecutor {
async fn ensure_resource(
&self,
resource_name: &str,
spec_json: &str,
) -> Result<(), tonic::Status> {
let spec: JsonValue =
serde_json::from_str(spec_json).map_err(invalid_qdrant_ensure_resource_spec_status)?;
let dimension = spec
.get("dimension")
.or_else(|| spec.get("vector_size"))
.or_else(|| spec.get("size"))
.or_else(|| spec.get("vectors").and_then(|vectors| vectors.get("size")))
.and_then(JsonValue::as_i64)
.unwrap_or(4)
.max(1)
.to_string();
let distance = spec
.get("distance")
.or_else(|| {
spec.get("vectors")
.and_then(|vectors| vectors.get("distance"))
})
.and_then(JsonValue::as_str)
.unwrap_or("cosine")
.to_string();
let store = ManifestStore {
store_kind: "vector".to_string(),
backend: "qdrant".to_string(),
logical_name: resource_name.to_string(),
resource_name: resource_name.to_string(),
options: vec![
crate::generation::ManifestStoreOption {
key: "dimension".to_string(),
value: dimension,
},
crate::generation::ManifestStoreOption {
key: "distance".to_string(),
value: distance,
},
],
..ManifestStore::default()
};
self.0.ensure_collection(&store).await
}
async fn drop_resource(&self, resource_name: &str) -> Result<(), tonic::Status> {
let url = format!("{}/collections/{}", self.0.base_url, resource_name);
let resp = self
.0
.http
.delete(url)
.send()
.await
.map_err(|e| backend_transport_status("qdrant", "delete", e))?;
if resp.status().is_success() || resp.status() == reqwest::StatusCode::NOT_FOUND {
Ok(())
} else {
Err(qdrant_internal_status(
"drop_resource",
format!("qdrant drop_resource status: {}", resp.status()),
))
}
}
async fn list_resources(&self) -> Result<Vec<String>, tonic::Status> {
let url = format!("{}/collections", self.0.base_url);
let resp = self
.0
.http
.get(url)
.send()
.await
.map_err(|e| backend_transport_status("qdrant", "list", e))?;
let body: JsonValue = resp
.json()
.await
.map_err(|e| backend_transport_status("qdrant", "list parse", e))?;
let names = body
.get("result")
.and_then(|r| r.get("collections"))
.and_then(|c| c.as_array())
.map(|arr| {
arr.iter()
.filter_map(|item| item.get("name").and_then(|n| n.as_str()))
.map(str::to_owned)
.collect()
})
.unwrap_or_default();
Ok(names)
}
}
impl BackendExecutor for QdrantExecutor {
async fn transaction(&self, _request_json: &str) -> Result<String, tonic::Status> {
Err(capability_status(
"qdrant",
"transaction",
"transactions",
"qdrant does not support transactions",
))
}
async fn probe(&self) -> Result<BackendProbe, tonic::Status> {
Ok(build_probe(
"qdrant",
<Self as BackendHealth>::ping(self).await,
))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::proto::{ErrorDetail, ErrorKind};
use crate::runtime::executor_utils::ERROR_DETAIL_METADATA_KEY;
fn exec() -> QdrantExecutor {
QdrantExecutor(QdrantHttpClient {
base_url: "http://localhost:6333".to_string(),
api_key: None,
http: reqwest::Client::new(),
})
}
fn decode_detail(status: &tonic::Status) -> ErrorDetail {
let raw = status
.metadata()
.get_bin(ERROR_DETAIL_METADATA_KEY)
.expect("typed detail trailer is present");
crate::runtime::executor_utils::decode_error_detail_from_raw(&raw)
}
fn assert_single_field(status: &tonic::Status, field: &str) {
let detail = decode_detail(status);
assert_eq!(detail.kind, ErrorKind::Validation as i32);
assert_eq!(detail.field_violations.len(), 1);
assert_eq!(detail.field_violations[0].field, field);
}
fn assert_schema_detail(
status: &tonic::Status,
backend: &str,
operation: &str,
schema_code: &str,
message: &str,
) {
assert_eq!(status.code(), tonic::Code::NotFound);
assert_eq!(status.message(), message);
let detail = decode_detail(status);
assert_eq!(detail.kind, ErrorKind::Schema as i32);
assert_eq!(detail.backend, backend);
assert_eq!(detail.operation, operation);
assert_eq!(detail.capability_required, schema_code);
assert!(!detail.retryable);
assert_eq!(detail.retry_after_ms, 0);
assert!(detail.field_violations.is_empty());
}
fn assert_internal_detail(status: &tonic::Status, operation: &str, message: &str) {
assert_eq!(status.code(), tonic::Code::Internal);
assert_eq!(status.message(), message);
let detail = decode_detail(status);
assert_eq!(detail.kind, ErrorKind::Internal as i32);
assert_eq!(detail.backend, "qdrant");
assert_eq!(detail.operation, operation);
assert!(!detail.retryable);
}
#[test]
fn qdrant_internal_status_carries_typed_detail() {
let status = qdrant_internal_status("drop_resource", "qdrant drop_resource status: 500");
assert_internal_detail(&status, "drop_resource", "qdrant drop_resource status: 500");
}
#[test]
fn qdrant_collection_not_found_carries_schema_detail() {
assert_schema_detail(
&qdrant_collection_not_found_status("items"),
"qdrant",
"collection_check",
"qdrant_collection_not_found",
"Qdrant collection items is missing",
);
}
#[test]
fn qdrant_collection_validation_carries_field_violations() {
for (name, message) in [
("", "Qdrant collection name must be 1–255 characters"),
(
"bad/name",
"Qdrant collection name may only contain ASCII letters, digits, hyphens, and underscores",
),
(
"-bad",
"Qdrant collection name may not start with '.' or '-'",
),
] {
let err = validate_collection_name(name).unwrap_err();
assert_eq!(err.message(), message);
assert_single_field(&err, "collection");
}
}
#[test]
fn qdrant_query_response_reads_nested_points_and_named_vectors() {
let points = query_result_points(json!({
"result": {
"points": [{
"id": "point-1",
"score": 0.75,
"payload": {"_tenant_id": "tenant-1"},
"vector": {"body": [0.1, 0.2, 0.3]}
}]
}
}));
assert_eq!(points.len(), 1);
assert_eq!(points[0].vector_name, "body");
assert_eq!(points[0].vector, vec![0.1, 0.2, 0.3]);
}
#[tokio::test]
async fn qdrant_search_validation_carries_field_violations() {
let e = exec();
let invalid_json = SearchExecutor::search(&e, "not json").await.unwrap_err();
assert_eq!(invalid_json.code(), tonic::Code::InvalidArgument);
assert!(invalid_json.message().starts_with("invalid request json:"));
assert_single_field(&invalid_json, "request_json");
let missing_collection = SearchExecutor::search(&e, r#"{"vector":[1,2,3]}"#)
.await
.unwrap_err();
assert_eq!(missing_collection.message(), "collection is required");
assert_single_field(&missing_collection, "collection");
}
#[tokio::test]
async fn qdrant_mutation_validation_carries_field_violations() {
let e = exec();
let invalid_json = MutationExecutor::mutate(&e, "not json").await.unwrap_err();
assert_eq!(invalid_json.code(), tonic::Code::InvalidArgument);
assert!(invalid_json.message().starts_with("invalid request json:"));
assert_single_field(&invalid_json, "request_json");
let missing_points =
MutationExecutor::mutate(&e, r#"{"operation":"upsert","collection":"items"}"#)
.await
.unwrap_err();
assert_eq!(missing_points.message(), "points must be an array");
assert_single_field(&missing_points, "points");
let missing_point_ids =
MutationExecutor::mutate(&e, r#"{"operation":"delete","collection":"items"}"#)
.await
.unwrap_err();
assert_eq!(
missing_point_ids.message(),
"point_ids must be an array when filter is absent"
);
assert_single_field(&missing_point_ids, "point_ids");
let missing_payload =
MutationExecutor::mutate(&e, r#"{"operation":"set_payload","collection":"items"}"#)
.await
.unwrap_err();
assert_eq!(missing_payload.message(), "payload is required");
assert_single_field(&missing_payload, "payload");
let missing_selector = MutationExecutor::mutate(
&e,
r#"{"operation":"set_payload","collection":"items","payload":{"a":1}}"#,
)
.await
.unwrap_err();
assert_eq!(
missing_selector.message(),
"set_payload requires point_ids or filter"
);
assert_single_field(&missing_selector, "point_ids");
let unsupported = MutationExecutor::mutate(&e, r#"{"operation":"bogus"}"#)
.await
.unwrap_err();
assert_eq!(
unsupported.message(),
"unsupported Qdrant mutation operation 'bogus'"
);
assert_single_field(&unsupported, "operation");
}
#[tokio::test]
async fn qdrant_resource_spec_validation_carries_field_violation() {
let e = exec();
let err = ResourceAdminExecutor::ensure_resource(&e, "items", "{")
.await
.unwrap_err();
assert!(
err.message()
.starts_with("invalid qdrant ensure_resource spec:")
);
assert_single_field(&err, "spec_json");
}
#[tokio::test]
async fn qdrant_executor_rejects_unsupported_and_malformed() {
let e = exec();
assert!(QueryExecutor::query(&e, "{}").await.is_err());
assert!(ObjectExecutor::get_object(&e, "{}").await.is_err());
assert!(BackendExecutor::transaction(&e, "{}").await.is_err());
assert!(SearchExecutor::search(&e, "not json").await.is_err());
assert!(SearchExecutor::search(&e, "{}").await.is_err()); assert!(
MutationExecutor::mutate(&e, r#"{"operation":"bogus"}"#)
.await
.is_err()
);
}
}