Fix rustfmt formatting

Author: mattmkim

quickwit/quickwit-cli/src/tool.rs

@@ -696,18 +696,27 @@ fn print_tantivy_gc_result(
         for split in &removal_info.failed_splits {
             println!(" - {}", split.split_id);
         }
-        println!("{} Splits were unable to be removed.", removal_info.failed_splits.len());
+        println!(
+            "{} Splits were unable to be removed.",
+            removal_info.failed_splits.len()
+        );
     }
 
     let deleted_bytes: u64 = removal_info
         .removed_split_entries
         .iter()
         .map(|s| s.file_size_bytes.as_u64())
         .sum();
-    println!("{}MB of storage garbage collected.", deleted_bytes / 1_000_000);
+    println!(
+        "{}MB of storage garbage collected.",
+        deleted_bytes / 1_000_000
+    );
 
     if removal_info.failed_splits.is_empty() {
-        println!("{} Index successfully garbage collected.", "✔".color(GREEN_COLOR));
+        println!(
+            "{} Index successfully garbage collected.",
+            "✔".color(GREEN_COLOR)
+        );
     } else if removal_info.removed_split_entries.is_empty() {
         println!("{} Failed to garbage collect index.", "✘".color(RED_COLOR));
     } else {
@@ -750,10 +759,16 @@ fn print_parquet_gc_result(
         );
     }
 
