RUST-1667 Add search index management helpers (#989)

Author: abr-egn

src/coll.rs

@@ -135,7 +135,7 @@ impl<T> Clone for Collection<T> {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 struct CollectionInner {
     client: Client,
     db: Database,
@@ -183,6 +183,16 @@ impl<T> Collection<T> {
         }
     }
 
+    pub(crate) fn clone_unconcerned(&self) -> Self {
+        let mut new_inner = CollectionInner::clone(&self.inner);
+        new_inner.write_concern = None;
+        new_inner.read_concern = None;
+        Self {
+            inner: Arc::new(new_inner),
+            _phantom: Default::default(),
+        }
+    }
+
     /// Get the `Client` that this collection descended from.
     pub fn client(&self) -> &Client {
         &self.inner.client

src/lib.rs

@@ -337,6 +337,7 @@ mod index;
 mod operation;
 pub mod results;
 pub(crate) mod runtime;
+mod search_index;
 mod sdam;
 mod selection_criteria;
 mod serde_util;
@@ -362,7 +363,7 @@ pub use crate::{
     gridfs::{GridFsBucket, GridFsDownloadStream, GridFsUploadStream},
 };
 
-pub use {client::session::ClusterTime, coll::Namespace, index::IndexModel, sdam::public::*};
+pub use {client::session::ClusterTime, coll::Namespace, index::IndexModel, sdam::public::*, search_index::SearchIndexModel};
 
 #[cfg(all(feature = "tokio-runtime", feature = "sync",))]
 compile_error!(

src/operation.rs

@@ -20,6 +20,7 @@ mod list_indexes;
 mod raw_output;
 mod run_command;
 mod run_cursor_command;
+mod search_index;
 mod update;
 
 #[cfg(test)]
@@ -73,6 +74,7 @@ pub(crate) use list_indexes::ListIndexes;
 pub(crate) use raw_output::RawOutput;
 pub(crate) use run_command::RunCommand;
 pub(crate) use run_cursor_command::RunCursorCommand;
+pub(crate) use search_index::{CreateSearchIndexes, DropSearchIndex, UpdateSearchIndex};
 pub(crate) use update::{Update, UpdateOrReplace};
 
 const SERVER_4_2_0_WIRE_VERSION: i32 = 8;

src/operation/search_index.rs

@@ -0,0 +1,179 @@
+use bson::{doc, Document};
+use serde::Deserialize;
+
+use crate::{cmap::Command, error::Result, Namespace, SearchIndexModel};
+
+use super::OperationWithDefaults;
+
+#[derive(Debug)]
+pub(crate) struct CreateSearchIndexes {
+    ns: Namespace,
+    indexes: Vec<SearchIndexModel>,
+}
+
+impl CreateSearchIndexes {
+    pub(crate) fn new(ns: Namespace, indexes: Vec<SearchIndexModel>) -> Self {
+        Self { ns, indexes }
+    }
+}
+
+impl OperationWithDefaults for CreateSearchIndexes {
+    type O = Vec<String>;
+    type Command = Document;
+    const NAME: &'static str = "createSearchIndexes";
+
+    fn build(&mut self, _description: &crate::cmap::StreamDescription) -> Result<Command> {
+        Ok(Command::new(
+            Self::NAME.to_string(),
+            self.ns.db.clone(),
+            doc! {
+                Self::NAME: self.ns.coll.clone(),
+                "indexes": bson::to_bson(&self.indexes)?,
+            },
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        response: crate::cmap::RawCommandResponse,
+        _description: &crate::cmap::StreamDescription,
+    ) -> Result<Self::O> {
+        #[derive(Debug, Deserialize)]
+        #[serde(rename_all = "camelCase")]
+        struct Response {
+            indexes_created: Vec<CreatedIndex>,
+        }
+
+        #[derive(Debug, Deserialize)]
+        struct CreatedIndex {
+            #[allow(unused)]
+            id: String,
+            name: String,
+        }
+
+        let response: Response = response.body()?;
+        Ok(response
+            .indexes_created
+            .into_iter()
+            .map(|ci| ci.name)
+            .collect())
+    }
+
+    fn supports_sessions(&self) -> bool {
+        false
+    }
+
+    fn supports_read_concern(&self, _description: &crate::cmap::StreamDescription) -> bool {
+        false
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct UpdateSearchIndex {
+    ns: Namespace,
+    name: String,
+    definition: Document,
+}
+
+impl UpdateSearchIndex {
+    pub(crate) fn new(ns: Namespace, name: String, definition: Document) -> Self {
+        Self {
+            ns,
+            name,
+            definition,
+        }
+    }
+}
+
+impl OperationWithDefaults for UpdateSearchIndex {
+    type O = ();
+    type Command = Document;
+    const NAME: &'static str = "updateSearchIndex";
+
+    fn build(
+        &mut self,
+        _description: &crate::cmap::StreamDescription,
+    ) -> crate::error::Result<crate::cmap::Command<Self::Command>> {
+        Ok(Command::new(
+            Self::NAME.to_string(),
+            self.ns.db.clone(),
+            doc! {
+                Self::NAME: self.ns.coll.clone(),
+                "name": &self.name,
+                "definition": &self.definition,
+            },
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        response: crate::cmap::RawCommandResponse,
+        _description: &crate::cmap::StreamDescription,
+    ) -> crate::error::Result<Self::O> {
+        response.body()
+    }
+
+    fn supports_sessions(&self) -> bool {
+        false
+    }
+
+    fn supports_read_concern(&self, _description: &crate::cmap::StreamDescription) -> bool {
+        false
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct DropSearchIndex {
+    ns: Namespace,
+    name: String,
+}
+
+impl DropSearchIndex {
+    pub(crate) fn new(ns: Namespace, name: String) -> Self {
+        Self { ns, name }
+    }
+}
+
+impl OperationWithDefaults for DropSearchIndex {
+    type O = ();
+    type Command = Document;
+    const NAME: &'static str = "dropSearchIndex";
+
+    fn build(
+        &mut self,
+        _description: &crate::cmap::StreamDescription,
+    ) -> Result<Command<Self::Command>> {
+        Ok(Command::new(
+            Self::NAME.to_string(),
+            self.ns.db.clone(),
+            doc! {
+                Self::NAME: self.ns.coll.clone(),
+                "name": &self.name,
+            },
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        response: crate::cmap::RawCommandResponse,
+        _description: &crate::cmap::StreamDescription,
+    ) -> Result<Self::O> {
+        response.body()
+    }
+
+    fn handle_error(&self, error: crate::error::Error) -> Result<Self::O> {
+        if error.is_ns_not_found() {
+            Ok(())
+        } else {
+            Err(error)
+        }
+    }
+
+    fn supports_sessions(&self) -> bool {
+        false
+    }
+
+    fn supports_read_concern(&self, _description: &crate::cmap::StreamDescription) -> bool {
+        false
+    }
+}

src/options.rs

@@ -25,6 +25,7 @@ pub use crate::{
     db::options::*,
     gridfs::options::*,
     index::options::*,
+    search_index::options::*,
     selection_criteria::*,
 };
 

src/search_index.rs

@@ -0,0 +1,135 @@
+use self::options::*;
+use crate::{
+    bson::Document,
+    coll::options::AggregateOptions,
+    error::{Error, Result},
+    operation::{CreateSearchIndexes, DropSearchIndex, UpdateSearchIndex},
+    Collection,
+    Cursor,
+};
+
+use bson::doc;
+use serde::{Deserialize, Serialize};
+use typed_builder::TypedBuilder;
+
+impl<T> Collection<T> {
+    /// Convenience method for creating a single search index.
+    pub async fn create_search_index(
+        &self,
+        model: SearchIndexModel,
+        options: impl Into<Option<CreateSearchIndexOptions>>,
+    ) -> Result<String> {
+        let mut names = self.create_search_indexes(Some(model), options).await?;
+        match names.len() {
+            1 => Ok(names.pop().unwrap()),
+            n => Err(Error::internal(format!("expected 1 index name, got {}", n))),
+        }
+    }
+
+    /// Creates multiple search indexes on the collection.
+    pub async fn create_search_indexes(
+        &self,
+        models: impl IntoIterator<Item = SearchIndexModel>,
+        _options: impl Into<Option<CreateSearchIndexOptions>>,
+    ) -> Result<Vec<String>> {
+        let op = CreateSearchIndexes::new(self.namespace(), models.into_iter().collect());
+        self.client().execute_operation(op, None).await
+    }
+
+    /// Updates the search index with the given name to use the provided definition.
+    pub async fn update_search_index(
+        &self,
+        name: impl AsRef<str>,
+        definition: Document,
+        _options: impl Into<Option<UpdateSearchIndexOptions>>,
+    ) -> Result<()> {
+        let op = UpdateSearchIndex::new(
+            self.namespace(),
+            name.as_ref().to_string(),
+            definition.clone(),
+        );
+        self.client().execute_operation(op, None).await
+    }
+
+    /// Drops the search index with the given name.
+    pub async fn drop_search_index(
+        &self,
+        name: impl AsRef<str>,
+        _options: impl Into<Option<DropSearchIndexOptions>>,
+    ) -> Result<()> {
+        let op = DropSearchIndex::new(self.namespace(), name.as_ref().to_string());
+        self.client().execute_operation(op, None).await
+    }
+
+    /// Gets index information for one or more search indexes in the collection.
+    ///
+    /// If name is not specified, information for all indexes on the specified collection will be
+    /// returned.
+    pub async fn list_search_indexes(
+        &self,
+        name: impl Into<Option<&str>>,
+        aggregation_options: impl Into<Option<AggregateOptions>>,
+        _list_index_options: impl Into<Option<ListSearchIndexOptions>>,
+    ) -> Result<Cursor<Document>> {
+        let mut inner = doc! {};
+        if let Some(name) = name.into() {
+            inner.insert("name", name.to_string());
+        }
+        self.clone_unconcerned()
+            .aggregate(
+                vec![doc! {
+                    "$listSearchIndexes": inner,
+                }],
+                aggregation_options,
+            )
+            .await
+    }
+}
+
+/// Specifies the options for a search index.
+#[derive(Debug, Clone, Default, TypedBuilder, Serialize, Deserialize)]
+#[builder(field_defaults(default, setter(into)))]
+#[non_exhaustive]
+pub struct SearchIndexModel {
+    /// The definition for this index.
+    pub definition: Document,
+
+    /// The name for this index, if present.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub name: Option<String>,
+}
+
+pub mod options {
+    #[cfg(docsrs)]
+    use crate::Collection;
+    use serde::Deserialize;
+    use typed_builder::TypedBuilder;
+
+    /// Options for [Collection::create_search_index].  Present to allow additional options to be
+    /// added in the future as a non-breaking change.
+    #[derive(Clone, Debug, Default, TypedBuilder, Deserialize)]
+    #[builder(field_defaults(default, setter(into)))]
+    #[non_exhaustive]
+    pub struct CreateSearchIndexOptions {}
+
+    /// Options for [Collection::update_search_index].  Present to allow additional options to be
+    /// added in the future as a non-breaking change.
+    #[derive(Clone, Debug, Default, TypedBuilder, Deserialize)]
+    #[builder(field_defaults(default, setter(into)))]
+    #[non_exhaustive]
+    pub struct UpdateSearchIndexOptions {}
+
+    /// Options for [Collection::list_search_indexes].  Present to allow additional options to be
+    /// added in the future as a non-breaking change.
+    #[derive(Clone, Debug, Default, TypedBuilder, Deserialize)]
+    #[builder(field_defaults(default, setter(into)))]
+    #[non_exhaustive]
+    pub struct ListSearchIndexOptions {}
+
+    /// Options for [Collection::drop_search_index].  Present to allow additional options to be
+    /// added in the future as a non-breaking change.
+    #[derive(Clone, Debug, Default, TypedBuilder, Deserialize)]
+    #[builder(field_defaults(default, setter(into)))]
+    #[non_exhaustive]
+    pub struct DropSearchIndexOptions {}
+}

src/test/spec.rs

@@ -10,6 +10,7 @@ mod crud;
 mod crud_v1;
 mod faas;
 mod gridfs;
+mod index_management;
 #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
 mod initial_dns_seedlist_discovery;
 mod load_balancers;

src/test/spec/index_management.rs

@@ -0,0 +1,7 @@
+use crate::test::spec::unified_runner::run_unified_tests;
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test)]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+async fn run() {
+    run_unified_tests(&["index-management"]).await;
+}