Upgrade mongodb library to 3.1.0

Cargo.toml

@@ -25,22 +25,12 @@ indexmap = { version = "2", features = [
   "serde",
 ] } # should match the version that ndc-models uses
 itertools = "^0.12.1"
-mongodb = { version = "2.8", features = ["tracing-unstable"] }
+mongodb = { version = "^3.1.0", features = ["tracing-unstable"] }
 schemars = "^0.8.12"
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] }
 ref-cast = "1.0.23"
 
-# Connecting to MongoDB Atlas database with time series collections fails in the
-# latest released version of the MongoDB Rust driver. A fix has been merged, but
-# it has not been released yet: https://github.com/mongodb/mongo-rust-driver/pull/1077
-#
-# We are using a branch of the driver that cherry-picks that fix onto the v2.8.2
-# release.
-[patch.crates-io.mongodb]
-git = "https://github.com/hasura/mongo-rust-driver.git"
-branch = "upstream-time-series-fix"
-
 # Set opt levels according to recommendations in insta documentation
 [profile.dev.package]
 insta.opt-level = 3

crates/cli/src/introspection/sampling.rs

@@ -28,7 +28,7 @@ pub async fn sample_schema_from_db(
 ) -> anyhow::Result<BTreeMap<std::string::String, Schema>> {
     let mut schemas = BTreeMap::new();
     let db = state.database();
-    let mut collections_cursor = db.list_collections(None, None).await?;
+    let mut collections_cursor = db.list_collections().await?;
 
     while let Some(collection_spec) = collections_cursor.try_next().await? {
         let collection_name = collection_spec.name;
@@ -60,7 +60,8 @@ async fn sample_schema_from_collection(
     let options = None;
     let mut cursor = db
         .collection::<Document>(collection_name)
-        .aggregate(vec![doc! {"$sample": { "size": sample_size }}], options)
+        .aggregate(vec![doc! {"$sample": { "size": sample_size }}])
+        .with_options(options)
         .await?;
     let mut collected_object_types = vec![];
     let is_collection_type = true;

crates/cli/src/introspection/validation_schema.rs

@@ -22,7 +22,7 @@ pub async fn get_metadata_from_validation_schema(
     state: &ConnectorState,
 ) -> Result<BTreeMap<String, Schema>, MongoAgentError> {
     let db = state.database();
-    let mut collections_cursor = db.list_collections(None, None).await?;
+    let mut collections_cursor = db.list_collections().await?;
 
     let mut schemas: Vec<WithName<String, Schema>> = vec![];
 

crates/mongodb-agent-common/src/explain.rs

@@ -41,7 +41,7 @@ pub async fn explain_query(
 
     tracing::debug!(explain_command = %serde_json::to_string(&explain_command).unwrap());
 
-    let explain_result = db.run_command(explain_command, None).await?;
+    let explain_result = db.run_command(explain_command).await?;
 
     let plan =
         serde_json::to_string_pretty(&explain_result).map_err(MongoAgentError::Serialization)?;

crates/mongodb-agent-common/src/mongodb/collection.rs

@@ -39,13 +39,12 @@ where
     where
         Options: Into<Option<AggregateOptions>> + Send + 'static;
 
-    async fn find<Filter, Options>(
+    async fn find<Options>(
         &self,
-        filter: Filter,
+        filter: Document,
         options: Options,
     ) -> Result<Self::RowCursor, Error>
     where
-        Filter: Into<Option<Document>> + Send + 'static,
         Options: Into<Option<FindOptions>> + Send + 'static;
 }
 
@@ -65,18 +64,19 @@ where
     where
         Options: Into<Option<AggregateOptions>> + Send + 'static,
     {
-        Collection::aggregate(self, pipeline, options).await
+        Collection::aggregate(self, pipeline)
+            .with_options(options)
+            .await
     }
 
-    async fn find<Filter, Options>(
+    async fn find<Options>(
         &self,
-        filter: Filter,
+        filter: Document,
         options: Options,
     ) -> Result<Self::RowCursor, Error>
     where
-        Filter: Into<Option<Document>> + Send + 'static,
         Options: Into<Option<FindOptions>> + Send + 'static,
     {
-        Collection::find(self, filter, options).await
+        Collection::find(self, filter).with_options(options).await
     }
 }

crates/mongodb-agent-common/src/mongodb/database.rs

@@ -55,7 +55,9 @@ impl DatabaseTrait for Database {
     where
         Options: Into<Option<AggregateOptions>> + Send + 'static,
     {
-        Database::aggregate(self, pipeline, options).await
+        Database::aggregate(self, pipeline)
+            .with_options(options)
+            .await
     }
 
     fn collection(&self, name: &str) -> Self::Collection {

crates/mongodb-agent-common/src/mongodb_connection.rs

@@ -9,9 +9,9 @@ const DRIVER_NAME: &str = "Hasura";
 
 pub async fn get_mongodb_client(database_uri: &str) -> Result<Client, MongoAgentError> {
     // An extra line of code to work around a DNS issue on Windows:
-    let mut options =
-        ClientOptions::parse_with_resolver_config(database_uri, ResolverConfig::cloudflare())
-            .await?;
+    let mut options = ClientOptions::parse(database_uri)
+        .resolver_config(ResolverConfig::cloudflare())
+        .await?;
 
     // Helps MongoDB to collect statistics on Hasura use
     options.driver_info = Some(DriverInfo::builder().name(DRIVER_NAME).build());

crates/mongodb-agent-common/src/procedure/mod.rs

@@ -44,9 +44,14 @@ impl<'a> Procedure<'a> {
         self,
         database: Database,
     ) -> Result<(bson::Document, Type), ProcedureError> {
-        let selection_criteria = self.selection_criteria.map(Cow::into_owned);
         let command = interpolate(self.arguments, &self.command)?;
-        let result = database.run_command(command, selection_criteria).await?;
+        let run_command = database.run_command(command);
+        let run_command = if let Some(selection_criteria) = self.selection_criteria {
+            run_command.selection_criteria(selection_criteria.into_owned())
+        } else {
+            run_command
+        };
+        let result = run_command.await?;
         Ok((result, self.result_type))
     }
 

crates/ndc-query-plan/src/lib.rs

@@ -6,10 +6,9 @@ pub mod vec_set;
 
 pub use mutation_plan::*;
 pub use plan_for_query_request::{
-    plan_for_query_request,
+    plan_for_mutation_request, plan_for_query_request,
     query_context::QueryContext,
     query_plan_error::QueryPlanError,
-    plan_for_mutation_request,
     type_annotated_field::{type_annotated_field, type_annotated_nested_field},
 };
 pub use query_plan::*;