feat(mongodb): support distinct over the legacy agent
This commit is contained in:
1 parent
927f2f10d5
commit
87d1e114c4
3 files changed
+115
-24
No files matched your search
@@ -545,8 +545,11 @@ pub async fn mongo_distinct_core(
|
||||
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
|
||||
match &pool {
|
||||
PoolKind::MongoDb(client) => mongo_driver::distinct(client, database, collection, field, filter).await,
|
||||
// The legacy agent protocol has no distinct method and no read that could stand in for it.
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support distinct".to_string()),
|
||||
PoolKind::Agent(client) => {
|
||||
let command = mongo_driver::distinct_command(collection, field, filter)?;
|
||||
let response = agent_run_command_document(client, database, command, "distinct").await?;
|
||||
mongo_driver::distinct_response_result(&response)
|
||||
}
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
+50
-1
@@ -1,4 +1,4 @@
|
||||
//! `findOneAndUpdate` / `findOneAndReplace` / `findOneAndDelete` over the MongoDB Legacy Agent,
|
||||
//! Shell commands that reach the MongoDB Legacy Agent through its generic `runCommand`,
|
||||
//! driven through the same parse → dispatch path the query tab uses.
|
||||
use dbx_core::{
|
||||
connection::{AppState, PoolKind},
|
||||
@@ -152,3 +152,52 @@ async fn find_and_modify_commands_run_over_the_legacy_agent() {
|
||||
|
||||
command(&state, id, &database, doc! { "dropDatabase": 1 }).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "opt-in: DBX_MONGO_LEGACY_DUMP_TEST_HOST (host:port, MongoDB 3.6+ without auth) and an installed MongoDB Legacy Agent; creates a temporary database"]
|
||||
async fn distinct_runs_over_the_legacy_agent() {
|
||||
let endpoint = std::env::var("DBX_MONGO_LEGACY_DUMP_TEST_HOST").expect("DBX_MONGO_LEGACY_DUMP_TEST_HOST");
|
||||
let (host, port) = endpoint.split_once(':').expect("host:port");
|
||||
let files = tempfile::tempdir().unwrap();
|
||||
let database = format!("dbx_legacy_distinct_{}", uuid::Uuid::new_v4().simple());
|
||||
let state = AppState::new(Storage::open(&files.path().join("storage.db")).await.unwrap());
|
||||
let id = "legacy-distinct-test";
|
||||
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({ "id": id, "name": "Legacy distinct test", "db_type": "mongodb", "host": host, "port": port.parse::<u16>().unwrap(), "username": "", "password": "", "database": database, "driver_profile": "mongodb-legacy" })).unwrap();
|
||||
state.configs.write().await.insert(id.into(), config);
|
||||
let key = state.get_or_create_pool(id, Some(&database)).await.unwrap();
|
||||
assert!(matches!(state.pool_handle(&key).await, Some(PoolKind::Agent(_))), "test must run over the legacy agent");
|
||||
|
||||
command(
|
||||
&state,
|
||||
id,
|
||||
&database,
|
||||
doc! { "insert": "orders", "documents": [
|
||||
{ "_id": 1, "status": "open", "tags": ["a", "b"], "meta": { "region": "eu" } },
|
||||
{ "_id": 2, "status": "closed", "tags": ["b"], "meta": { "region": "us" } },
|
||||
{ "_id": 3, "status": "open", "tags": [], "meta": { "region": "eu" } },
|
||||
] },
|
||||
)
|
||||
.await;
|
||||
|
||||
let statuses = shell(&state, id, &database, r#"db.orders.distinct("status")"#).await.unwrap();
|
||||
let mut shown: Vec<String> = statuses.rows.iter().map(|row| format!("{:?}", row[0])).collect();
|
||||
shown.sort();
|
||||
assert_eq!(shown.len(), 2, "{statuses:?}");
|
||||
assert!(shown[0].contains("closed") && shown[1].contains("open"), "{shown:?}");
|
||||
|
||||
// Array fields are flattened and a filter narrows the input, as the native helper does.
|
||||
let tags = shell(&state, id, &database, r#"db.orders.distinct("tags", {status: "open"})"#).await.unwrap();
|
||||
let mut tags: Vec<String> = tags.rows.iter().map(|row| format!("{:?}", row[0])).collect();
|
||||
tags.sort();
|
||||
assert_eq!(tags.len(), 2, "{tags:?}");
|
||||
|
||||
// Dotted paths reach nested fields.
|
||||
let regions = shell(&state, id, &database, r#"db.orders.distinct("meta.region")"#).await.unwrap();
|
||||
assert_eq!(regions.rows.len(), 2, "{regions:?}");
|
||||
|
||||
// No matches: empty result, not an error.
|
||||
let none = shell(&state, id, &database, r#"db.orders.distinct("status", {status: "missing"})"#).await.unwrap();
|
||||
assert_eq!(none.rows.len(), 0);
|
||||
|
||||
command(&state, id, &database, doc! { "dropDatabase": 1 }).await;
|
||||
}
|
||||
@@ -1689,6 +1689,48 @@ fn aggregate_options_explain(options: &Document) -> Result<bool, String> {
|
||||
/// array fields contribute their elements rather than the whole array, and the server may
|
||||
/// answer from an index with a DISTINCT_SCAN. Values are returned in the `documents` slot as
|
||||
/// bare scalars, which `mongoDocumentsToQueryResult` already renders as a single column.
|
||||
fn distinct_filter_document(filter: Option<&str>) -> Result<Document, String> {
|
||||
match filter {
|
||||
Some(f) if !f.trim().is_empty() => {
|
||||
let json: serde_json::Value = serde_json::from_str(f).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
json_filter_to_document(&json)
|
||||
}
|
||||
_ => Ok(doc! {}),
|
||||
}
|
||||
}
|
||||
|
||||
fn distinct_values_result(values: Vec<Bson>) -> MongoDocumentResult {
|
||||
let documents = values.iter().map(bson_to_json).collect::<Vec<_>>();
|
||||
let extended_documents = values.into_iter().map(|value| value.into_relaxed_extjson()).collect::<Vec<_>>();
|
||||
let total = documents.len() as u64;
|
||||
MongoDocumentResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: Some(extended_documents),
|
||||
total,
|
||||
total_is_exact: true,
|
||||
next_cursor: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The `distinct` server command behind `distinct()`, for drivers that only expose `runCommand`
|
||||
/// (the legacy agent). Same field and filter validation as [`distinct`].
|
||||
pub fn distinct_command(collection: &str, field: &str, filter: Option<&str>) -> Result<Document, String> {
|
||||
if field.trim().is_empty() {
|
||||
return Err("Distinct field name is required".to_string());
|
||||
}
|
||||
let query = distinct_filter_document(filter)?;
|
||||
Ok(doc! { "distinct": collection, "key": field, "query": query })
|
||||
}
|
||||
|
||||
/// Turns a `distinct` response (`{ values: [...] }`) into the same result shape as [`distinct`].
|
||||
pub fn distinct_response_result(response: &Document) -> Result<MongoDocumentResult, String> {
|
||||
match response.get("values") {
|
||||
Some(Bson::Array(values)) => Ok(distinct_values_result(values.clone())),
|
||||
other => Err(format!("Unexpected distinct values: {other:?}")),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn distinct(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
@@ -1699,29 +1741,10 @@ pub async fn distinct(
|
||||
if field.trim().is_empty() {
|
||||
return Err("Distinct field name is required".to_string());
|
||||
}
|
||||
|
||||
let filter_doc: Document = match filter {
|
||||
Some(f) if !f.trim().is_empty() => {
|
||||
let json: serde_json::Value = serde_json::from_str(f).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
json_filter_to_document(&json)?
|
||||
}
|
||||
_ => doc! {},
|
||||
};
|
||||
|
||||
let filter_doc = distinct_filter_document(filter)?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let values = col.distinct(field, filter_doc).await.map_err(|e| e.to_string())?;
|
||||
let documents = values.iter().map(bson_to_json).collect::<Vec<_>>();
|
||||
let extended_documents = values.into_iter().map(|value| value.into_relaxed_extjson()).collect::<Vec<_>>();
|
||||
let total = documents.len() as u64;
|
||||
|
||||
Ok(MongoDocumentResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: Some(extended_documents),
|
||||
total,
|
||||
total_is_exact: true,
|
||||
next_cursor: None,
|
||||
})
|
||||
Ok(distinct_values_result(values))
|
||||
}
|
||||
|
||||
pub async fn create_index(
|
||||
@@ -5095,6 +5118,22 @@ mod tests {
|
||||
assert!(!delete.contains_key("update"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn distinct_command_and_response_mirror_the_native_helper() {
|
||||
let command = distinct_command("users", "status", Some(r#"{"age":{"$gte":18}}"#)).unwrap();
|
||||
assert_eq!(command.get_str("distinct"), Ok("users"));
|
||||
assert_eq!(command.get_str("key"), Ok("status"));
|
||||
assert_eq!(command.get_document("query"), Ok(&doc! { "age": { "$gte": 18i64 } }));
|
||||
assert_eq!(distinct_command("users", "status", None).unwrap().get_document("query"), Ok(&doc! {}));
|
||||
assert!(distinct_command("users", " ", None).unwrap_err().contains("field name is required"));
|
||||
assert!(distinct_command("users", "status", Some("{not json")).unwrap_err().contains("Invalid filter JSON"));
|
||||
|
||||
let result = distinct_response_result(&doc! { "values": ["a", 2i64, Bson::Null], "ok": 1.0 }).unwrap();
|
||||
assert_eq!(result.total, 3);
|
||||
assert_eq!(result.documents, vec![serde_json::json!("a"), serde_json::json!(2), serde_json::Value::Null]);
|
||||
assert!(distinct_response_result(&doc! { "ok": 1.0 }).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn find_and_modify_response_maps_value_to_zero_or_one_document() {
|
||||
let matched =
|
||||
|
||||
Reference in new issue
Block a user