(datafusion) 01/01: fix: Remove nested async block causing Stacked Borrows violation in PushDecoderStreamState (#21663)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-bot pushed a commit to branch gh-readonly-queue/main/pr-21663-7bfa3fbdf8e0eedfacb844048b3ee764155a1670
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit 8b47f45ad216e6275ea6cc728d807dc6e34e05ff
Author: Matt Butrovich <mb...@users.noreply.github.com>
AuthorDate: Thu Apr 16 07:52:54 2026 -0400

    fix: Remove nested async block causing Stacked Borrows violation in PushDecoderStreamState (#21663)
    
    ## Which issue does this PR close?
    
    - Closes #21662.
    
    ## Rationale for this change
    
    Miri detects a Stacked Borrows violation in
    `PushDecoderStreamState::transition`. A nested `async` block captures
    `&mut self` as a single opaque mutable reference. At the `.await` on
    `get_byte_ranges`, the future yields, and the `Unique` tag on the borrow
    stack is invalidated by a `SharedReadOnly` retag. When the future
    resumes, `push_ranges` attempts a two-phase retag through the
    now-invalidated tag.
    
    This was found by [Apache DataFusion
    Comet](https://github.com/apache/datafusion-comet), which runs Miri in
    CI.
    
    ## What changes are included in this PR?
    
    Remove the nested `async` block in the `NeedsData` arm of
    `PushDecoderStreamState::transition` and inline the IO
    (`get_byte_ranges`) and CPU (`push_ranges`) operations as separate
    statements. Since `transition` is already an `async fn`, the `.await`
    works directly in the loop body. Without the nested block, the compiler
    can split the borrows of `self.reader` and `self.decoder` into disjoint
    field borrows, keeping the borrow stack valid across the yield point.
    
    Also removes the now-unused `parquet::errors::ParquetError` import.
    
    ## Are these changes tested?
    
    Covered by existing parquet reader tests. The original violation was
    caught by Miri, which DataFusion does not currently run in CI.
    
    ## Are there any user-facing changes?
    
    No.
---
 datafusion/datasource-parquet/src/opener.rs | 25 +++++++++++++++++--------
 1 file changed, 17 insertions(+), 8 deletions(-)

diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
index 6d1c22c0b0..72549d5b87 100644
--- a/datafusion/datasource-parquet/src/opener.rs
+++ b/datafusion/datasource-parquet/src/opener.rs
@@ -75,7 +75,6 @@ use parquet::arrow::parquet_column;
 use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder};
 use parquet::basic::Type;
 use parquet::bloom_filter::Sbbf;
-use parquet::errors::ParquetError;
 use parquet::file::metadata::{PageIndexPolicy, ParquetMetaDataReader};
 
 /// Stateless Parquet morselizer implementation.
@@ -1253,13 +1252,23 @@ impl PushDecoderStreamState {
         loop {
             match self.decoder.try_decode() {
                 Ok(DecodeResult::NeedsData(ranges)) => {
-                    let fetch = async {
-                        let data = self.reader.get_byte_ranges(ranges.clone()).await?;
-                        self.decoder.push_ranges(ranges, data)?;
-                        Ok::<_, ParquetError>(())
-                    };
-                    if let Err(e) = fetch.await {
-                        return Some(Err(DataFusionError::from(e)));
+                    // 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())
+                        .await
+                        .map_err(DataFusionError::from);
+                    match data {
+                        Ok(data) => {
+                            if let Err(e) = self.decoder.push_ranges(ranges, data) {
+                                return Some(Err(DataFusionError::from(e)));
+                            }
+                        }
+                        Err(e) => return Some(Err(e)),
                     }
                 }
                 Ok(DecodeResult::Data(batch)) => {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org