style: nightly rustfmt + drop useless borrows in assert!

Reformats doc comments / format strings under nightly rustfmt
(`wrap_comments`, `format_strings`), and removes two redundant `&` in
`assert!` arguments flagged by clippy.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: g-talbot

quickwit/quickwit-config/src/node_config/mod.rs

@@ -453,9 +453,7 @@ impl CacheConfig {
     fn deserialize_with_default<'de, D, const DEFAULT_CAPACITY: u64>(
         deserializer: D,
     ) -> Result<CacheConfig, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
+    where D: Deserializer<'de> {
         use serde_with::{DeserializeAs, FromInto, PickFirst, Same};
 
         let mut cache_config: CacheConfig =

quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_executor.rs

@@ -16,21 +16,19 @@
 //!
 //! Receives a `ParquetMergeScratch` from the downloader. Two engines are available:
 //!
-//! - **Streaming engine** (`execute_merge_operation`): column-major, page-bounded body cache.
-//!   Used unconditionally for promotion merges (the in-memory path can't handle mixed
-//!   prefix lengths). Optionally used for regular merges when the node-level
+//! - **Streaming engine** (`execute_merge_operation`): column-major, page-bounded body cache. Used
+//!   unconditionally for promotion merges (the in-memory path can't handle mixed prefix lengths).
+//!   Optionally used for regular merges when the node-level
 //!   `IndexerConfig::parquet_merge_use_streaming_engine` flag is true.
-//! - **In-memory engine** (`merge_sorted_parquet_files`): buffers all inputs in memory and
-//!   runs inside `run_cpu_intensive`. Kept as the runtime fallback so production can flip
-//!   back via YAML config if the streaming engine hits a bug. To be removed once the
-//!   streaming path has soaked.
+//! - **In-memory engine** (`merge_sorted_parquet_files`): buffers all inputs in memory and runs
+//!   inside `run_cpu_intensive`. Kept as the runtime fallback so production can flip back via YAML
+//!   config if the streaming engine hits a bug. To be removed once the streaming path has soaked.
 //!
 //! Routing in `handle()`:
 //!
-//! - `target_prefix_len_override.is_some()` → streaming engine. Promotion is the whole point
-//!   of `target_prefix_len_override`, and the in-memory path's
-//!   `extract_and_validate_input_metadata` would bail on the mixed `rg_partition_prefix_len`
-//!   before any output is produced.
+//! - `target_prefix_len_override.is_some()` → streaming engine. Promotion is the whole point of
+//!   `target_prefix_len_override`, and the in-memory path's `extract_and_validate_input_metadata`
+//!   would bail on the mixed `rg_partition_prefix_len` before any output is produced.
 //! - Else `use_streaming_engine == true` → streaming engine (the new default once soaked).
 //! - Else → in-memory engine (the runtime fallback).
 //!

quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline_multi_metric_test.rs

@@ -18,27 +18,24 @@
 //! (which covers the simpler two-input, one-metric-per-input case) with
 //! the harder scenarios:
 //!
