Fmt

Author: matthewmturner

crates/datafusion-app/src/flightsql.rs

@@ -551,8 +551,8 @@ impl FlightSQLContext {
         }
 
         // 5. Validate we got the SQL query in metadata
-        let query_str = sql_query
-            .ok_or_else(|| eyre::eyre!("SQL query not found in response metadata"))?;
+        let query_str =
+            sql_query.ok_or_else(|| eyre::eyre!("SQL query not found in response metadata"))?;
 
         // 6. Decode metrics batch
         // batches_to_flight_data creates [schema, data] for the batch

crates/datafusion-app/src/stats.rs

@@ -1023,13 +1023,20 @@ impl ExecutionStats {
         // Add compute metrics if present
         if let Some(compute) = &self.compute {
             if let Some(elapsed) = compute.elapsed_compute {
-                rows.add("elapsed_compute", elapsed as u64, "duration_ns", None, None, None);
+                rows.add(
+                    "elapsed_compute",
+                    elapsed as u64,
+                    "duration_ns",
+                    None,
+                    None,
+                    None,
+                );
             }
 
             // Helper to add compute metrics for a category
             let add_compute_category = |rows: &mut MetricsTableBuilder,
-                                         compute_stats: &Option<Vec<PartitionsComputeStats>>,
-                                         category: &str| {
+                                        compute_stats: &Option<Vec<PartitionsComputeStats>>,
+                                        category: &str| {
                 if let Some(stats) = compute_stats {
                     for stat in stats {
                         for (partition_id, elapsed) in stat.elapsed_computes.iter().enumerate() {
@@ -1058,10 +1065,7 @@ impl ExecutionStats {
     }
 
     /// Deserialize ExecutionStats from metrics table
-    pub fn from_metrics_table(
-        batch: RecordBatch,
-        query: String,
-    ) -> color_eyre::Result<Self> {
+    pub fn from_metrics_table(batch: RecordBatch, query: String) -> color_eyre::Result<Self> {
         let mut stats_builder = ExecutionStatsBuilder::new(query);
 
         // Extract column arrays
@@ -1191,11 +1195,7 @@ impl ExecutionStatsBuilder {
                 self.compute_metrics
                     .entry(cat.to_string())
                     .or_default()
-                    .push((
-                        operator.unwrap_or("Unknown").to_string(),
-                        partition,
-                        value,
-                    ));
+                    .push((operator.unwrap_or("Unknown").to_string(), partition, value));
             }
             _ => {
                 debug!("Unknown metric: {} (category: {:?})", name, category);
@@ -1275,12 +1275,20 @@ impl ExecutionIOStats {
             time_opening: metrics.get("time_opening").map(|v| create_time(*v)),
             time_scanning: metrics.get("time_scanning").map(|v| create_time(*v)),
             parquet_output_rows: metrics.get("output_rows").map(|v| *v as usize),
-            parquet_pruned_page_index: metrics.get("parquet_page_index_pruned").map(|v| create_count(*v)),
-            parquet_matched_page_index: metrics.get("parquet_page_index_matched").map(|v| create_count(*v)),
+            parquet_pruned_page_index: metrics
+                .get("parquet_page_index_pruned")
+                .map(|v| create_count(*v)),
+            parquet_matched_page_index: metrics
+                .get("parquet_page_index_matched")
+                .map(|v| create_count(*v)),
             parquet_rg_pruned_stats: metrics.get("parquet_rg_pruned").map(|v| create_count(*v)),
             parquet_rg_matched_stats: metrics.get("parquet_rg_matched").map(|v| create_count(*v)),
-            parquet_rg_pruned_bloom_filter: metrics.get("parquet_bloom_pruned").map(|v| create_count(*v)),
-            parquet_rg_matched_bloom_filter: metrics.get("parquet_bloom_matched").map(|v| create_count(*v)),
+            parquet_rg_pruned_bloom_filter: metrics
+                .get("parquet_bloom_pruned")
+                .map(|v| create_count(*v)),
+            parquet_rg_matched_bloom_filter: metrics
+                .get("parquet_bloom_matched")
+                .map(|v| create_count(*v)),
         })
     }
 }

src/cli/mod.rs

@@ -242,9 +242,9 @@ impl CliApp {
             (_, _, false, true, true) => Err(eyre!(
                 "The `benchmark` and `analyze` flags are mutually exclusive"
             )),
-            (false, false, _, _, true) => Err(eyre!(
-                "Analyze requires exactly one command or file"
-            )),
+            (false, false, _, _, true) => {
+                Err(eyre!("Analyze requires exactly one command or file"))
+            }
 
             // Execution cases
             (false, true, _, false, false) => self.execute_files(&self.args.files).await,
@@ -287,9 +287,9 @@ impl CliApp {
             (_, _, _, true, true) => Err(eyre!(
                 "The `benchmark` and `analyze` flags are mutually exclusive"
             )),
-            (false, false, _, _, true) => Err(eyre!(
-                "Analyze requires exactly one command or file"
-            )),
+            (false, false, _, _, true) => {
+                Err(eyre!("Analyze requires exactly one command or file"))
+            }
 
             // Execution cases
             (true, false, false, false, false) => self.execute_commands(&self.args.commands).await,

src/server/flightsql/service.rs

@@ -866,13 +866,15 @@ impl FlightSqlService for FlightSqlServiceImpl {
                 stats.collect_stats(); // Collect IO and compute metrics from plan
 
                 // 3. Convert ExecutionStats to metrics table format
-                let metrics_batch = stats
-                    .to_metrics_table()
-                    .map_err(|e| Status::internal(format!("Metrics serialization failed: {}", e)))?;
+                let metrics_batch = stats.to_metrics_table().map_err(|e| {
+                    Status::internal(format!("Metrics serialization failed: {}", e))
+                })?;
 
                 // 4. Encode metrics batch as FlightData
-                let mut flight_data = batches_to_flight_data(&metrics_batch.schema(), vec![metrics_batch])
-                    .map_err(|e| Status::internal(format!("Failed to encode metrics batch: {}", e)))?;
+                let mut flight_data =
+                    batches_to_flight_data(&metrics_batch.schema(), vec![metrics_batch]).map_err(
+                        |e| Status::internal(format!("Failed to encode metrics batch: {}", e)),
+                    )?;
 
                 // 5. Add SQL query to schema message metadata
                 // The first FlightData message contains the schema
@@ -886,9 +888,7 @@ impl FlightSqlService for FlightSqlServiceImpl {
                     .map(|fd| {
                         // Serialize FlightData to bytes
                         let bytes = fd.encode_to_vec();
-                        arrow_flight::Result {
-                            body: bytes.into(),
-                        }
+                        arrow_flight::Result { body: bytes.into() }
                     })
                     .collect();
 

tests/extension_cases/flightsql.rs

@@ -1618,7 +1618,10 @@ pub async fn test_analyze_with_timing_metrics() {
         output.contains("Physical Planning"),
         "Should contain Physical Planning time"
     );
-    assert!(output.contains("Execution"), "Should contain Execution time");
+    assert!(
+        output.contains("Execution"),
+        "Should contain Execution time"
+    );
     assert!(output.contains("Total"), "Should contain Total time");
 
     fixture.shutdown_and_wait().await;