-    println!("{}MB of storage garbage collected.", removal_info.removed_bytes() / 1_000_000);
+    println!(
+        "{}MB of storage garbage collected.",
+        removal_info.removed_bytes() / 1_000_000
+    );
 
     if removal_info.failed_parquet_splits.is_empty() {
-        println!("{} Index successfully garbage collected.", "✔".color(GREEN_COLOR));
+        println!(
+            "{} Index successfully garbage collected.",
+            "✔".color(GREEN_COLOR)
+        );
     } else if removal_info.removed_parquet_splits_entries.is_empty() {
         println!("{} Failed to garbage collect index.", "✘".color(RED_COLOR));
     } else {

quickwit/quickwit-index-management/src/garbage_collection.rs

@@ -424,7 +424,11 @@ pub(crate) async fn delete_split_files(
     storage: &dyn Storage,
     splits: Vec<SplitToDelete>,
     progress_opt: Option<&Progress>,
-) -> (Vec<SplitToDelete>, Vec<SplitToDelete>, Option<BulkDeleteError>) {
+) -> (
+    Vec<SplitToDelete>,
+    Vec<SplitToDelete>,
+    Option<BulkDeleteError>,
+) {
     if splits.is_empty() {
         return (Vec::new(), Vec::new(), None);
     }
@@ -438,8 +442,9 @@ pub(crate) async fn delete_split_files(
         Ok(()) => (splits, Vec::new(), None),
         Err(bulk_err) => {
             let success_paths: HashSet<&PathBuf> = bulk_err.successes.iter().collect();
-            let (succeeded, failed) =
-                splits.into_iter().partition(|s| success_paths.contains(&s.path));
+            let (succeeded, failed) = splits
+                .into_iter()
+                .partition(|s| success_paths.contains(&s.path));
             (succeeded, failed, Some(bulk_err))
         }
     }
@@ -478,8 +483,14 @@ pub async fn delete_splits_from_storage_and_metastore(
     let (succeeded_stds, failed_stds, storage_err) =
         delete_split_files(&*storage, splits_to_delete, progress_opt).await;
 
-    let successes: Vec<SplitInfo> = succeeded_stds.iter().map(|s| split_infos[&s.path].clone()).collect();
-    let storage_failures: Vec<SplitInfo> = failed_stds.iter().map(|s| split_infos[&s.path].clone()).collect();
+    let successes: Vec<SplitInfo> = succeeded_stds
+        .iter()
+        .map(|s| split_infos[&s.path].clone())
+        .collect();
+    let storage_failures: Vec<SplitInfo> = failed_stds
+        .iter()
+        .map(|s| split_infos[&s.path].clone())
+        .collect();
 
     let mut storage_error: Option<BulkDeleteError> = None;
     if let Some(bulk_delete_error) = storage_err {

quickwit/quickwit-index-management/src/parquet_garbage_collection.rs

@@ -301,8 +301,7 @@ async fn delete_marked_parquet_splits(
         // To detect if this is the last page, we check if the number of splits is less than the
         // limit.
         assert!(splits.len() <= DELETE_PARQUET_SPLITS_BATCH_SIZE);
-        let splits_to_delete_possibly_remaining =
-            splits.len() == DELETE_PARQUET_SPLITS_BATCH_SIZE;
+        let splits_to_delete_possibly_remaining = splits.len() == DELETE_PARQUET_SPLITS_BATCH_SIZE;
 
         // Set split after which to search for the next loop.
         let Some(last_split) = splits.last() else {
@@ -368,7 +367,10 @@ async fn delete_parquet_splits_from_storage_and_metastore(
 
     let storage_failed: Vec<ParquetSplitInfo> = failed_stds
         .into_iter()
-        .map(|s| ParquetSplitInfo { split_id: s.split_id, file_size_bytes: s.size_bytes })
+        .map(|s| ParquetSplitInfo {
+            split_id: s.split_id,
+            file_size_bytes: s.size_bytes,
+        })
         .collect();
 
     if succeeded_stds.is_empty() {
@@ -381,16 +383,22 @@ async fn delete_parquet_splits_from_storage_and_metastore(
         index_uid: Some(index_uid.clone()),
         split_ids: ids_to_delete,
     };
-    let metastore_result =
-        protect_future(progress_opt, metastore.delete_metrics_splits(delete_request)).await;
+    let metastore_result = protect_future(
+        progress_opt,
+        metastore.delete_metrics_splits(delete_request),
+    )
+    .await;
 
     if let Some(progress) = progress_opt {
         progress.record_progress();
     }
 
     let succeeded: Vec<ParquetSplitInfo> = succeeded_stds
         .into_iter()
-        .map(|s| ParquetSplitInfo { split_id: s.split_id, file_size_bytes: s.size_bytes })
+        .map(|s| ParquetSplitInfo {
+            split_id: s.split_id,
+            file_size_bytes: s.size_bytes,
+        })
         .collect();
 
     match metastore_result {

quickwit/quickwit-janitor/src/actors/garbage_collector.rs

@@ -113,32 +113,13 @@ impl GarbageCollector {
             counters: GarbageCollectorCounters::default(),
         }
     }
-    // if !deleted_file_entries.is_empty() {
-    //         let num_deleted_splits = deleted_file_entries.len();
-    //         let num_deleted_bytes = deleted_file_entries
-    //             .iter()
-    //             .map(|entry| entry.file_size_bytes.as_u64() as usize)
-    //             .sum::<usize>();
-    //         let deleted_files: HashSet<&Path> = deleted_file_entries
-    //             .iter()
-    //             .map(|deleted_entry| deleted_entry.file_name.as_path())
-    //             .take(5)
-    //             .collect();
-    //         info!(
-    //             num_deleted_splits = num_deleted_splits,
-    //             "Janitor deleted {:?} and {} other splits.", deleted_files, num_deleted_splits,
-    //         );
-    //         self.counters.num_deleted_files += num_deleted_splits;
-    //         self.counters.num_deleted_bytes += num_deleted_bytes;
 
     fn record_gc_result(&mut self, result: &GcRunResult, split_type: &str) {
         self.counters.num_failed_splits += result.num_failed;
         if result.num_deleted_splits > 0 {
             info!(
                 "Janitor deleted {:?} and {} other {} splits.",
-                result.sample_deleted_files,
-                result.num_deleted_splits,
-                split_type,
+                result.sample_deleted_files, result.num_deleted_splits, split_type,
             );
             self.counters.num_deleted_files += result.num_deleted_splits;
             self.counters.num_deleted_bytes += result.num_deleted_bytes;
@@ -222,22 +203,40 @@ impl GarbageCollector {
             .await;
 
             let tantivy_run_duration = tantivy_start.elapsed().as_secs();
-            JANITOR_METRICS.gc_seconds_total.with_label_values(["tantivy"]).inc_by(tantivy_run_duration);
+            JANITOR_METRICS
+                .gc_seconds_total
+                .with_label_values(["tantivy"])
+                .inc_by(tantivy_run_duration);
 
             let result = match gc_res {
                 Ok(removal_info) => {
                     self.counters.num_successful_gc_run += 1;
-                    JANITOR_METRICS.gc_runs.with_label_values(["success", "tantivy"]).inc();
+                    JANITOR_METRICS
+                        .gc_runs
+                        .with_label_values(["success", "tantivy"])
+                        .inc();
                     GcRunResult {
                         num_deleted_splits: removal_info.removed_split_entries.len(),
-                        num_deleted_bytes: removal_info.removed_split_entries.iter().map(|e| e.file_size_bytes.as_u64() as usize).sum(),
+                        num_deleted_bytes: removal_info
+                            .removed_split_entries
+                            .iter()
+                            .map(|e| e.file_size_bytes.as_u64() as usize)
+                            .sum(),
                         num_failed: removal_info.failed_splits.len(),
-                        sample_deleted_files: removal_info.removed_split_entries.iter().take(5).map(|e| e.file_name.display().to_string()).collect(),
+                        sample_deleted_files: removal_info
+                            .removed_split_entries
+                            .iter()
+                            .take(5)
+                            .map(|e| e.file_name.display().to_string())
+                            .collect(),
                     }
                 }
                 Err(error) => {
                     self.counters.num_failed_gc_run += 1;
-                    JANITOR_METRICS.gc_runs.with_label_values(["error", "tantivy"]).inc();
+                    JANITOR_METRICS
+                        .gc_runs
+                        .with_label_values(["error", "tantivy"])
+                        .inc();
                     error!(error=?error, "failed to run garbage collection");
                     GcRunResult::failed()
                 }
@@ -260,22 +259,36 @@ impl GarbageCollector {
             .await;
 
             let parquet_run_duration = parquet_start.elapsed().as_secs();
-            JANITOR_METRICS.gc_seconds_total.with_label_values(["parquet"]).inc_by(parquet_run_duration);
+            JANITOR_METRICS
+                .gc_seconds_total
+                .with_label_values(["parquet"])
+                .inc_by(parquet_run_duration);
 
             let result = match gc_res {
                 Ok(removal_info) => {
                     self.counters.num_successful_gc_run += 1;
-                    JANITOR_METRICS.gc_runs.with_label_values(["success", "parquet"]).inc();
+                    JANITOR_METRICS
+                        .gc_runs
+                        .with_label_values(["success", "parquet"])
+                        .inc();
                     GcRunResult {
                         num_deleted_splits: removal_info.removed_split_count(),
                         num_deleted_bytes: removal_info.removed_bytes() as usize,
                         num_failed: removal_info.failed_split_count(),
-                        sample_deleted_files: removal_info.removed_parquet_splits_entries.iter().take(5).map(|e| format!("{}.parquet", e.split_id)).collect(),
+                        sample_deleted_files: removal_info
+                            .removed_parquet_splits_entries
+                            .iter()
+                            .take(5)
+                            .map(|e| format!("{}.parquet", e.split_id))
+                            .collect(),
                     }
                 }
                 Err(error) => {
                     self.counters.num_failed_gc_run += 1;
-                    JANITOR_METRICS.gc_runs.with_label_values(["error", "parquet"]).inc();
+                    JANITOR_METRICS
+                        .gc_runs
+                        .with_label_values(["error", "parquet"])
+                        .inc();
                     error!(error=?error, "failed to run parquet garbage collection");
                     GcRunResult::failed()
                 }
@@ -865,15 +878,13 @@ mod tests {
         let storage_resolver = StorageResolver::unconfigured();
         let mut mock = MockMetastoreService::new();
 
-        mock.expect_list_indexes_metadata()
-            .times(1)
-            .returning(|_| {
-                let indexes = vec![IndexMetadata::for_test(
-                    "otel-metrics-v0_1",
-                    "ram://indexes/otel-metrics-v0_1",
-                )];
-                Ok(ListIndexesMetadataResponse::for_test(indexes))
-            });
+        mock.expect_list_indexes_metadata().times(1).returning(|_| {
+            let indexes = vec![IndexMetadata::for_test(
+                "otel-metrics-v0_1",
+                "ram://indexes/otel-metrics-v0_1",
+            )];
+            Ok(ListIndexesMetadataResponse::for_test(indexes))
+        });
 
         let marked_split = MetricsSplitRecord {
             state: MetricsSplitState::MarkedForDeletion,
@@ -903,10 +914,8 @@ mod tests {
                 Ok(EmptyResponse {})
             });
 
-        let garbage_collect_actor = GarbageCollector::new(
-            MetastoreServiceClient::from_mock(mock),
-            storage_resolver,
-        );
+        let garbage_collect_actor =
+            GarbageCollector::new(MetastoreServiceClient::from_mock(mock), storage_resolver);
         let universe = Universe::with_accelerated_time();
         let (_mailbox, handle) = universe.spawn_builder().spawn(garbage_collect_actor);
 

quickwit/quickwit-janitor/src/lib.rs

@@ -18,7 +18,7 @@ use quickwit_actors::{Mailbox, Universe};
 use quickwit_common::pubsub::EventBroker;
 use quickwit_config::NodeConfig;
 use quickwit_indexing::actors::MergeSchedulerService;
-use quickwit_metastore::{SplitInfo};
+use quickwit_metastore::SplitInfo;
 use quickwit_proto::metastore::MetastoreServiceClient;
 use quickwit_search::SearchJobPlacer;
 use quickwit_storage::StorageResolver;

quickwit/quickwit-janitor/src/metrics.rs

@@ -13,9 +13,7 @@
 // limitations under the License.
 
 use once_cell::sync::Lazy;
-use quickwit_common::metrics::{
-    IntCounterVec, IntGaugeVec, new_counter_vec, new_gauge_vec,
-};
+use quickwit_common::metrics::{IntCounterVec, IntGaugeVec, new_counter_vec, new_gauge_vec};
 
 pub struct JanitorMetrics {
     pub ongoing_num_delete_operations_total: IntGaugeVec<1>,

quickwit/quickwit-janitor/src/retention_policy_execution.rs

@@ -95,7 +95,7 @@ pub async fn run_execute_retention_policy(
 }
 
 /// Detect all expired parquet splits based on a retention policy and
-/// mark them as `MarkedForDeletion`. 
+/// mark them as `MarkedForDeletion`.
 pub async fn run_execute_parquet_retention_policy(
     index_uid: &IndexUid,
     metastore: MetastoreServiceClient,
@@ -133,12 +133,12 @@ pub async fn run_execute_parquet_retention_policy(
         expired_split_ids.len()
     );
 
-    ctx.protect_future(
-        metastore.mark_metrics_splits_for_deletion(MarkMetricsSplitsForDeletionRequest {
+    ctx.protect_future(metastore.mark_metrics_splits_for_deletion(
+        MarkMetricsSplitsForDeletionRequest {
             index_uid: Some(index_uid.clone()),
             split_ids: expired_split_ids,
-        }),
-    )
+        },
+    ))
     .await?;
 
     Ok(expired_splits.len())

quickwit/quickwit-metastore/src/metastore/file_backed/file_backed_index/mod.rs

@@ -862,13 +862,15 @@ impl FileBackedIndex {
 
         // Sort by split_id for stable pagination (mirrors Postgres ORDER BY split_id ASC).
         splits.sort_unstable_by(|a, b| {
-            a.metadata.split_id.as_str().cmp(b.metadata.split_id.as_str())
+            a.metadata
+                .split_id
+                .as_str()
+                .cmp(b.metadata.split_id.as_str())
         });
 
         // Apply cursor: skip splits up to and including after_split_id.
         let splits = if let Some(ref after) = query.after_split_id {
-            let pos =
-                splits.partition_point(|s| s.metadata.split_id.as_str() <= after.as_str());
+            let pos = splits.partition_point(|s| s.metadata.split_id.as_str() <= after.as_str());
             &splits[pos..]
         } else {
             &splits[..]

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

@@ -47,9 +47,9 @@ pub const OTEL_METRICS_INDEX_ID: &str = "otel-metrics-v0_9";
 /// directly and queries via DataFusion. The doc mapping is unused; this config
 /// exists only so the index can be registered in the metastore for source
 /// assignment and lifecycle management.
-/// 
-/// TODO: As a temporary hack, we are including a timestamp_field, so that 
-/// we can pass the retention policy validation. 
+///
+/// TODO: As a temporary hack, we are including a timestamp_field, so that
+/// we can pass the retention policy validation.
 const OTEL_METRICS_INDEX_CONFIG: &str = r#"
 version: 0.8
 
@@ -554,7 +554,9 @@ mod tests {
         let index_config =
             OtlpGrpcMetricsService::index_config(&Uri::for_test("ram:///indexes")).unwrap();
         assert_eq!(index_config.index_id, OTEL_METRICS_INDEX_ID);
-        let retention = index_config.retention_policy_opt.expect("retention policy should be set");
+        let retention = index_config
+            .retention_policy_opt
+            .expect("retention policy should be set");
         assert_eq!(
             retention.retention_period().unwrap(),
             std::time::Duration::from_secs(30 * 24 * 3600)