-//! - **Three inputs**, each carrying **three metrics** (`aaa.alpha`, `bbb.beta`,
-//!   `ccc.gamma`). Across inputs, the metrics overlap and the per-metric
-//!   timeseries IDs collide (each row's `timeseries_id` is derived from the
-//!   metric name, so input-x, input-y, input-z all share the same set of IDs
-//!   per metric). Timestamps within each (metric, timeseries) overlap across
-//!   inputs but are unique — the merge must interleave rows from all three
-//!   inputs heavily, not concatenate them.
-//! - **Multi-row-group output** via `ParquetWriterConfig::row_group_size = 50`
-//!   on the n=1 tests, so the 180-row merge output breaks into 4 row groups.
-//!   Exercises the writer's multi-RG path in both engines.
-//! - **Multi-row-group inputs with `rg_partition_prefix_len = 1`** in the
-//!   bonus tests (`write_prefix_aligned_input`): the writer flushes one row
-//!   group per distinct `metric_name`, so each input file carries three row
-//!   groups in alignment with the sort prefix. The streaming engine reads
-//!   these through its prefix-aware fast path.
+//! - **Three inputs**, each carrying **three metrics** (`aaa.alpha`, `bbb.beta`, `ccc.gamma`).
+//!   Across inputs, the metrics overlap and the per-metric timeseries IDs collide (each row's
+//!   `timeseries_id` is derived from the metric name, so input-x, input-y, input-z all share the
+//!   same set of IDs per metric). Timestamps within each (metric, timeseries) overlap across inputs
+//!   but are unique — the merge must interleave rows from all three inputs heavily, not concatenate
+//!   them.
+//! - **Multi-row-group output** via `ParquetWriterConfig::row_group_size = 50` on the n=1 tests, so
+//!   the 180-row merge output breaks into 4 row groups. Exercises the writer's multi-RG path in
+//!   both engines.
+//! - **Multi-row-group inputs with `rg_partition_prefix_len = 1`** in the bonus tests
+//!   (`write_prefix_aligned_input`): the writer flushes one row group per distinct `metric_name`,
+//!   so each input file carries three row groups in alignment with the sort prefix. The streaming
+//!   engine reads these through its prefix-aware fast path.
 //! - **m:n merges** in the bonus tests: a small
-//!   `ParquetMergePipelineParams::target_split_size_bytes` forces the
-//!   executor to ask the engine for `num_outputs > 1`. The bonus
-//!   assertions cover the multi-output contract — sum-equals-total,
-//!   internal monotonicity, inter-output `sorted_series` disjointness,
-//!   and union-equals-full-set on metrics/services.
+//!   `ParquetMergePipelineParams::target_split_size_bytes` forces the executor to ask the engine
+//!   for `num_outputs > 1`. The bonus assertions cover the multi-output contract —
+//!   sum-equals-total, internal monotonicity, inter-output `sorted_series` disjointness, and
+//!   union-equals-full-set on metrics/services.
 //!
 //! Both `ParquetMergePipelineParams::use_streaming_engine = false` (the
 //! in-memory engine) and `= true` (the streaming engine) are exercised
