Fix metrics for repartition when preserve_order=true

[xanderbailey]

Which issue does this PR close?

Found this when working on #20875

• Closes #.

Rationale for this change

Metric reporting was previously incorrect for RepartitionExec if preserve order was set to true.

What changes are included in this PR?

Create new metrics before creating PerPartitionStream

Are these changes tested?

Yes and confirmed that this fails on main:

     thread 'repartition::test::test_preserve_order_output_rows_not_double_counted' (12487869)
      panicked at datafusion/physical-plan/src/repartition/mod.rs:3007:9:
     assertion `left == right` failed: metrics output_rows (8) should match actual rows
     collected (4), not double-count
       left: 8
      right: 4
     note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Are there any user-facing changes?

[alamb]

Thanks @xanderbailey

[alamb]

if we are just going to ignore the metrics, should we just remove them from PerPartitionStream ?

It seems like using a local copy of ExecutionPlanMetrics means they metrics in the PerPartitionStream are no longer accessable. So we can probably just remove the metrics to make it clearer they aren't used

[xanderbailey]

Yeah good question, we still use PerPartitionStream here

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

Line 1089 in 01598d9

Ok(Box::pin(PerPartitionStream::new(

. We can make the metrics optional which is maybe more explicit?

[xanderbailey]

Any thoughts here on what you'd prefer @alamb?

[xanderbailey]

Making it optional made the code inside PerPartitionStream less clean I think but I don't have a strong opinion I have to say so I'm happy to do whatever

[xanderbailey]

@alamb I made the metrics optional in 6ee4c22 let me know if you think that works?

[github-actions]

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.

[xanderbailey]

Not stale please don’t close

[xanderbailey]

I’m away for a week or so and I’ll resolve the conflict then but I’d love to get the one merged if we can

[alamb]

Thanks @xanderbailey -- this looks good to me

[alamb]

I verified this test covers the new code

  cargo test -p datafusion-physical-plan -- --nocapture

This fails without the code change

  thread 'repartition::test::test_preserve_order_output_rows_not_double_counted' panicked at datafusion/physical-plan/src/repartition/mod.rs:2991:9:
  assertion `left == right` failed: metrics output_rows (8) should match actual rows collected (4), not double-count
    left: 8
   right: 4

  Verification