[CHORE] Clippy for more. (chroma-core#2871)

Left some of worker untouched for concurrent PRs.

## Description of changes

*Summarize the changes made by this PR.*
 - Cleans up warnings.

## Test plan
*How are these changes tested?*

- [ ] cargo clippy && cargo test w/ tilt up.

## Documentation Changes
*Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
repository](https://github.com/chroma-core/docs)?*

Author: rescrv

rust/benchmark-datasets/src/types.rs

@@ -1,5 +1,4 @@
 use anyhow::Result;
-use chroma_types::{LogRecord, OperationRecord, UpdateMetadataValue};
 use rand::prelude::SliceRandom;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, future::Future, path::PathBuf, sync::Arc};
@@ -40,8 +39,8 @@ impl<T: RecordDataset> RecordDataset for Arc<T> {
     const NAME: &'static str = T::NAME;
     const DISPLAY_NAME: &'static str = T::DISPLAY_NAME;
 
-    fn init() -> impl Future<Output = Result<Self>> + Send {
-        async move { Ok(Arc::new(T::init().await?)) }
+    async fn init() -> Result<Self> {
+        Ok(Arc::new(T::init().await?))
     }
 
     fn create_records_stream(
@@ -202,17 +201,13 @@ mod tests {
         const NAME: &'static str = "test";
         const DISPLAY_NAME: &'static str = "Test";
 
-        fn init() -> impl Future<Output = Result<Self>> + Send {
-            async move { Ok(TestDataset { records: vec![] }) }
+        async fn init() -> Result<Self> {
+            Ok(TestDataset { records: vec![] })
         }
 
-        fn create_records_stream(
-            &self,
-        ) -> impl Future<Output = Result<impl Stream<Item = Result<Record>>>> + Send {
-            async move {
-                let records = self.records.clone();
-                Ok(futures::stream::iter(records.into_iter().map(Ok)))
-            }
+        async fn create_records_stream(&self) -> Result<impl Stream<Item = Result<Record>>> {
+            let records = self.records.clone();
+            Ok(futures::stream::iter(records.into_iter().map(Ok)))
         }
     }
 
@@ -221,7 +216,7 @@ mod tests {
     #[tokio::test]
     async fn test_frozen_query_subset() {
         let mut test_dataset = TestDataset::init().await.unwrap();
-        test_dataset.records = vec!["foo 0", "foo 1", "foo 3", "bar 0", "bar 2"]
+        test_dataset.records = ["foo 0", "foo 1", "foo 3", "bar 0", "bar 2"]
             .iter()
             .map(|&content| Record {
                 document: content.to_string(),
@@ -230,7 +225,7 @@ mod tests {
             .collect();
 
         let mut test_query_dataset = TestDataset::init().await.unwrap();
-        test_query_dataset.records = vec!["foo", "bar", "baz"]
+        test_query_dataset.records = ["foo", "bar", "baz"]
             .iter()
             .map(|&content| Record {
                 document: content.to_string(),

rust/index/benches/dataset_utilities.rs

@@ -18,17 +18,15 @@ pub async fn get_record_dataset<RecordCorpus: RecordDataset>() -> RecordCorpus {
         .with_style(style.clone());
     record_corpus_spinner.enable_steady_tick(Duration::from_millis(50));
 
-    let record_corpus = RecordCorpus::init()
+    RecordCorpus::init()
         .and_then(|r| async {
             record_corpus_spinner.set_style(finish_style.clone());
             record_corpus_spinner.set_prefix("✔︎");
             record_corpus_spinner.finish_and_clear();
             Ok(r)
         })
         .await
-        .expect("Failed to initialize record corpus");
-
-    record_corpus
+        .expect("Failed to initialize record corpus")
 }
 
 pub async fn get_record_query_dataset_pair<

rust/index/benches/full_text.rs

@@ -170,7 +170,7 @@ fn bench_querying(c: &mut Criterion) {
                     .await
                     .unwrap();
 
-                assert!(result.len() > 0, "Query result is empty");
+                assert!(!result.is_empty(), "Query result is empty");
             },
             criterion::BatchSize::SmallInput,
         )

rust/index/src/hnsw_provider.rs

@@ -416,7 +416,7 @@ impl HnswIndexProvider {
     /// Purge entries from the cache by index ID and remove temporary files from disk.
     pub async fn purge_by_id(&mut self, index_ids: &[Uuid]) {
         for index_id in index_ids {
-            match self.remove_temporary_files(&index_id).await {
+            match self.remove_temporary_files(index_id).await {
                 Ok(_) => {}
                 Err(e) => {
                     tracing::error!("Failed to remove temporary files: {}", e);

rust/index/src/utils.rs

@@ -1,6 +1,8 @@
 #[cfg(test)]
 use rand::Rng;
 
+// This is not reserved for testing.  If you need to use it outside test contexts, remove this
+// line.  It exists solely to satisfy the linter.
 #[cfg(test)]
 pub(super) fn generate_random_data(n: usize, d: usize) -> Vec<f32> {
     let mut rng: rand::prelude::ThreadRng = rand::thread_rng();

rust/worker/src/assignment/assignment_policy.rs

@@ -37,12 +37,15 @@ pub(crate) struct RendezvousHashingAssignmentPolicy {
     members: Vec<String>,
 }
 
+// This is not reserved for testing.  If you need to use it outside test contexts, remove this
+// line.  It exists solely to satisfy the linter.
+#[cfg(test)]
 impl RendezvousHashingAssignmentPolicy {
     pub(crate) fn new() -> RendezvousHashingAssignmentPolicy {
-        return RendezvousHashingAssignmentPolicy {
+        RendezvousHashingAssignmentPolicy {
             hasher: Murmur3Hasher {},
             members: vec![],
-        };
+        }
     }
 }
 
@@ -51,14 +54,12 @@ impl Configurable<AssignmentPolicyConfig> for RendezvousHashingAssignmentPolicy
     async fn try_from_config(
         config: &AssignmentPolicyConfig,
     ) -> Result<Self, Box<dyn ChromaError>> {
-        let assignment_policy_config = match config {
-            AssignmentPolicyConfig::RendezvousHashing(config) => config,
-        };
+        let AssignmentPolicyConfig::RendezvousHashing(assignment_policy_config) = config;
         let hasher = match assignment_policy_config.hasher {
             HasherType::Murmur3 => Murmur3Hasher {},
         };
         return Ok(RendezvousHashingAssignmentPolicy {
-            hasher: hasher,
+            hasher,
             members: vec![],
         });
     }
@@ -73,7 +74,7 @@ impl AssignmentPolicy for RendezvousHashingAssignmentPolicy {
     fn get_members(&self) -> Vec<String> {
         // This is not designed to be used frequently for now, nor is the number of members
         // expected to be large, so we can just clone the members
-        return self.members.clone();
+        self.members.clone()
     }
 
     fn set_members(&mut self, members: Vec<String>) {

rust/worker/src/assignment/rendezvous_hash.rs

@@ -83,7 +83,7 @@ pub(crate) fn assign<H: Hasher>(
 
     match max_member {
         Some(max_member) => return Ok(max_member.as_ref().to_string()),
-        None => return Err(AssignmentError::NoMembers),
+        None => Err(AssignmentError::NoMembers),
     }
 }
 
@@ -109,10 +109,10 @@ impl Hasher for Murmur3Hasher {
                 let member_hash_64 = member_hash as u64;
                 let key_hash_64 = key_hash as u64;
                 let merged = merge_hashes(member_hash_64, key_hash_64);
-                return Ok(merged);
+                Ok(merged)
             }
-            _ => return Err(AssignmentError::HashError),
-        };
+            _ => Err(AssignmentError::HashError),
+        }
     }
 }
 
@@ -164,8 +164,7 @@ mod tests {
         }
 
         let expected = num_keys / member_count;
-        for i in 0..member_count {
-            let count = counts[i];
+        for count in counts.iter().take(member_count).copied() {
             let diff = count - expected as i32;
             assert!(diff.abs() < tolerance);
         }

rust/worker/src/compactor/compaction_manager.rs

@@ -37,6 +37,7 @@ pub(crate) struct CompactionManager {
     // Dependencies
     log: Box<Log>,
     sysdb: Box<SysDb>,
+    #[allow(dead_code)]
     storage: Storage,
     blockfile_provider: BlockfileProvider,
     hnsw_index_provider: HnswIndexProvider,
@@ -45,6 +46,7 @@ pub(crate) struct CompactionManager {
     // Config
     compaction_manager_queue_size: usize,
     compaction_interval: Duration,
+    #[allow(dead_code)]
     min_compaction_size: usize,
     max_compaction_size: usize,
     max_partition_size: usize,
@@ -65,6 +67,7 @@ impl ChromaError for CompactionError {
 }
 
 impl CompactionManager {
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         scheduler: Scheduler,
         log: Box<Log>,
@@ -344,7 +347,7 @@ mod tests {
     #[tokio::test]
     async fn test_compaction_manager() {
         let mut log = Box::new(Log::InMemory(InMemoryLog::new()));
-        let mut in_memory_log = match *log {
+        let in_memory_log = match *log {
             Log::InMemory(ref mut log) => log,
             _ => panic!("Expected InMemoryLog"),
         };
@@ -353,9 +356,9 @@ mod tests {
 
         let collection_uuid_1 = Uuid::from_str("00000000-0000-0000-0000-000000000001").unwrap();
         in_memory_log.add_log(
-            collection_uuid_1.clone(),
-            Box::new(InternalLogRecord {
-                collection_id: collection_uuid_1.clone(),
+            collection_uuid_1,
+            InternalLogRecord {
+                collection_id: collection_uuid_1,
                 log_offset: 0,
                 log_ts: 1,
                 record: LogRecord {
@@ -369,14 +372,14 @@ mod tests {
                         operation: Operation::Add,
                     },
                 },
-            }),
+            },
         );
 
         let collection_uuid_2 = Uuid::from_str("00000000-0000-0000-0000-000000000002").unwrap();
         in_memory_log.add_log(
-            collection_uuid_2.clone(),
-            Box::new(InternalLogRecord {
-                collection_id: collection_uuid_2.clone(),
+            collection_uuid_2,
+            InternalLogRecord {
+                collection_id: collection_uuid_2,
                 log_offset: 0,
                 log_ts: 2,
                 record: LogRecord {
@@ -390,7 +393,7 @@ mod tests {
                         operation: Operation::Add,
                     },
                 },
-            }),
+            },
         );
 
         let mut sysdb = Box::new(SysDb::Test(TestSysDb::new()));