diff --git a/crates/dbx-core/src/query/mongo_ops.rs b/crates/dbx-core/src/query/mongo_ops.rs index 5ef467d73d..164ec07172 100644 --- a/crates/dbx-core/src/query/mongo_ops.rs +++ b/crates/dbx-core/src/query/mongo_ops.rs @@ -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()), } } diff --git a/crates/dbx-core/tests/live_mongodb_legacy_agent_find_and_modify.rs b/crates/dbx-core/tests/live_mongodb_legacy_agent_commands.rs similarity index 71% rename from crates/dbx-core/tests/live_mongodb_legacy_agent_find_and_modify.rs rename to crates/dbx-core/tests/live_mongodb_legacy_agent_commands.rs index 1b423a3d8a..0d3abfa2dd 100644 --- a/crates/dbx-core/tests/live_mongodb_legacy_agent_find_and_modify.rs +++ b/crates/dbx-core/tests/live_mongodb_legacy_agent_commands.rs @@ -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::().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 = 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 = 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; +} diff --git a/crates/dbx-drivers/src/db/mongo_driver.rs b/crates/dbx-drivers/src/db/mongo_driver.rs index ac1f05ebc0..3ba27d3d28 100644 --- a/crates/dbx-drivers/src/db/mongo_driver.rs +++ b/crates/dbx-drivers/src/db/mongo_driver.rs @@ -1689,6 +1689,48 @@ fn aggregate_options_explain(options: &Document) -> Result { /// 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 { + 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) -> MongoDocumentResult { + let documents = values.iter().map(bson_to_json).collect::>(); + let extended_documents = values.into_iter().map(|value| value.into_relaxed_extjson()).collect::>(); + 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 { + 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 { + 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::(collection); let values = col.distinct(field, filter_doc).await.map_err(|e| e.to_string())?; - let documents = values.iter().map(bson_to_json).collect::>(); - let extended_documents = values.into_iter().map(|value| value.into_relaxed_extjson()).collect::>(); - 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 =