fix: compilation errors and rustfmt in base branch test code

- Fix test assertions using .load(Ordering::Relaxed) on plain u64
  counter fields (ParquetDocProcessorCounters, ParquetIndexerCounters)
- Remove duplicate http-body = "0.4" in quickwit-serve/Cargo.toml
  (workspace already provides http-body 1.0)
- Add missing `use std::sync::Arc` in parquet_indexer tests
- Run rustfmt on all changed files

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

Author: g-talbot

quickwit/Cargo.lock

@@ -143,10 +143,7 @@ impl ParquetDocProcessor {
         indexer_mailbox: Mailbox<ParquetIndexer>,
     ) -> Self {
         let processor = ParquetIngestProcessor::new();
-        let counters = ParquetDocProcessorCounters::new(
-            index_id.clone(),
-            source_id.clone(),
-        );
+        let counters = ParquetDocProcessorCounters::new(index_id.clone(), source_id.clone());
 
         info!(
             index_id = %index_id,
@@ -306,9 +303,8 @@ impl Handler<RawDocBatch> for ParquetDocProcessor {
         // Without this, a batch of consistently malformed data blocks offset progress
         // forever.
         if !checkpoint_forwarded && !checkpoint_delta.is_empty() {
-            let empty_batch = RecordBatch::new_empty(std::sync::Arc::new(
-                arrow::datatypes::Schema::empty(),
-            ));
+            let empty_batch =
+                RecordBatch::new_empty(std::sync::Arc::new(arrow::datatypes::Schema::empty()));
             let processed_batch =
                 ProcessedParquetBatch::new(empty_batch, checkpoint_delta, force_commit);
             ctx.send_message(&self.indexer_mailbox, processed_batch)
@@ -445,8 +441,7 @@ mod tests {
             StdArc::new(DictionaryArray::<Int32Type>::try_new(keys, StdArc::new(vals)).unwrap())
         };
         let metric_type: ArrayRef = StdArc::new(UInt8Array::from(vec![0u8; num_rows]));
-        let timestamp_secs: ArrayRef =
-            StdArc::new(UInt64Array::from(vec![100u64, 101u64, 102u64]));
+        let timestamp_secs: ArrayRef = StdArc::new(UInt64Array::from(vec![100u64, 101u64, 102u64]));
         let value: ArrayRef = StdArc::new(Float64Array::from(vec![42.0, 43.0, 44.0]));
         let service: ArrayRef = {
             let keys = Int32Array::from(vec![0i32; num_rows]);
@@ -699,7 +694,12 @@ mod tests {
 
         // Verify packager produced a split
         let packager_counters = packager_handle.process_pending_and_observe().await.state;
-        assert_eq!(packager_counters.splits_produced.load(Ordering::Relaxed), 1);
+        assert_eq!(
+            packager_counters
+                .splits_produced
+                .load(std::sync::atomic::Ordering::Relaxed),
+            1
+        );
 
         universe.assert_quit().await;
     }

quickwit/quickwit-indexing/src/actors/parquet_doc_processor.rs

@@ -537,9 +537,9 @@ mod tests {
     use quickwit_storage::RamStorage;
 
     use super::*;
+    use crate::actors::parquet_test_helpers::create_test_batch;
     use crate::actors::{
         ParquetPackager, ParquetPublisher, ParquetUploader, SplitsUpdateMailbox, UploaderType,
-        parquet_test_helpers::create_test_batch,
     };
 
     /// Create a test ParquetUploader and return its mailbox.
@@ -617,7 +617,6 @@ mod tests {
         .map_err(|_| anyhow::anyhow!("Timeout waiting for {} staged splits", expected_splits))
     }
 
-
     #[tokio::test]
     async fn test_metrics_indexer_receives_batch() {
         let universe = Universe::with_accelerated_time();

quickwit/quickwit-indexing/src/actors/parquet_indexer.rs

@@ -243,10 +243,8 @@ mod tests {
     use quickwit_storage::RamStorage;
 
     use super::*;
-    use crate::actors::{
-        ParquetPublisher, SplitsUpdateMailbox, UploaderType,
-        parquet_test_helpers::create_test_batch,
-    };
+    use crate::actors::parquet_test_helpers::create_test_batch;
+    use crate::actors::{ParquetPublisher, SplitsUpdateMailbox, UploaderType};
 
     fn create_test_uploader(
         universe: &Universe,

quickwit/quickwit-indexing/src/actors/parquet_packager.rs

@@ -16,7 +16,9 @@
 
 use std::sync::Arc;
 
-use arrow::array::{ArrayRef, DictionaryArray, Float64Array, Int32Array, StringArray, UInt8Array, UInt64Array};
+use arrow::array::{
+    ArrayRef, DictionaryArray, Float64Array, Int32Array, StringArray, UInt8Array, UInt64Array,
+};
 use arrow::datatypes::{DataType, Field, Int32Type, Schema as ArrowSchema};
 use arrow::record_batch::RecordBatch;
 
@@ -57,8 +59,7 @@ pub fn create_test_batch_with_tags(num_rows: usize, tags: &[&str]) -> RecordBatc
     for tag in tags {
         let keys = Int32Array::from(vec![0i32; num_rows]);
         let vals = StringArray::from(vec![*tag]);
-        let col =
-            Arc::new(DictionaryArray::<Int32Type>::try_new(keys, Arc::new(vals)).unwrap());
+        let col = Arc::new(DictionaryArray::<Int32Type>::try_new(keys, Arc::new(vals)).unwrap());
         columns.push(col);
     }
 

quickwit/quickwit-indexing/src/actors/parquet_test_helpers.rs

@@ -96,9 +96,8 @@ impl fmt::Debug for ProcessedParquetBatch {
 
 #[cfg(test)]
 mod tests {
-    use crate::actors::parquet_test_helpers::create_test_batch;
-
     use super::*;
+    use crate::actors::parquet_test_helpers::create_test_batch;
 
     #[test]
     fn test_processed_parquet_batch_new() {

quickwit/quickwit-indexing/src/models/processed_parquet_batch.rs

@@ -15,7 +15,6 @@ anyhow = { workspace = true }
 arrow = { workspace = true }
 async-trait = { workspace = true }
 once_cell = { workspace = true }
-parquet = { workspace = true }
 prost = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

quickwit/quickwit-opentelemetry/Cargo.toml

@@ -24,7 +24,7 @@ use std::io::Cursor;
 use std::sync::Arc;
 
 use arrow::array::{
-    ArrayRef, Float64Builder, RecordBatch, StringDictionaryBuilder, UInt64Builder, UInt8Builder,
+    ArrayRef, Float64Builder, RecordBatch, StringDictionaryBuilder, UInt8Builder, UInt64Builder,
 };
 use arrow::datatypes::{DataType, Field, Int32Type, Schema as ArrowSchema};
 use arrow::ipc::reader::StreamReader;
@@ -131,8 +131,7 @@ impl ArrowMetricsBatchBuilder {
             arrays.push(Arc::new(tag_builder.finish()));
         }
 
-        RecordBatch::try_new(schema, arrays)
-            .expect("record batch should match Arrow schema")
+        RecordBatch::try_new(schema, arrays).expect("record batch should match Arrow schema")
     }
 
     /// Returns the number of rows appended so far.

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

@@ -246,8 +246,7 @@ impl OtlpGrpcMetricsService {
         let num_data_points = data_points.len() as u64 + num_rejected;
 
         // Build Arrow RecordBatch from valid data points
-        let mut arrow_builder =
-            ArrowMetricsBatchBuilder::with_capacity(data_points.len());
+        let mut arrow_builder = ArrowMetricsBatchBuilder::with_capacity(data_points.len());
         let mut doc_uid_generator = DocUidGenerator::default();
         let mut doc_uids = Vec::with_capacity(data_points.len());
 
@@ -746,10 +745,7 @@ mod tests {
         assert_eq!(dp.tags.get("metric_unit").map(|s| s.as_str()), Some("1"));
         assert_eq!(dp.timestamp_secs, 2);
         assert_eq!(dp.value, 100.0); // int converted to f64
-        assert_eq!(
-            dp.tags.get("host").map(|s| s.as_str()),
-            Some("server-1")
-        );
+        assert_eq!(dp.tags.get("host").map(|s| s.as_str()), Some("server-1"));
         assert_eq!(
             dp.tags.get("service").map(|s| s.as_str()),
             Some("counter-service")
@@ -844,10 +840,7 @@ mod tests {
         assert_eq!(dp.tags.get("service").map(|s| s.as_str()), Some("test"));
 
         // Verify data point attributes are in tags as strings
-        assert_eq!(
-            dp.tags.get("string_tag").map(|s| s.as_str()),
-            Some("value")
-        );
+        assert_eq!(dp.tags.get("string_tag").map(|s| s.as_str()), Some("value"));
     }
 
     /// Test metrics with empty and missing values
@@ -893,9 +886,9 @@ mod tests {
             Some("unknown_service")
         );
         // No metric_unit tag when unit is empty
-        assert!(dp.tags.get("metric_unit").is_none());
+        assert!(!dp.tags.contains_key("metric_unit"));
         // No start_timestamp_secs tag when start time is 0
-        assert!(dp.tags.get("start_timestamp_secs").is_none());
+        assert!(!dp.tags.contains_key("start_timestamp_secs"));
         // Only "service" should be in tags (no attributes, no unit, no start time)
         assert_eq!(dp.tags.len(), 1);
     }

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

@@ -14,7 +14,6 @@ license.workspace = true
 arrow = { workspace = true }
 chrono = { workspace = true }
 parquet = { workspace = true }
-prost = { workspace = true }
 quickwit-common = { workspace = true }
 quickwit-proto = { workspace = true }
 sea-query = { workspace = true, optional = true }