Fix clippy warnings

Author: guilload

quickwit/quickwit-actors/src/registry.rs

@@ -166,8 +166,7 @@ impl ActorRegistry {
 
     pub fn get_one<A: Actor>(&self) -> Option<Mailbox<A>> {
         let mut lock = self.actors.write().unwrap();
-        let opt = get_iter::<A>(&mut lock).next();
-        opt
+        get_iter::<A>(&mut lock).next()
     }
 
     fn gc(&self) {

quickwit/quickwit-cli/src/index.rs

@@ -785,14 +785,12 @@ impl IndexStats {
             tables.push(size_stats_table);
         }
 
-        let table = Table::builder(tables.into_iter().map(|table| table.to_string()))
+        Table::builder(tables.into_iter().map(|table| table.to_string()))
             .build()
             .with(Modify::new(Segment::all()).with(Alignment::center_vertical()))
             .with(Disable::row(FirstRow))
             .with(Style::empty())
-            .to_string();
-
-        table
+            .to_string()
     }
 }
 

quickwit/quickwit-common/src/io.rs

@@ -125,10 +125,7 @@ impl IoControls {
 
     pub fn check_if_alive(&self) -> io::Result<ProtectedZoneGuard> {
         if self.kill_switch.is_dead() {
-            return Err(io::Error::new(
-                io::ErrorKind::Other,
-                "Directory kill switch was activated.",
-            ));
+            return Err(io::Error::other("directory kill switch was activated"));
         }
         let guard = self.progress.protect_zone();
         Ok(guard)

quickwit/quickwit-common/src/net.rs

@@ -327,10 +327,9 @@ fn _get_hostname(hostname: OsString) -> io::Result<String> {
     if is_valid_hostname(&hostname_lossy) {
         Ok(hostname_lossy.to_string())
     } else {
-        Err(io::Error::new(
-            io::ErrorKind::Other,
-            format!("invalid hostname: `{hostname_lossy}`"),
-        ))
+        Err(io::Error::other(format!(
+            "invalid hostname: `{hostname_lossy}`"
+        )))
     }
 }
 

quickwit/quickwit-directories/src/bundle_directory.rs

@@ -113,8 +113,7 @@ impl BundleDirectory {
     pub fn open_split(split_file: FileSlice) -> io::Result<BundleDirectory> {
         // First we remove the hotcache from our file slice.
         let (body_and_bundle_metadata, _hot_cache) = split_footer(split_file)?;
-        BundleDirectory::open_bundle(body_and_bundle_metadata)
-            .map_err(|anyhow_err| io::Error::new(io::ErrorKind::Other, anyhow_err))
+        BundleDirectory::open_bundle(body_and_bundle_metadata).map_err(io::Error::other)
     }
 
     /// Opens a BundleDirectory, given a file containing the bundle data.

quickwit/quickwit-directories/src/storage_directory.rs

@@ -111,9 +111,9 @@ impl StorageDirectory {
 }
 
 fn unsupported_operation(path: &Path) -> io::Error {
-    let msg = "Unsupported operation. StorageDirectory only supports async reads";
-    error!(path=?path, msg);
-    io::Error::new(io::ErrorKind::Other, format!("{msg}: {path:?}"))
+    let error = "unsupported operation: `StorageDirectory` only supports async reads";
+    error!(error, ?path);
+    io::Error::other(format!("{error}: {}", path.display()))
 }
 
 impl Directory for StorageDirectory {

quickwit/quickwit-indexing/src/source/file_source.rs

@@ -30,7 +30,7 @@ use crate::source::{Source, SourceContext, SourceRuntime, TypedSourceFactory};
 
 enum FileSourceState {
     #[cfg(feature = "queue-sources")]
-    Notification(QueueCoordinator),
+    Notification(Box<QueueCoordinator>),
     Filepath {
         batch_reader: ObjectUriBatchReader,
         num_bytes_processed: u64,
@@ -180,7 +180,7 @@ impl TypedSourceFactory for FileSourceFactory {
             )) => {
                 let coordinator =
                     QueueCoordinator::try_from_sqs_config(sqs_config, source_runtime).await?;
-                FileSourceState::Notification(coordinator)
+                FileSourceState::Notification(Box::new(coordinator))
             }
             #[cfg(not(feature = "sqs"))]
             FileSourceParams::Notifications(quickwit_config::FileSourceNotification::Sqs(_)) => {

quickwit/quickwit-indexing/src/source/kafka_source.rs

@@ -785,11 +785,10 @@ mod kafka_broker_tests {
     }
 
     fn create_admin_client() -> AdminClient<DefaultClientContext> {
-        let admin_client = ClientConfig::new()
+        ClientConfig::new()
             .set("bootstrap.servers", "localhost:9092")
             .create()
-            .unwrap();
-        admin_client
+            .unwrap()
     }
 
     async fn create_topic(

quickwit/quickwit-ingest/src/mrecordlog_async.rs

@@ -59,10 +59,7 @@ impl MultiRecordLogAsync {
         .await
         .map_err(|join_err| {
             error!(error=?join_err, "failed to load WAL");
-            ReadRecordError::IoError(io::Error::new(
-                io::ErrorKind::Other,
-                "loading wal from directory failed",
-            ))
+            ReadRecordError::IoError(io::Error::other("loading wal from directory failed"))
         })??;
         Ok(Self {
             mrecordlog_opt: Some(mrecordlog),

quickwit/quickwit-jaeger/src/lib.rs

@@ -756,6 +756,7 @@ fn build_aggregations_query(num_traces: usize) -> String {
     query
 }
 
+#[allow(clippy::result_large_err)]
 fn qw_span_to_jaeger_span(qw_span_json: &str) -> Result<JaegerSpan, Status> {
     let mut qw_span: QwSpan = json_deserialize(qw_span_json, "span")?;
 
@@ -1027,6 +1028,7 @@ fn otlp_attributes_to_jaeger_tags_inner(
 
 /// Converts OpenTelemetry links to Jaeger span references.
 /// <https://opentelemetry.io/docs/specs/otel/trace/sdk_exporters/jaeger/#links>
+#[allow(clippy::result_large_err)]
 fn otlp_links_to_jaeger_references(
     trace_id: &TraceId,
     parent_span_id_opt: Option<SpanId>,
@@ -1058,6 +1060,7 @@ fn otlp_links_to_jaeger_references(
     Ok(references)
 }
 
+#[allow(clippy::result_large_err)]
 fn qw_event_to_jaeger_log(event: QwEvent) -> Result<JaegerLog, Status> {
     let timestamp = to_well_known_timestamp(event.event_timestamp_nanos);
     // "OpenTelemetry Event’s name field should be added to Jaeger Log’s fields map as follows: name
@@ -1087,6 +1090,7 @@ fn qw_event_to_jaeger_log(event: QwEvent) -> Result<JaegerLog, Status> {
     Ok(log)
 }
 
+#[allow(clippy::result_large_err)]
 fn collect_trace_ids(
     trace_ids_postcard: &[u8],
 ) -> Result<(Vec<TraceId>, TimeIntervalSecs), Status> {
@@ -1107,6 +1111,7 @@ fn collect_trace_ids(
     Ok((trace_ids, start..=end))
 }
 
+#[allow(clippy::result_large_err)]
 fn json_deserialize<'a, T>(json: &'a str, label: &'static str) -> Result<T, Status>
 where T: Deserialize<'a> {
     match serde_json::from_str(json) {
@@ -1120,6 +1125,7 @@ where T: Deserialize<'a> {
     }
 }
 
+#[allow(clippy::result_large_err)]
 fn postcard_deserialize<'a, T>(json: &'a [u8], label: &'static str) -> Result<T, Status>
 where T: Deserialize<'a> {
     match postcard::from_bytes(json) {

quickwit/quickwit-metastore/src/tests/mod.rs

@@ -88,11 +88,7 @@ async fn create_channel(client: tokio::io::DuplexStream) -> anyhow::Result<Chann
     let channel = Endpoint::try_from("http://test.server")?
         .connect_with_connector(tower::service_fn(move |_: Uri| {
             let client = client.take();
-            async move {
-                client.ok_or_else(|| {
-                    std::io::Error::new(std::io::ErrorKind::Other, "client already taken")
-                })
-            }
+            async move { client.ok_or_else(|| std::io::Error::other("client already taken")) }
         }))
         .await?;
     Ok(channel)

quickwit/quickwit-opentelemetry/src/otlp/mod.rs

@@ -177,6 +177,7 @@ fn is_zero(count: &u32) -> bool {
     *count == 0
 }
 
+#[allow(clippy::result_large_err)]
 pub fn extract_otel_traces_index_id_patterns_from_metadata(
     metadata: &tonic::metadata::MetadataMap,
 ) -> Result<Vec<String>, Status> {
@@ -205,6 +206,7 @@ pub fn extract_otel_traces_index_id_patterns_from_metadata(
     Ok(index_id_patterns)
 }
 
+#[allow(clippy::result_large_err)]
 pub(crate) fn extract_otel_index_id_from_metadata(
     metadata: &tonic::metadata::MetadataMap,
     otel_signal: OtelSignal,

quickwit/quickwit-proto/src/error.rs

@@ -183,6 +183,7 @@ fn decode_error<E: DeserializeOwned>(header_value: &BinaryMetadataValue) -> anyh
     Ok(service_error)
 }
 
+#[allow(clippy::result_large_err)]
 pub fn convert_to_grpc_result<T, E: GrpcServiceError>(
     result: Result<T, E>,
 ) -> tonic::Result<tonic::Response<T>> {

quickwit/quickwit-proto/src/search/mod.rs

@@ -263,6 +263,7 @@ pub fn deserialize_split_fields<R: Read>(mut reader: R) -> io::Result<ListFields
 }
 
 /// Reads the Split fields from a stream of bytes
+#[allow(clippy::unbuffered_bytes)]
 fn read_split_fields_from_zstd<R: Read>(reader: R) -> io::Result<ListFields> {
     let all_bytes: Vec<_> = reader.bytes().collect::<io::Result<_>>()?;
     let serialized_list_fields: ListFields = prost::Message::decode(&all_bytes[..])?;

quickwit/quickwit-serve/src/elasticsearch_api/bulk_v2.rs

@@ -166,6 +166,7 @@ pub(crate) async fn elastic_bulk_ingest_v2(
     )
 }
 
+#[allow(clippy::result_large_err)]
 fn make_elastic_bulk_response_v2(
     ingest_response_v2: IngestResponseV2,
     mut per_subrequest_doc_handles: HashMap<u32, Vec<DocHandle>>,
@@ -348,6 +349,7 @@ fn make_elastic_bulk_response_v2(
     Ok(bulk_response)
 }
 
+#[allow(clippy::result_large_err)]
 fn remove_doc_handles(
     per_subrequest_doc_handles: &mut HashMap<u32, Vec<DocHandle>>,
     subrequest_id: u32,

quickwit/quickwit-serve/src/elasticsearch_api/model/cat_indices.rs

@@ -53,6 +53,7 @@ pub struct CatIndexQueryParams {
     pub v: Option<bool>,
 }
 impl CatIndexQueryParams {
+    #[allow(clippy::result_large_err)]
     pub fn validate(&self) -> Result<(), ElasticsearchError> {
         if let Some(format) = &self.format {
             if format.to_lowercase() != "json" {

quickwit/quickwit-serve/src/elasticsearch_api/model/field_capability.rs

@@ -179,6 +179,7 @@ pub fn convert_to_es_field_capabilities_response(
     FieldCapabilityResponse { indices, fields }
 }
 
+#[allow(clippy::result_large_err)]
 pub fn build_list_field_request_for_es_api(
     index_id_patterns: Vec<String>,
     search_params: FieldCapabilityQueryParams,

quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs

@@ -302,6 +302,7 @@ pub fn es_compat_scroll_handler(
         .boxed()
 }
 
+#[allow(clippy::result_large_err)]
 fn build_request_for_es_api(
     index_id_patterns: Vec<String>,
     search_params: SearchQueryParams,
@@ -418,6 +419,7 @@ fn is_doc_field(field: &quickwit_proto::search::SortField) -> bool {
     field.field_name == "_shard_doc" || field.field_name == "_doc"
 }
 
+#[allow(clippy::result_large_err)]
 fn partial_hit_from_search_after_param(
     search_after: Vec<serde_json::Value>,
     sort_order: &[quickwit_proto::search::SortField],
@@ -991,6 +993,7 @@ fn convert_to_es_stats_response(
     ElasticsearchStatsResponse { _all, indices }
 }
 
+#[allow(clippy::result_large_err)]
 fn convert_to_es_search_response(
     resp: SearchResponse,
     append_shard_doc: bool,

quickwit/quickwit-serve/src/rest.rs

@@ -494,33 +494,34 @@ mod tls {
     use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
     use tokio_rustls::rustls::ServerConfig;
 
-    fn error(err: String) -> io::Error {
-        io::Error::new(io::ErrorKind::Other, err)
+    fn io_error(error: String) -> io::Error {
+        io::Error::other(error)
     }
 
     // Load public certificate from file.
     fn load_certs(filename: &str) -> io::Result<Vec<rustls::Certificate>> {
         // Open certificate file.
-        let certfile =
-            fs::read(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?;
+        let certfile = fs::read(filename)
+            .map_err(|error| io_error(format!("failed to open {filename}: {error}")))?;
 
         // Load and return certificate.
         let certs = rustls_pemfile::certs(&mut certfile.as_ref())
-            .map_err(|_| error("failed to load certificate".into()))?;
+            .map_err(|_| io_error("failed to load certificate".to_string()))?;
         Ok(certs.into_iter().map(rustls::Certificate).collect())
     }
 
     // Load private key from file.
     fn load_private_key(filename: &str) -> io::Result<rustls::PrivateKey> {
         // Open keyfile.
-        let keyfile =
-            fs::read(filename).map_err(|e| error(format!("failed to open {}: {}", filename, e)))?;
+        let keyfile = fs::read(filename)
+            .map_err(|error| io_error(format!("failed to open {filename}: {error}")))?;
 
         // Load and return a single private key.
         let keys = rustls_pemfile::pkcs8_private_keys(&mut keyfile.as_ref())
-            .map_err(|_| error("failed to load private key".into()))?;
+            .map_err(|_| io_error("failed to load private key".to_string()))?;
+
         if keys.len() != 1 {
-            return Err(error(format!(
+            return Err(io_error(format!(
                 "expected a single private key, got {}",
                 keys.len()
             )));
@@ -648,7 +649,7 @@ mod tls {
             .with_safe_defaults()
             .with_no_client_auth()
             .with_single_cert(certs, key)
-            .map_err(|e| error(format!("{}", e)))?;
+            .map_err(|error| io_error(error.to_string()))?;
         // Configure ALPN to accept HTTP/2, HTTP/1.1, and HTTP/1.0 in that order.
         cfg.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
         Ok(Arc::new(cfg))

quickwit/quickwit-storage/src/bundle_storage.rs

@@ -315,7 +315,7 @@ impl fmt::Debug for BundleStorage {
 fn unsupported_operation(paths: &[&Path]) -> StorageError {
     let msg = "Unsupported operation. BundleStorage only supports async reads";
     error!(paths=?paths, msg);
-    io::Error::new(io::ErrorKind::Other, format!("{msg}: {paths:?}")).into()
+    io::Error::other(format!("{msg}: {paths:?}")).into()
 }
 
 #[cfg(test)]

quickwit/quickwit-storage/src/file_descriptor_cache.rs

@@ -138,10 +138,7 @@ impl FileDescriptorCache {
         let file: File = tokio::task::spawn_blocking(move || std::fs::File::open(split_path))
             .await
             .map_err(|join_error| {
-                io::Error::new(
-                    io::ErrorKind::Other,
-                    format!("Failed to open file: {:?}", join_error),
-                )
+                io::Error::other(format!("failed to open file: {join_error:?}"))
             })??;
         let split_file = SplitFile(Arc::new(SplitFileInner {
             num_bytes,

quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs

@@ -27,7 +27,7 @@ use azure_storage::prelude::*;
 use azure_storage_blobs::blob::operations::GetBlobResponse;
 use azure_storage_blobs::prelude::*;
 use bytes::Bytes;
-use futures::io::{Error as FutureError, ErrorKind as FutureErrorKind};
+use futures::io::Error as FutureError;
 use futures::stream::{StreamExt, TryStreamExt};
 use md5::Digest;
 use once_cell::sync::OnceCell;
@@ -349,7 +349,7 @@ impl Storage for AzureBlobStorage {
             let chunk_response = chunk_result.map_err(AzureErrorWrapper::from)?;
             let chunk_response_body_stream = chunk_response
                 .data
-                .map_err(|err| FutureError::new(FutureErrorKind::Other, err))
+                .map_err(FutureError::other)
                 .into_async_read()
                 .compat();
             let mut body_stream_reader = BufReader::new(chunk_response_body_stream);
@@ -448,13 +448,9 @@ impl Storage for AzureBlobStorage {
                 .range(range)
                 .into_stream();
             let mut bytes_stream = page_stream
-                .map(|page_res| {
-                    page_res
-                        .map(|page| page.data)
-                        .map_err(|err| FutureError::new(FutureErrorKind::Other, err))
-                })
+                .map(|page_res| page_res.map(|page| page.data).map_err(FutureError::other))
                 .try_flatten()
-                .map(|e| e.map_err(|err| FutureError::new(FutureErrorKind::Other, err)));
+                .map(|bytes_res| bytes_res.map_err(FutureError::other));
             // Peek into the stream so that any early error can be retried
             let first_chunk = bytes_stream.next().await;
             let reader: Box<dyn AsyncRead + Send + Unpin> = if let Some(res) = first_chunk {
@@ -552,7 +548,7 @@ async fn download_all(
         let chunk_response = chunk_result?;
         let chunk_response_body_stream = chunk_response
             .data
-            .map_err(|err| FutureError::new(FutureErrorKind::Other, err))
+            .map_err(FutureError::other)
             .into_async_read()
             .compat();
         let mut body_stream_reader = BufReader::new(chunk_response_body_stream);