fix: metrics displaying on new line (#328)

Previously, metrics were displaying like this:
```
  │   ProjectionExec: expr=[Rainfall@0 as MinTemp, RainToday@1 as RainToday]
  │, metrics=[output_rows_2=33, elapsed_compute_2=4.13µs, output_bytes_2=576.0 KB, output_batches_2=3, expr_0_eval_time_2=541ns, expr_1_eval_time_2=291ns]
  │     FilterExec: Rainfall@0 > 5
  │, metrics=[output_rows_2=33, elapsed_compute_2=38.79µs, output_bytes_2=576.0 KB, output_batches_2=3, selectivity_2=14% (33/234)o# Please enter the commit message for your changes. Lines starting
```

This change removes the extra new line
```
  │   ProjectionExec: expr=[Rainfall@0 as MinTemp, RainToday@1 as RainToday], metrics=[output_rows_2=33, elapsed_compute_2=4.75µs, output_bytes_2=576.0 KB, output_batches_2=3, expr_0_ev
al_time_2=542ns, expr_1_eval_time_2=125ns]
  │     FilterExec: Rainfall@0 > 5, metrics=[output_rows_2=33, elapsed_compute_2=45.25µs, output_bytes_2=576.0 KB, output_batches_2=3, selectivity_2=14% (33/234)]
```
(If you are wondering why we only see metrics for task 2 in this
snippet, it's because this stage is the child of a union).

Display without metrics looks fine still
```
  │   ProjectionExec: expr=[Rainfall@0 as MinTemp, RainToday@1 as RainToday]
  │     FilterExec: Rainfall@0 > 5
```

Author: jayshrivastava

src/distributed_planner/batch_coalescing_below_network_boundaries.rs

@@ -90,7 +90,7 @@ mod tests {
             │ RepartitionExec: partitioning=Hash([RainToday@0, WindGustDir@1], 8), input_partitions=4
             │   AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday, WindGustDir@1 as WindGustDir], aggr=[]
             │     RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-            │       PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+            │       PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
             │         DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[RainToday, WindGustDir], file_type=parquet
             └──────────────────────────────────────────────────
         ");
@@ -118,7 +118,7 @@ mod tests {
             │ RepartitionExec: partitioning=Hash([RainToday@0, WindGustDir@1], 8), input_partitions=4
             │   AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday, WindGustDir@1 as WindGustDir], aggr=[]
             │     RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-            │       PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+            │       PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
             │         DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[RainToday, WindGustDir], file_type=parquet
             └──────────────────────────────────────────────────
         ");
@@ -148,7 +148,7 @@ mod tests {
             │   RepartitionExec: partitioning=Hash([RainToday@0, WindGustDir@1], 8), input_partitions=4
             │     AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday, WindGustDir@1 as WindGustDir], aggr=[]
             │       RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-            │         PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+            │         PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
             │           DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[RainToday, WindGustDir], file_type=parquet
             └──────────────────────────────────────────────────
         ");

src/distributed_planner/distributed_physical_optimizer_rule.rs

@@ -307,7 +307,12 @@ fn display_inner_ascii(
     };
 
     let node_str = displayable(plan.as_ref()).one_line().to_string();
-    writeln!(f, "{} {node_str}{metrics_str}", " ".repeat(indent))?;
+    writeln!(
+        f,
+        "{} {}{metrics_str}",
+        " ".repeat(indent),
+        node_str.trim_end() // remove trailing newline
+    )?;
 
     if plan.is_network_boundary() {
         return Ok(());

src/stage.rs

@@ -63,7 +63,7 @@ mod tests {
             ┌───── Stage 1 ── Tasks: t0:[p0..p5] t1:[p0..p5] t2:[p0..p5] 
             │ RepartitionExec: partitioning=Hash([RainToday@0], 6), input_partitions=1
             │   AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
-            │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+            │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
             │       DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[RainToday], file_type=parquet
             └──────────────────────────────────────────────────
         ",
@@ -143,7 +143,7 @@ mod tests {
             ┌───── Stage 1 ── Tasks: t0:[p0..p5] t1:[p0..p5] t2:[p0..p5] 
             │ RepartitionExec: partitioning=Hash([RainToday@0], 6), input_partitions=1
             │   AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
-            │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+            │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
             │       DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[RainToday], file_type=parquet
             └──────────────────────────────────────────────────
         ",

tests/clickbench_plans_test.rs

@@ -50,7 +50,7 @@ mod tests {
           │   SortExec: expr=[MinTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true]
           │     ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday]
           │       FilterExec: MaxTemp@0 < 30
-          │         PartitionIsolatorExec: t0:[p0,p1,__] t1:[__,__,p0] 
+          │         PartitionIsolatorExec: t0:[p0,p1,__] t1:[__,__,p0]
           │           DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30, pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[]
           └──────────────────────────────────────────────────
         ",

tests/distributed_aggregation.rs

@@ -107,7 +107,7 @@ mod tests {
         // Ensure the distributed plan matches our target plan, registering
         // hive-style partitioning and avoiding data-shuffling repartitions.
         assert_snapshot!(&distributed_plan,
-        @"
+        @r"
         ┌───── DistributedExec ── Tasks: t0:[p0] 
         │ SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST, timestamp@1 ASC NULLS LAST]
         │   [Stage 1] => NetworkCoalesceExec: output_partitions=4, input_tasks=2
@@ -116,9 +116,9 @@ mod tests {
           │ ProjectionExec: expr=[f_dkey@5 as f_dkey, timestamp@3 as timestamp, value@4 as value, env@0 as env, service@1 as service, host@2 as host]
           │   HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@3, f_dkey@2)], projection=[env@0, service@1, host@2, timestamp@4, value@5, f_dkey@6]
           │     FilterExec: service@1 = log
-          │       PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │       PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │         DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/dim/d_dkey=A/data0.parquet], [/testdata/join/parquet/dim/d_dkey=B/data0.parquet], [/testdata/join/parquet/dim/d_dkey=C/data0.parquet], [/testdata/join/parquet/dim/d_dkey=D/data0.parquet]]}, projection=[env, service, host, d_dkey], file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]
-          │     PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │     PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │       DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/fact/f_dkey=A/data0.parquet], [/testdata/join/parquet/fact/f_dkey=B/data0.parquet], [/testdata/join/parquet/fact/f_dkey=C/data0.parquet], [/testdata/join/parquet/fact/f_dkey=D/data0.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
           └──────────────────────────────────────────────────
         ");
@@ -196,9 +196,9 @@ mod tests {
           │     ProjectionExec: expr=[f_dkey@3 as f_dkey, env@0 as env, timestamp@1 as timestamp, value@2 as value]
           │       HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[env@0, timestamp@2, value@3, f_dkey@4]
           │         FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
-          │           PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │           PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │             DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/dim/d_dkey=A/data0.parquet], [/testdata/join/parquet/dim/d_dkey=B/data0.parquet], [/testdata/join/parquet/dim/d_dkey=C/data0.parquet], [/testdata/join/parquet/dim/d_dkey=D/data0.parquet]]}, projection=[env, service, d_dkey], file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]
-          │         PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │         PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │           DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/fact/f_dkey=A/data0.parquet], [/testdata/join/parquet/fact/f_dkey=B/data0.parquet], [/testdata/join/parquet/fact/f_dkey=C/data0.parquet], [/testdata/join/parquet/fact/f_dkey=D/data0.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
           └──────────────────────────────────────────────────
         "#);
@@ -277,9 +277,9 @@ mod tests {
           │         ProjectionExec: expr=[f_dkey@3 as f_dkey, env@0 as env, timestamp@1 as timestamp, value@2 as value]
           │           HashJoinExec: mode=Partitioned, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[env@0, timestamp@2, value@3, f_dkey@4]
           │             FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
-          │               PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │               PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │                 DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/dim/d_dkey=A/data0.parquet], [/testdata/join/parquet/dim/d_dkey=B/data0.parquet], [/testdata/join/parquet/dim/d_dkey=C/data0.parquet], [/testdata/join/parquet/dim/d_dkey=D/data0.parquet]]}, projection=[env, service, d_dkey], file_type=parquet, predicate=service@1 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]
-          │             PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1] 
+          │             PartitionIsolatorExec: t0:[p0,p1,__,__] t1:[__,__,p0,p1]
           │               DataSourceExec: file_groups={4 groups: [[/testdata/join/parquet/fact/f_dkey=A/data0.parquet], [/testdata/join/parquet/fact/f_dkey=B/data0.parquet], [/testdata/join/parquet/fact/f_dkey=C/data0.parquet], [/testdata/join/parquet/fact/f_dkey=D/data0.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, timestamp@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
           └──────────────────────────────────────────────────
         "#);

tests/distributed_unions.rs

@@ -71,7 +71,7 @@ mod tests {
           ┌───── Stage 1 ── Tasks: t0:[p0] t1:[p1] t2:[p2] 
           │ SortExec: expr=[MinTemp@0 DESC], preserve_partitioning=[true]
           │   FilterExec: MinTemp@0 > 20
-          │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0] 
+          │     PartitionIsolatorExec: t0:[p0,__,__] t1:[__,p0,__] t2:[__,__,p0]
           │       StatefulPassThroughExec
           │         DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 20, pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 20, required_guarantees=[]
           └──────────────────────────────────────────────────