@@ -689,14 +686,11 @@ async fn assert_three_input_three_metric_single_output_correct(
 /// - The pipeline staged at least two output splits (proves splitting happened).
 /// - The sum of per-output row counts equals the total input row count.
 /// - Each output is internally sorted on `sorted_series`.
-/// - Across outputs, the `sorted_series` partition is **disjoint** (no two
-///   outputs share any `sorted_series` value — the merge engine splits at
-///   series boundaries, never inside).
-/// - The union of metric_names / services across outputs covers the full
-///   input set.
-/// - Every output declares `num_merge_ops = 1` (first merge over level-0
-///   inputs) and has `row_keys_proto` + `metric_name` zonemap regex
-///   populated.
+/// - Across outputs, the `sorted_series` partition is **disjoint** (no two outputs share any
+///   `sorted_series` value — the merge engine splits at series boundaries, never inside).
+/// - The union of metric_names / services across outputs covers the full input set.
+/// - Every output declares `num_merge_ops = 1` (first merge over level-0 inputs) and has
+///   `row_keys_proto` + `metric_name` zonemap regex populated.
 async fn assert_three_input_three_metric_multi_output_correct(
     staged_metadata: &Arc<std::sync::Mutex<Vec<ParquetSplitMetadata>>>,
     replaced_ids: &Arc<std::sync::Mutex<Vec<String>>>,
@@ -749,8 +743,8 @@ async fn assert_three_input_three_metric_multi_output_correct(
         let series = extract_binary_column(&batch, SORTED_SERIES_COLUMN);
         assert!(
             !series.is_empty(),
-            "every output must have at least one row (empty outputs should be dropped \
-             by the engine)"
+            "every output must have at least one row (empty outputs should be dropped by the \
+             engine)"
         );
         for i in 1..series.len() {
             assert!(
@@ -773,8 +767,7 @@ async fn assert_three_input_three_metric_multi_output_correct(
         let (right_min, _, right_file) = &window[1];
         assert!(
             left_max < right_min,
-            "outputs {} and {} overlap on sorted_series: \
-             left max = {:?}, right min = {:?}",
+            "outputs {} and {} overlap on sorted_series: left max = {:?}, right min = {:?}",
             left_file,
             right_file,
             left_max,
@@ -929,8 +922,8 @@ async fn test_multi_metric_three_input_single_output_streaming_engine() {
         |staged, replaced, storage| async move {
             assert!(
                 PEAK_BODY_COL_PAGE_CACHE_LEN.load(Ordering::Relaxed) > 0,
-                "streaming engine did not write to PEAK_BODY_COL_PAGE_CACHE_LEN — \
-                 routing may have silently fallen back to the in-memory engine",
+                "streaming engine did not write to PEAK_BODY_COL_PAGE_CACHE_LEN — routing may \
+                 have silently fallen back to the in-memory engine",
             );
             assert_three_input_three_metric_single_output_correct(&staged, &replaced, &storage)
                 .await;
@@ -1052,8 +1045,8 @@ async fn test_prefix_aligned_multi_metric_three_input_multi_output_streaming_eng
         |staged, replaced, storage| async move {
             assert!(
                 PEAK_BODY_COL_PAGE_CACHE_LEN.load(Ordering::Relaxed) > 0,
-                "streaming engine did not write to PEAK_BODY_COL_PAGE_CACHE_LEN — \
-                 routing may have silently fallen back to the in-memory engine",
+                "streaming engine did not write to PEAK_BODY_COL_PAGE_CACHE_LEN — routing may \
+                 have silently fallen back to the in-memory engine",
             );
             assert_three_input_three_metric_multi_output_correct(
                 &staged,

quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline_test.rs

@@ -447,19 +447,17 @@ async fn test_merge_pipeline_end_to_end() {
 ///
 /// Verifies, in order:
 /// - **Step 5**: `replaced_split_ids = [split-a, split-b]`.
-/// - **Step 6**: staged `ParquetSplitMetadata` — `num_rows = 100`,
-///   `time_range = [100, 250)`, `metric_names = {cpu.usage, mem.usage}`,
-///   `num_merge_ops = 1`, `sort_fields` matches the table config,
-///   `row_keys_proto` present + non-empty, `zonemap_regexes` contains
-///   `metric_name`, `low_cardinality_tags["service"] = {web, api}`.
-/// - **Step 7**: merged Parquet file content — row count, all 100 timestamps
-///   match expected and metadata `time_range`, both service / host values
-///   survive, `sorted_series` is monotonically non-decreasing, and
-///   `cpu.usage` rows precede `mem.usage` rows (the global sort order
+/// - **Step 6**: staged `ParquetSplitMetadata` — `num_rows = 100`, `time_range = [100, 250)`,
+///   `metric_names = {cpu.usage, mem.usage}`, `num_merge_ops = 1`, `sort_fields` matches the table
+///   config, `row_keys_proto` present + non-empty, `zonemap_regexes` contains `metric_name`,
+///   `low_cardinality_tags["service"] = {web, api}`.
+/// - **Step 7**: merged Parquet file content — row count, all 100 timestamps match expected and
+///   metadata `time_range`, both service / host values survive, `sorted_series` is monotonically
+///   non-decreasing, and `cpu.usage` rows precede `mem.usage` rows (the global sort order
 ///   semantics).
-/// - **Step 8**: Parquet KV metadata — `qh.sort_fields`, `qh.num_merge_ops`,
-///   `qh.row_keys` (non-empty), `qh.zonemap_regexes` parses as JSON and
-///   contains `metric_name`, and cross-validates with the staged metadata.
+/// - **Step 8**: Parquet KV metadata — `qh.sort_fields`, `qh.num_merge_ops`, `qh.row_keys`
+///   (non-empty), `qh.zonemap_regexes` parses as JSON and contains `metric_name`, and
+///   cross-validates with the staged metadata.
 async fn assert_cpu_mem_merge_outputs_correct(
     staged_metadata: &Arc<std::sync::Mutex<Vec<ParquetSplitMetadata>>>,
     replaced_ids: &Arc<std::sync::Mutex<Vec<String>>>,
@@ -639,8 +637,8 @@ async fn assert_cpu_mem_merge_outputs_correct(
             sorted_series[i] >= sorted_series[i - 1],
             "sorted_series must be monotonically non-decreasing at row {}: {:?} < {:?}",
             i,
-            &sorted_series[i],
-            &sorted_series[i - 1]
+            sorted_series[i],
+            sorted_series[i - 1]
         );
     }
 
@@ -718,20 +716,18 @@ async fn assert_cpu_mem_merge_outputs_correct(
 /// reachable path — there is no silent fallback. The test confirms
 /// this by:
 ///
-/// 1. Asserting the merge published with the right `replaced_split_ids`
-///    (the merge actually ran end-to-end through the executor).
-/// 2. Reading `PEAK_BODY_COL_PAGE_CACHE_LEN` and asserting it is
-///    non-zero (the streaming engine writes to this atomic on every
-///    body-col page assembly; if the in-memory engine had run instead
-///    the counter would stay at zero).
-/// 3. Asserting row count and metric names on the output match the
-///    inputs (the streaming engine produces correct results, not just
-///    "something").
+/// 1. Asserting the merge published with the right `replaced_split_ids` (the merge actually ran
+///    end-to-end through the executor).
+/// 2. Reading `PEAK_BODY_COL_PAGE_CACHE_LEN` and asserting it is non-zero (the streaming engine
+///    writes to this atomic on every body-col page assembly; if the in-memory engine had run
+///    instead the counter would stay at zero).
+/// 3. Asserting row count and metric names on the output match the inputs (the streaming engine
+///    produces correct results, not just "something").
 #[allow(
     clippy::await_holding_lock,
-    reason = "the lock is `std::sync::Mutex` and the `#[tokio::test]` runtime is \
-              single-threaded, so holding the guard across await won't deadlock another \
-              thread — see `ms7_serial_lock` rationale"
+    reason = "the lock is `std::sync::Mutex` and the `#[tokio::test]` runtime is single-threaded, \
+              so holding the guard across await won't deadlock another thread — see \
+              `ms7_serial_lock` rationale"
 )]
 #[tokio::test]
 async fn test_merge_pipeline_end_to_end_with_streaming_engine_flag() {

quickwit/quickwit-parquet-engine/src/merge/tests.rs

@@ -1923,13 +1923,10 @@ mod parity {
 
     /// Verify the m:n merge contract on a single engine's outputs:
     ///
-    /// 1. Sum of per-output row counts equals the total input row count
-    ///    (no duplication, no loss).
-    /// 2. Within each output, the `sorted_series` column is monotonically
-    ///    non-decreasing.
-    /// 3. Across outputs, after sorting by min `sorted_series`, every
-    ///    output's max sorted_series is strictly less than the next
-    ///    output's min — the partition is disjoint on the keyspace.
+    /// 1. Sum of per-output row counts equals the total input row count (no duplication, no loss).
+    /// 2. Within each output, the `sorted_series` column is monotonically non-decreasing.
+    /// 3. Across outputs, after sorting by min `sorted_series`, every output's max sorted_series is
+    ///    strictly less than the next output's min — the partition is disjoint on the keyspace.
     ///
     /// Holds for any merge with `num_outputs >= 1`. Trivial for n=1
     /// (only invariant 2 is non-trivial there).
@@ -1976,8 +1973,7 @@ mod parity {
             let (right_min, _, right_path) = &window[1];
             assert!(
                 left_max < right_min,
-                "outputs {} and {} overlap on sorted_series: \
-                 left max = {:?}, right min = {:?}",
+                "outputs {} and {} overlap on sorted_series: left max = {:?}, right min = {:?}",
                 left_path.display(),
                 right_path.display(),
                 left_max,