fix(mongodb): use dedicated count for legacy collections
This commit is contained in:
1 parent
03bc87c77a
commit
a99a34b4ab
28 files changed
+916
-19
No files matched your search
@@ -252,6 +252,7 @@ pub enum MongoAgentMethod {
|
||||
ListCollections,
|
||||
FindDocuments,
|
||||
FindDocumentsExtendedJson,
|
||||
CountDocuments,
|
||||
ServerVersion,
|
||||
CreateIndex,
|
||||
DropIndexes,
|
||||
@@ -264,11 +265,12 @@ pub enum MongoAgentMethod {
|
||||
}
|
||||
|
||||
impl MongoAgentMethod {
|
||||
pub const ALL: [Self; 13] = [
|
||||
pub const ALL: [Self; 14] = [
|
||||
Self::ListDatabases,
|
||||
Self::ListCollections,
|
||||
Self::FindDocuments,
|
||||
Self::FindDocumentsExtendedJson,
|
||||
Self::CountDocuments,
|
||||
Self::ServerVersion,
|
||||
Self::CreateIndex,
|
||||
Self::DropIndexes,
|
||||
@@ -286,6 +288,7 @@ impl MongoAgentMethod {
|
||||
Self::ListCollections => "list_collections",
|
||||
Self::FindDocuments => "find_documents",
|
||||
Self::FindDocumentsExtendedJson => "find_documents_extended_json",
|
||||
Self::CountDocuments => "count_documents",
|
||||
Self::ServerVersion => "server_version",
|
||||
Self::CreateIndex => "create_index",
|
||||
Self::DropIndexes => "drop_indexes",
|
||||
@@ -999,6 +1002,13 @@ impl AgentDriverClient {
|
||||
self.call_mongo_method(MongoAgentMethod::FindDocumentsExtendedJson, params).await
|
||||
}
|
||||
|
||||
pub async fn mongo_count_documents<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
params: Value,
|
||||
) -> Result<T, String> {
|
||||
self.call_mongo_method(MongoAgentMethod::CountDocuments, params).await
|
||||
}
|
||||
|
||||
pub async fn mongo_server_version<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
database: &str,
|
||||
@@ -1727,6 +1737,7 @@ mod tests {
|
||||
assert_eq!(MongoAgentMethod::ListCollections.as_str(), "list_collections");
|
||||
assert_eq!(MongoAgentMethod::FindDocuments.as_str(), "find_documents");
|
||||
assert_eq!(MongoAgentMethod::FindDocumentsExtendedJson.as_str(), "find_documents_extended_json");
|
||||
assert_eq!(MongoAgentMethod::CountDocuments.as_str(), "count_documents");
|
||||
assert_eq!(MongoAgentMethod::ServerVersion.as_str(), "server_version");
|
||||
assert_eq!(MongoAgentMethod::CreateIndex.as_str(), "create_index");
|
||||
assert_eq!(MongoAgentMethod::DropIndexes.as_str(), "drop_indexes");
|
||||
|
||||
@@ -592,6 +592,31 @@ pub async fn find_documents(
|
||||
Ok(MongoDocumentResult { documents, total })
|
||||
}
|
||||
|
||||
pub async fn count_documents(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter: Option<&str>,
|
||||
accurate: bool,
|
||||
) -> Result<u64, String> {
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
|
||||
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! {},
|
||||
};
|
||||
|
||||
if !accurate && filter_doc.is_empty() {
|
||||
// Legacy count() permits the metadata-backed fast path; countDocuments() must scan accurately.
|
||||
col.estimated_document_count().await.map_err(|e| e.to_string())
|
||||
} else {
|
||||
col.count_documents(filter_doc).await.map_err(|e| e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
/// Find MongoDB documents as relaxed Extended JSON for MongoDB transfer paths.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn find_documents_extended_json(
|
||||
|
||||
@@ -121,6 +121,50 @@ pub async fn mongo_find_documents_core(
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn mongo_count_documents_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter: Option<&str>,
|
||||
mode: Option<&str>,
|
||||
) -> Result<u64, String> {
|
||||
let accurate = mode != Some("legacy");
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::count_documents(client, database, collection, filter, accurate).await
|
||||
}
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
let params = serde_json::json!({
|
||||
"database": database,
|
||||
"collection": collection,
|
||||
"filter": filter,
|
||||
"accurate": accurate,
|
||||
});
|
||||
match client.mongo_count_documents(params.clone()).await {
|
||||
Ok(total) => Ok(total),
|
||||
Err(error) if is_unknown_agent_method_error(&error, "count_documents") => {
|
||||
let result: MongoDocumentResult = client
|
||||
.mongo_find_documents(serde_json::json!({
|
||||
"database": database,
|
||||
"collection": collection,
|
||||
"skip": 0,
|
||||
"limit": 1,
|
||||
"filter": filter,
|
||||
}))
|
||||
.await?;
|
||||
Ok(result.total)
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Read MongoDB documents as relaxed Extended JSON for MongoDB transfer paths.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn mongo_find_documents_extended_json_core(
|
||||
|
||||
Reference in new issue
Block a user