fix: try again to fix Miri in ParquetOpener (#21680)

## Which issue does this PR close?

- Closes #21662.

## Rationale for this change

#21663 removed the nested async block in
`PushDecoderStreamState::transition` to fix a Stacked Borrows violation
under miri. However, miri still flags the same violation because `&mut
self` is still a single opaque borrow — inlining the async block alone
doesn't let miri split the disjoint field borrows across the `.await`
yield point.

Comet CI reproduces this reliably:
https://github.com/apache/datafusion-comet/actions/runs/24518967597/job/71671004017?pr=3916

## What changes are included in this PR?

Change `PushDecoderStreamState::transition` to take `self` by value
instead of `&mut self`. With `&mut self`, the generated future stores a
mutable reference, and when `unfold` pins and polls it, miri sees the
`&mut self` as a single opaque borrow that conflicts across the `.await`
yield point. With owned `self`, the future owns the state directly — no
reference means no Stacked Borrows conflict. The struct fields are all
heap-allocated or reference-counted, so the move is just pointer-sized
copies, not a deep copy.

## Are these changes tested?

Existing tests cover this code path. The fix is validated by miri
passing in CI (the same test that currently fails:
`test_nested_types_extract_missing_struct_names_missing_field`). We'll
run Comet CI against this branch first to confirm the miri violation is
resolved before merging.

## Are there any user-facing changes?

No.

Author: mbutrovich

datafusion/datasource-parquet/src/opener.rs

@@ -1199,10 +1199,7 @@ impl RowGroupsPrunedParquetOpen {
                 predicate_cache_records,
                 baseline_metrics: prepared.baseline_metrics,
             },
-            |mut state| async move {
-                let result = state.transition().await;
-                result.map(|r| (r, state))
-            },
+            |state| async move { state.transition().await },
         )
         .fuse();
 
@@ -1248,15 +1245,15 @@ impl PushDecoderStreamState {
     ///   fetched from the [`AsyncFileReader`] and fed back into the decoder.
     /// - [`Data`](DecodeResult::Data) – a decoded batch is projected and returned.
     /// - [`Finished`](DecodeResult::Finished) – signals end-of-stream (`None`).
-    async fn transition(&mut self) -> Option<Result<RecordBatch>> {
+    ///
+    /// Takes `self` by value (rather than `&mut self`) so the generated future
+    /// owns the state directly. This avoids a Stacked Borrows violation under
+    /// miri where `&mut self` creates a single opaque borrow that conflicts
+    /// with `unfold`'s ownership across yield points.
+    async fn transition(mut self) -> Option<(Result<RecordBatch>, Self)> {
         loop {
             match self.decoder.try_decode() {
                 Ok(DecodeResult::NeedsData(ranges)) => {
-                    // IO (get_byte_ranges) and CPU (push_ranges) are still
-                    // decoupled — they just can't live in a nested async block
-                    // because that captures `&mut self` as one opaque borrow,
-                    // which violates Stacked Borrows across the yield point.
-                    // Inlining lets the compiler split the disjoint field borrows.
                     let data = self
                         .reader
                         .get_byte_ranges(ranges.clone())
@@ -1265,24 +1262,26 @@ impl PushDecoderStreamState {
                     match data {
                         Ok(data) => {
                             if let Err(e) = self.decoder.push_ranges(ranges, data) {
-                                return Some(Err(DataFusionError::from(e)));
+                                return Some((Err(DataFusionError::from(e)), self));
                             }
                         }
-                        Err(e) => return Some(Err(e)),
+                        Err(e) => return Some((Err(e), self)),
                     }
                 }
                 Ok(DecodeResult::Data(batch)) => {
                     let mut timer = self.baseline_metrics.elapsed_compute().timer();
                     self.copy_arrow_reader_metrics();
                     let result = self.project_batch(&batch);
                     timer.stop();
-                    return Some(result);
+                    // Release the borrow on baseline_metrics before moving self
+                    drop(timer);
+                    return Some((result, self));
                 }
                 Ok(DecodeResult::Finished) => {
                     return None;
                 }
                 Err(e) => {
-                    return Some(Err(DataFusionError::from(e)));
+                    return Some((Err(DataFusionError::from(e)), self));
                 }
             }
         }