chore: fix clippy lints for rust 1.97

src/alerts/target.rs

@@ -142,7 +142,7 @@ impl TargetConfigs {
             return Err(AlertError::CustomError("No AlertManager set".into()));
         };
 
-        for (_, alert) in alerts.get_all_alerts(tenant_id).await.iter() {
+        for alert in alerts.get_all_alerts(tenant_id).await.values() {
             if alert.get_targets().contains(target_id) {
                 return Err(AlertError::TargetInUse);
             }

src/connectors/kafka/metrics.rs

@@ -820,7 +820,7 @@ impl Collector for KafkaMetricsCollector {
         mfs.extend(self.core_metrics.collect_metrics(&stats));
 
         // Collect broker metrics
-        for (_broker_id, broker) in stats.brokers.iter() {
+        for broker in stats.brokers.values() {
             mfs.extend(self.broker_metrics.collect_metrics(broker));
         }
 

src/handlers/airplane.rs

@@ -156,7 +156,7 @@ impl FlightService for AirServiceImpl {
             query.time_range.start.timestamp_millis(),
             query.time_range.end.timestamp_millis(),
         ) {
-            let sql = format!("select * from \"{}\"", &stream_name);
+            let sql = format!("select * from \"{}\"", stream_name);
             let start_time = ticket.start_time.clone();
             let end_time = ticket.end_time.clone();
             let out_ticket = json!({

src/handlers/http/alerts.rs

@@ -419,7 +419,7 @@ pub async fn update_notification_state(
             } else {
                 return Err(AlertError::InvalidStateChange(format!(
                     "Invalid notification state change request. Expected `notify`, `indefinite` or human-time or UTC datetime. Got `{}`",
-                    &new_notification_state.state
+                    new_notification_state.state
                 )));
             };
             NotificationState::Mute(till_time)

src/handlers/http/cluster/mod.rs

@@ -1984,7 +1984,7 @@ pub async fn send_query_request(
     let send_null = query_request.send_null;
     let uri = format!(
         "{}api/v1/query?fields={fields}&streaming={streaming}&send_null={send_null}",
-        &querier.domain_name,
+        querier.domain_name,
     );
 
     let body = match serde_json::to_string(&query_request) {

src/metastore/metastores/object_store_metastore.rs

@@ -965,9 +965,9 @@ impl Metastore for ObjectStoreMetastore {
             async move {
                 let t_start = std::time::Instant::now();
                 let date_path = if let Some(tenant) = tenant.as_ref() {
-                    object_store::path::Path::from(format!("{}/{}/{}", tenant, &stream, &date))
+                    object_store::path::Path::from(format!("{}/{}/{}", tenant, stream, date))
                 } else {
-                    object_store::path::Path::from(format!("{}/{}", &stream, &date))
+                    object_store::path::Path::from(format!("{}/{}", stream, date))
                 };
 
                 let t_list = std::time::Instant::now();
@@ -1420,7 +1420,7 @@ impl Metastore for ObjectStoreMetastore {
                         STREAM_ROOT_DIRECTORY,
                     ])
                 } else {
-                    object_store::path::Path::from(format!("{}/{}", &stream, STREAM_ROOT_DIRECTORY))
+                    object_store::path::Path::from(format!("{}/{}", stream, STREAM_ROOT_DIRECTORY))
                 };
                 let resp = self.storage.list_with_delimiter(Some(stream_path)).await?;
                 if resp

src/metrics/prom_utils.rs

@@ -228,7 +228,7 @@ impl Metrics {
     ) -> Result<(String, String), PostError> {
         let uri = Url::parse(&format!(
             "{}{}/about",
-            &metadata.domain_name(),
+            metadata.domain_name(),
             base_path_without_preceding_slash()
         ))
         .map_err(|err| {
@@ -260,7 +260,7 @@ impl Metrics {
             );
             Err(PostError::Invalid(anyhow::anyhow!(
                 "Failed to fetch about API response from server: {}\n",
-                &metadata.domain_name()
+                metadata.domain_name()
             )))
         }
     }

src/parseable/streams.rs

@@ -966,13 +966,11 @@ impl Stream {
         .collect();
 
         for res in _schemas {
-            match res {
-                Ok(s) => {
-                    if let Some(s) = s {
-                        schemas.push(s)
-                    }
+            {
+                let s = res?;
+                if let Some(s) = s {
+                    schemas.push(s)
                 }
-                Err(e) => return Err(e),
             }
         }
         if schemas.is_empty() {

src/query/listing_table_builder.rs

@@ -91,7 +91,7 @@ impl ListingTableBuilder {
         let prefixes: Vec<_> = prefixes
             .into_iter()
             .map(|prefix| {
-                relative_path::RelativePathBuf::from(format!("{}/{}", &self.stream, prefix))
+                relative_path::RelativePathBuf::from(format!("{}/{}", self.stream, prefix))
             })
             .collect();
 

src/rbac/mod.rs

@@ -336,8 +336,8 @@ impl Users {
     }
 
     pub fn get_user_tenant_from_basic(&self, username: &str, password: &str) -> Option<String> {
-        for (_, usermap) in users().iter() {
-            for (_, user) in usermap.iter() {
+        for usermap in users().values() {
+            for user in usermap.values() {
                 if let UserType::Native(basic) = &user.ty
                     && basic.username.eq(username)
                     && basic.verify_password(password)

src/sse/mod.rs

@@ -166,7 +166,7 @@ impl Broadcaster {
         } else {
             // broadcast
             let mut futures = vec![];
-            for (_, clients) in tenant_clients.iter() {
+            for clients in tenant_clients.values() {
                 clients
                     .iter()
                     .for_each(|client| futures.push(client.send(sse::Data::new(msg).into())));

src/storage/gcs.rs

@@ -142,7 +142,7 @@ impl ObjectStorageProvider for GcsConfig {
         let object_store_registry = DefaultObjectStoreRegistry::new();
         // Register GCS client under the "gs://" scheme so DataFusion can route
         // object store calls to our GoogleCloudStorage implementation
-        let url = ObjectStoreUrl::parse(format!("gs://{}", &self.bucket_name)).unwrap();
+        let url = ObjectStoreUrl::parse(format!("gs://{}", self.bucket_name)).unwrap();
         object_store_registry.register_store(url.as_ref(), Arc::new(gcs));
 
         RuntimeEnvBuilder::new().with_object_store_registry(Arc::new(object_store_registry))
@@ -1028,7 +1028,7 @@ impl ObjectStorage for Gcs {
         prefixes
             .into_iter()
             .map(|prefix| {
-                let path = format!("gs://{}/{}", &self.bucket, prefix);
+                let path = format!("gs://{}/{}", self.bucket, prefix);
                 ListingTableUrl::parse(path).unwrap()
             })
             .collect()

src/storage/s3.rs

@@ -338,7 +338,7 @@ impl ObjectStorageProvider for S3Config {
         let s3 = MetricLayer::new(s3, "s3");
 
         let object_store_registry = DefaultObjectStoreRegistry::new();
-        let url = ObjectStoreUrl::parse(format!("s3://{}", &self.bucket_name)).unwrap();
+        let url = ObjectStoreUrl::parse(format!("s3://{}", self.bucket_name)).unwrap();
         object_store_registry.register_store(url.as_ref(), Arc::new(s3));
 
         RuntimeEnvBuilder::new().with_object_store_registry(Arc::new(object_store_registry))
@@ -1291,7 +1291,7 @@ impl ObjectStorage for S3 {
         prefixes
             .into_iter()
             .map(|prefix| {
-                let path = format!("s3://{}/{}", &self.bucket, prefix);
+                let path = format!("s3://{}/{}", self.bucket, prefix);
                 ListingTableUrl::parse(path).unwrap()
             })
             .collect()

src/users/filters.rs

@@ -187,7 +187,7 @@ impl Filters {
                     }
                 } else if *filter_type == FilterType::Search || *filter_type == FilterType::Filter {
                     let dataset_name = &f.stream_name;
-                    if user_auth_for_datasets(&permissions, &[dataset_name.to_string()], &tenant_id)
+                    if user_auth_for_datasets(&permissions, std::slice::from_ref(dataset_name), &tenant_id)
                         .await
                         .is_ok()
                     {