refactor(parquet): extract InitialDecoderState from the opener decoder-setup block

datafusion/datasource-parquet/src/opener/mod.rs

@@ -27,7 +27,8 @@ use crate::access_plan::PreparedAccessPlan;
 use crate::decoder_projection::DecoderProjection;
 use crate::page_filter::PagePruningAccessPlanFilter;
 use crate::push_decoder::{
-    DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner,
+    DecoderBuilderConfig, InitialDecoderState, PushDecoderStreamState, RgPlanEntry,
+    RowGroupPruner,
 };
 use crate::row_filter::RowFilterGenerator;
 use crate::row_group_filter::RowGroupAccessPlanFilter;
@@ -1435,7 +1436,11 @@ impl RowGroupsPrunedParquetOpen {
             prepared.virtual_state.as_deref(),
         )?;
 
-        let (decoder, rg_plan, has_row_selection) = {
+        let InitialDecoderState {
+            decoder,
+            rg_plan,
+            has_row_selection,
+        } = {
             let pushdown_predicate = prepared
                 .pushdown_filters
                 .then_some(prepared.predicate.as_ref())
@@ -1494,7 +1499,11 @@ impl RowGroupsPrunedParquetOpen {
                 }
             }
 
-            (builder.build()?, rg_plan, has_row_selection)
+            InitialDecoderState {
+                decoder: builder.build()?,
+                rg_plan,
+                has_row_selection,
+            }
         };
 
         let predicate_cache_inner_records =

datafusion/datasource-parquet/src/push_decoder.rs

@@ -111,6 +111,21 @@ pub(crate) struct RgPlanEntry {
     pub(crate) rg_index: usize,
 }
 
+/// The initial per-file decoder state the opener builds and hands off to the
+/// [`PushDecoderStreamState`] stream driver.
+///
+/// Named rather than a bare tuple so the fields carried out of the decoder
+/// setup block stay self-documenting as more are added.
+pub(crate) struct InitialDecoderState {
+    /// The freshly built push decoder for this file.
+    pub(crate) decoder: ParquetPushDecoder,
+    /// The per-row-group plan, in the physical scan order the decoder reads.
+    pub(crate) rg_plan: VecDeque<RgPlanEntry>,
+    /// Whether a row selection is live for this scan. Runtime row-group
+    /// pruning is disabled when it is (see the opener for why).
+    pub(crate) has_row_selection: bool,
+}
+
 /// Runtime row-group pruner driven by a dynamic predicate (e.g. the
 /// threshold expression a `TopK` operator pushes down).
 ///