Rewrite ParquetOpener to use push-based ParquetPushDecoder

Replace the async pull-based ParquetRecordBatchStreamBuilder with
arrow-rs's SansIO ParquetPushDecoder for reading Parquet files.
The caller now controls IO explicitly via DecodeResult::NeedsData,
pushing byte ranges to the decoder and receiving decoded batches.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Dandandan

datafusion/datasource-parquet/src/opener.rs

@@ -621,10 +621,8 @@ impl FileOpener for ParquetOpener {
             let arrow_reader_metrics = ArrowReaderMetrics::enabled();
 
             let indices = projection.column_indices();
-            let mask = ProjectionMask::roots(
-                reader_metadata.parquet_schema(),
-                indices.clone(),
-            );
+            let mask =
+                ProjectionMask::roots(reader_metadata.parquet_schema(), indices.clone());
 
             let decoder = builder
                 .with_projection(mask)
@@ -640,8 +638,7 @@ impl FileOpener for ParquetOpener {
             // Rebase column indices to match the narrowed stream schema.
             // The projection expressions have indices based on physical_file_schema,
             // but the stream only contains the columns selected by the ProjectionMask.
-            let stream_schema =
-                Arc::new(physical_file_schema.project(&indices)?);
+            let stream_schema = Arc::new(physical_file_schema.project(&indices)?);
             let replace_schema = stream_schema != output_schema;
             let projection = projection
                 .try_map_exprs(|expr| reassign_expr_columns(expr, &stream_schema))?;
@@ -686,7 +683,7 @@ impl FileOpener for ParquetOpener {
                                     &state.predicate_cache_inner_records,
                                     &state.predicate_cache_records,
                                 );
-                                let result = state.project_batch(batch);
+                                let result = state.project_batch(&batch);
                                 return Some((result, state));
                             }
                             Ok(DecodeResult::Finished) => {
@@ -698,7 +695,8 @@ impl FileOpener for ParquetOpener {
                         }
                     }
                 },
-            );
+            )
+            .fuse();
 
             // ----------------------------------------------------------------------
             // Step: wrap the stream so a dynamic filter can stop the file scan early
@@ -730,8 +728,8 @@ struct PushDecoderStreamState {
 }
 
 impl PushDecoderStreamState {
-    fn project_batch(&self, batch: RecordBatch) -> Result<RecordBatch> {
-        let mut batch = self.projector.project_batch(&batch)?;
+    fn project_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
+        let mut batch = self.projector.project_batch(batch)?;
         if self.replace_schema {
             let (_schema, arrays, num_rows) = batch.into_parts();
             let options = RecordBatchOptions::new().with_row_count(Some(num_rows));