add smoke tests and chagne explain docs

Author: gene-bordegaray

datafusion/physical-plan/src/repartition/mod.rs

@@ -3007,13 +3007,31 @@ mod test {
 
         // Verify metrics are available
         let metrics = exec.metrics().unwrap();
-        // Just verify the metrics can be retrieved (spilling may or may not occur)
-        let spill_count = metrics.spill_count().unwrap_or(0);
+        // Verify spilling occurred and the repartition metrics were recorded.
+        let spill_count = metrics.spill_count().unwrap();
         assert!(spill_count > 0);
-        let spilled_bytes = metrics.spilled_bytes().unwrap_or(0);
+        let spilled_bytes = metrics.spilled_bytes().unwrap();
         assert!(spilled_bytes > 0);
-        let spilled_rows = metrics.spilled_rows().unwrap_or(0);
+        let spilled_rows = metrics.spilled_rows().unwrap();
         assert!(spilled_rows > 0);
+        let fetch_time = metrics.sum_by_name("fetch_time").unwrap().as_usize();
+        assert!(fetch_time > 0);
+        let repartition_time =
+            metrics.sum_by_name("repartition_time").unwrap().as_usize();
+        assert!(repartition_time > 0);
+        let route_time = metrics.sum_by_name("route_time").unwrap().as_usize();
+        assert!(route_time > 0);
+        let batch_build_time =
+            metrics.sum_by_name("batch_build_time").unwrap().as_usize();
+        assert!(batch_build_time > 0);
+        assert!(repartition_time >= route_time + batch_build_time);
+        let channel_wait_time =
+            metrics.sum_by_name("channel_wait_time").unwrap().as_usize();
+        assert!(channel_wait_time > 0);
+        let spill_write_time =
+            metrics.sum_by_name("spill_write_time").unwrap().as_usize();
+        assert!(spill_write_time > 0);
+        assert!(metrics.sum_by_name("spill_read_wait_time").is_some());
 
         Ok(())
     }

docs/source/user-guide/sql/explain.md

@@ -236,14 +236,13 @@ EXPLAIN ANALYZE SELECT SUM(x) FROM table GROUP BY b;
 +-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
 | plan_type         | plan                                                                                                                                                      |
 +-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
-| Plan with Metrics | CoalescePartitionsExec, metrics=[]                                                                                                                        |
-|                   |   ProjectionExec: expr=[SUM(table.x)@1 as SUM(x)], metrics=[]                                                                                             |
-|                   |     HashAggregateExec: mode=FinalPartitioned, gby=[b@0 as b], aggr=[SUM(x)], metrics=[outputRows=2]                                                       |
-|                   |       CoalesceBatchesExec: target_batch_size=4096, metrics=[]                                                                                             |
-|                   |         RepartitionExec: partitioning=Hash([Column { name: "b", index: 0 }], 16), metrics=[sendTime=839560, fetchTime=122528525, repartitionTime=5327877] |
-|                   |           HashAggregateExec: mode=Partial, gby=[b@1 as b], aggr=[SUM(x)], metrics=[outputRows=2]                                                          |
-|                   |             RepartitionExec: partitioning=RoundRobinBatch(16), metrics=[fetchTime=5660489, repartitionTime=0, sendTime=8012]                              |
-|                   |               DataSourceExec: file_groups={1 group: [[/tmp/table.csv]]}, has_header=false, metrics=[]                                                        |
+| Plan with Metrics | ProjectionExec: expr=[sum(table.x)@1 as sum(table.x)], metrics=[output_rows=2, elapsed_compute=7.30µs, output_bytes=64.0 B, output_batches=2, expr_0_eval_time=764ns]                                                                                                                                                                                                                                                                                                              |
+|                   |   AggregateExec: mode=FinalPartitioned, gby=[b@0 as b], aggr=[sum(table.x)], metrics=[output_rows=2, elapsed_compute=699.21µs, output_bytes=1088.0 B, output_batches=2, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, peak_mem_used=17.60 K, aggregate_arguments_time=6.55µs, aggregation_time=6.97µs, emitting_time=8.47µs, time_calculating_group_ids=5.89µs]                                                                                                              |
+|                   |     RepartitionExec: partitioning=Hash([b@0], 16), input_partitions=16, metrics=[output_rows=2, elapsed_compute=135.12µs, output_bytes=192.0 KB, output_batches=2, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, batch_build_time=29.35µs, channel_wait_time=4.34µs, fetch_time=14.53ms, repartition_time=49.60µs, route_time=1.97µs, spill_read_wait_time=16ns, spill_write_time=256ns]                                                                                     |
+|                   |       AggregateExec: mode=Partial, gby=[b@1 as b], aggr=[sum(table.x)], metrics=[output_rows=2, elapsed_compute=242.71µs, output_bytes=544.0 B, output_batches=1, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, skipped_aggregation_rows=0, peak_mem_used=5.28 K, aggregate_arguments_time=31.31µs, aggregation_time=19.39µs, emitting_time=9.64µs, time_calculating_group_ids=10.77µs, reduction_factor=20% (2/10)]                                                         |
+|                   |         RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1, metrics=[output_rows=10, elapsed_compute=50.25µs, output_bytes=64.0 KB, output_batches=1, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, batch_build_time=1ns, channel_wait_time=3.43µs, fetch_time=792.21µs, repartition_time=292ns, route_time=1ns, spill_read_wait_time=16ns, spill_write_time=16ns]                                                                                        |
+|                   |           DataSourceExec: file_groups={1 group: [[tmp/datafusion-explain/table.csv]]}, projection=[x, b], file_type=csv, has_header=true, metrics=[output_rows=10, elapsed_compute=582.33µs, output_bytes=128.0 B, output_batches=1, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, time_elapsed_opening=140.33µs, time_elapsed_processing=721.33µs, time_elapsed_scanning_total=613.13µs, time_elapsed_scanning_until_data=588.17µs] |
+|                   |                                                                                                                                                            |
 +-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+
 ```