fix bug

Author: alamb

datafusion/datasource-parquet/src/opener.rs

@@ -659,7 +659,7 @@ impl RowGroupsPreparedParquetOpen {
             file_metrics,
             mut file_pruner,
             metadata_size_hint,
-            async_file_reader: _,
+            async_file_reader,
             logical_file_schema: _,
             output_schema,
             projection: _,
@@ -693,7 +693,7 @@ impl RowGroupsPreparedParquetOpen {
         // --------------------------------------------------------
         let mut access_plan = row_groups.build();
         if !access_plan.is_empty()
-            && let Some(p) = page_pruning_predicate
+            && let Some(ref p) = page_pruning_predicate
         {
             access_plan = p.prune_plan_with_page_index(
                 access_plan,
@@ -719,14 +719,26 @@ impl RowGroupsPreparedParquetOpen {
             return Ok(vec![]);
         }
 
-        let split_by_row_group =
-            limit.is_none() && file_pruner.is_none() && !preserve_order;
+        // Row-group fanout is currently only safe for plain unordered scans.
+        // Filter pushdown and page-index pruning tests compare full output
+        // batches across scan modes, and allowing individual row-group planners
+        // to overtake one another changes the observable row order within a
+        // single file. Keep those scans on the single-planner path until the
+        // scheduler has explicit within-file ordering support for child
+        // planners.
+        let split_by_row_group = limit.is_none()
+            && file_pruner.is_none()
+            && !preserve_order
+            && predicate.is_none()
+            && !pushdown_filters
+            && page_pruning_predicate.is_none();
         let prepared_plans = if split_by_row_group {
             prepared_plan.into_single_row_group_plans(file_metadata.as_ref())?
         } else {
             vec![prepared_plan]
         };
 
+        let mut reusable_reader = Some(async_file_reader);
         let mut reading_states = Vec::with_capacity(prepared_plans.len());
         for prepared_plan in prepared_plans {
             // ---------------------------------------------------------
@@ -811,12 +823,16 @@ impl RowGroupsPreparedParquetOpen {
             let projector = projection.make_projector(&stream_schema)?;
             let push_decoder_state = PushDecoderStreamState {
                 decoder,
-                reader: state.parquet_file_reader_factory.create_reader(
-                    state.partition_index,
-                    partitioned_file.clone(),
-                    metadata_size_hint,
-                    &state.metrics,
-                )?,
+                reader: if let Some(reader) = reusable_reader.take() {
+                    reader
+                } else {
+                    state.parquet_file_reader_factory.create_reader(
+                        state.partition_index,
+                        partitioned_file.clone(),
+                        metadata_size_hint,
+                        &state.metrics,
+                    )?
+                },
                 projector,
                 output_schema: Arc::clone(&output_schema),
                 replace_schema,