feat: separate Parquet I/O policy from filter pushdown

datafusion/common/src/config.rs

@@ -1336,6 +1336,13 @@ config_namespace! {
         /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
         pub pushdown_filters: bool, default = false
 
+        /// (reading) Fetch columns progressively as row filtering and decoding need
+        /// them. If false, fetch output and predicate pages together before evaluating
+        /// row filters, preserving page-index pruning performed at file open. This
+        /// can reduce dependent I/O rounds but may fetch pages that row filtering
+        /// would otherwise avoid. Does not enable filter pushdown or prefetch.
+        pub progressive_io: bool, default = true
+
         /// (reading) If true, filter expressions evaluated during the parquet decoding operation
         /// will be reordered heuristically to minimize the cost of evaluation. If false,
         /// the filters are applied in the same order as written in the query

datafusion/common/src/file_options/parquet_writer.rs

@@ -236,6 +236,7 @@ impl ParquetOptions {
             skip_metadata: _,
             metadata_size_hint: _,
             pushdown_filters: _,
+            progressive_io: _,
             reorder_filters: _,
             force_filter_selections: _, // not used for writer props
             allow_single_file_parallelism: _,
@@ -493,6 +494,7 @@ mod tests {
             skip_metadata: defaults.skip_metadata,
             metadata_size_hint: defaults.metadata_size_hint,
             pushdown_filters: defaults.pushdown_filters,
+            progressive_io: defaults.progressive_io,
             reorder_filters: defaults.reorder_filters,
             force_filter_selections: defaults.force_filter_selections,
             allow_single_file_parallelism: defaults.allow_single_file_parallelism,
@@ -579,6 +581,7 @@ mod tests {
 
         TableParquetOptions {
             global: ParquetOptions {
+                progressive_io: true,
                 // global options
                 data_pagesize_limit: props.data_page_size_limit(),
                 write_batch_size: props.write_batch_size(),

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -809,3 +809,180 @@ async fn pushed_down_predicate_reports_the_original_error() {
         "expected the original cast error, got {root:?}"
     );
 }
+
+#[tokio::test]
+async fn upfront_io_preserves_page_pruning_from_session_configuration() {
+    use arrow::array::{Int64Array, StringArray, StructArray};
+    use arrow::datatypes::{DataType, Field, Schema};
+    use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
+    use datafusion_datasource::source::DataSourceExec;
+    use datafusion_datasource_parquet::source::ParquetSource;
+    use parquet::arrow::ArrowWriter;
+    use parquet::file::properties::EnabledStatistics;
+    use std::sync::Arc;
+
+    let nested_fields = vec![Arc::new(Field::new("value", DataType::Int64, false))];
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("key", DataType::Int64, false),
+        Field::new("payload", DataType::Utf8, false),
+        Field::new(
+            "nested",
+            DataType::Struct(nested_fields.clone().into()),
+            false,
+        ),
+    ]));
+    let temp = TempDir::new().unwrap();
+    for clustered in [false, true] {
+        let expected: Vec<i64> = (0..3)
+            .flat_map(|group| {
+                (0..4096)
+                    .filter(move |row| {
+                        if clustered {
+                            *row < 100
+                        } else {
+                            row % 128 < 100
+                        }
+                    })
+                    .map(move |row| group * 4096 + row)
+            })
+            .collect();
+        for indexed in [false, true] {
+            let path = temp
+                .path()
+                .join(format!("pages-{indexed}-{clustered}.parquet"));
+            let props = WriterProperties::builder()
+                .set_max_row_group_row_count(Some(4096))
+                .set_data_page_row_count_limit(128)
+                .set_write_batch_size(128)
+                .set_dictionary_enabled(true)
+                .set_statistics_enabled(if indexed {
+                    EnabledStatistics::Page
+                } else {
+                    EnabledStatistics::Chunk
+                })
+                .set_offset_index_disabled(!indexed)
+                .build();
+            let mut writer = ArrowWriter::try_new(
+                File::create(&path).unwrap(),
+                Arc::clone(&schema),
+                Some(props),
+            )
+            .unwrap();
+            for group in 0..3i64 {
+                let nested = StructArray::new(
+                    nested_fields.clone().into(),
+                    vec![Arc::new(Int64Array::from_iter_values(
+                        (0..4096).map(|row| group * 4096 + row),
+                    ))],
+                    None,
+                );
+                let batch = RecordBatch::try_new(
+                    Arc::clone(&schema),
+                    vec![
+                        Arc::new(Int64Array::from_iter_values(
+                            (0..4096).map(|row| if clustered { row } else { row % 128 }),
+                        )),
+                        Arc::new(StringArray::from_iter_values(
+                            (0..4096).map(|row| format!("value-{}", row % 11)),
+                        )),
+                        Arc::new(nested),
+                    ],
+                )
+                .unwrap();
+                writer.write(&batch).unwrap();
+                writer.flush().unwrap();
+            }
+            let metadata = writer.close().unwrap();
+            assert_eq!(metadata.num_row_groups(), 3);
+            for group in metadata.row_groups() {
+                assert!(group.column(1).dictionary_page_offset().is_some());
+                assert_eq!(group.column(0).offset_index_offset().is_some(), indexed);
+            }
+            for predicate_in_output in [false, true] {
+                let mut expected_batch = None;
+                let mut baseline_bytes = None;
+                for (pushdown, progressive) in
+                    [(false, true), (true, true), (true, false)]
+                {
+                    let config = SessionConfig::new()
+                        .with_target_partitions(1)
+                        .with_batch_size(127);
+                    let ctx = SessionContext::new_with_config(config);
+                    for (key, value) in [
+                        ("pushdown_filters", pushdown),
+                        ("progressive_io", progressive),
+                    ] {
+                        ctx.sql(&format!(
+                            "SET datafusion.execution.parquet.{key} = {value}"
+                        ))
+                        .await
+                        .unwrap()
+                        .collect()
+                        .await
+                        .unwrap();
+                    }
+                    ctx.register_parquet(
+                        "t",
+                        path.to_str().unwrap(),
+                        ParquetReadOptions::default(),
+                    )
+                    .await
+                    .unwrap();
+                    let key = if predicate_in_output { "key," } else { "" };
+                    let plan = ctx.sql(&format!("SELECT {key} payload, nested.value AS value FROM t WHERE key < 100 ORDER BY value"))
+                    .await.unwrap().create_physical_plan().await.unwrap();
+                    let mut scans = 0;
+                    plan.apply(|plan| {
+                        if let Some(exec) = plan.downcast_ref::<DataSourceExec>()
+                            && let Some((_, source)) =
+                                exec.downcast_to_file_source::<ParquetSource>()
+                        {
+                            scans += 1;
+                            assert_eq!(
+                                source.table_parquet_options().global.progressive_io,
+                                progressive
+                            );
+                        }
+                        Ok(TreeNodeRecursion::Continue)
+                    })
+                    .unwrap();
+                    assert_eq!(scans, 1);
+                    let batches =
+                        collect(Arc::clone(&plan), ctx.task_ctx()).await.unwrap();
+                    let batch = concat_batches(&plan.schema(), &batches).unwrap();
+                    let values = batch
+                        .column(batch.num_columns() - 1)
+                        .as_any()
+                        .downcast_ref::<Int64Array>()
+                        .unwrap();
+                    assert_eq!(values.values().as_ref(), expected.as_slice());
+                    if let Some(expected) = &expected_batch {
+                        assert_eq!(&batch, expected);
+                    } else {
+                        expected_batch = Some(batch);
+                    }
+                    let metrics = TestParquetFile::parquet_metrics(&plan).unwrap();
+                    let bytes = get_value(&metrics, "bytes_scanned");
+                    if indexed && clustered {
+                        assert!(
+                            get_pruning_metrics(&metrics, "page_index_rows_pruned").0 > 0
+                        );
+                    }
+                    if !predicate_in_output {
+                        if let Some(expected) = baseline_bytes {
+                            assert_eq!(
+                                bytes, expected,
+                                "clustered={clustered}, indexed={indexed}, pushdown={pushdown}, progressive={progressive}"
+                            );
+                        } else {
+                            baseline_bytes = Some(bytes);
+                        }
+                    }
+                    if pushdown {
+                        assert!(get_value(&metrics, "pushdown_rows_pruned") > 0);
+                    }
+                }
+            }
+        }
+    }
+}

datafusion/datasource-parquet/src/file_format.rs

@@ -728,6 +728,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
                 parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64)
             }),
             pushdown_filters: global_options.global.pushdown_filters,
+            progressive_io: Some(global_options.global.progressive_io),
             reorder_filters: global_options.global.reorder_filters,
             force_filter_selections: global_options.global.force_filter_selections,
             data_pagesize_limit: global_options.global.data_pagesize_limit as u64,

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

@@ -236,6 +236,7 @@ fn validate_predicate_does_not_reference_virtual_columns(
 /// as an explicit state machine -- see [`ParquetOpenState`] for details.
 #[derive(Clone)]
 pub(super) struct ParquetMorselizer {
+    pub(crate) progressive_io: bool,
     /// Execution partition index
     pub(crate) partition_index: usize,
     /// Projection to apply on top of the table schema (i.e. can reference partition columns).
@@ -421,6 +422,7 @@ impl fmt::Debug for ParquetOpenState {
 }
 
 struct PreparedParquetOpen {
+    progressive_io: bool,
     partition_index: usize,
     partitioned_file: PartitionedFile,
     /// Tracks how much of this file range the scan has finished with.
@@ -854,6 +856,7 @@ impl ParquetMorselizer {
             metrics: self.metrics.clone(),
             parquet_file_reader_factory: Arc::clone(&self.parquet_file_reader_factory),
             async_file_reader,
+            progressive_io: self.progressive_io,
             batch_size: self.batch_size,
             logical_file_schema: Arc::clone(&logical_file_schema),
             physical_file_schema: logical_file_schema,
@@ -1471,6 +1474,7 @@ impl RowGroupsPrunedParquetOpen {
             prepared.virtual_state.as_deref(),
         )?;
 
+        let mut fetch_projection = decoder_projection.projection_mask().clone();
         // Lazily-registered suppression counter shared by the open-time first-RG
         // skip below and the stream's per-RG toggle (registered on first use so
         // scans that never suppress don't carry a zero-valued counter).
@@ -1522,6 +1526,20 @@ impl RowGroupsPrunedParquetOpen {
                 decoder_limit: prepared.limit,
             };
 
+            let mut fetch_selections = if !prepared.progressive_io {
+                access_plan
+                    .inner()
+                    .iter()
+                    .map(|access| match access {
+                        crate::RowGroupAccess::Selection(selection) => {
+                            Some(selection.clone())
+                        }
+                        _ => None,
+                    })
+                    .collect::<Vec<_>>()
+            } else {
+                vec![]
+            };
             let prepared_access_plan = prepare_access_plan(access_plan)?;
             // #24355: a row selection (from page-index pruning, or an externally
             // supplied `ParquetRowSelection`) is carried by the decoder as one
@@ -1548,6 +1566,9 @@ impl RowGroupsPrunedParquetOpen {
                 .zip(prepared_access_plan.fully_matched.iter().copied())
                 .map(|(rg_index, fully_matched)| RgPlanEntry {
                     rg_index,
+                    row_selection: fetch_selections
+                        .get_mut(rg_index)
+                        .and_then(Option::take),
                     fully_matched,
                     bytes: row_group_bytes(&rg_metadata[rg_index]),
                 })
@@ -1571,6 +1592,7 @@ impl RowGroupsPrunedParquetOpen {
                 decoder_config.build(prepared_access_plan, reader_metadata.clone());
             let mut filter_installed = false;
             if let Some(ctx) = row_filter_context.as_ref() {
+                ctx.extend_projection(&mut fetch_projection);
                 if first_rg_fully_matched {
                     // The first RG is fully matched: install an empty filter
                     // and count the suppression, exactly as the per-RG toggle
@@ -1679,6 +1701,10 @@ impl RowGroupsPrunedParquetOpen {
             active_reader: None,
             rg_plan,
             reader: prepared.async_file_reader,
+            progressive_io: prepared.progressive_io,
+            fetch_projection,
+            upfront_row_group: None,
+            parquet_metadata: Arc::clone(reader_metadata.metadata()),
             decoder_projection,
             arrow_reader_metrics,
             predicate_cache_inner_records,
@@ -2311,6 +2337,7 @@ mod test {
             )?;
 
             Ok(ParquetMorselizer {
+                progressive_io: true,
                 partition_index: self.partition_index,
                 projection,
                 batch_size: self.batch_size,
@@ -3367,6 +3394,8 @@ mod test {
 
         let props = WriterProperties::builder()
             .set_max_row_group_row_count(Some(4))
+            .set_data_page_row_count_limit(1)
+            .set_write_batch_size(1)
             .build();
 
         let data_len = write_parquet_batches(
@@ -3401,43 +3430,22 @@ mod test {
         )
         .with_extension(access_plan);
 
-        let make_opener = |reverse_scan: bool| {
-            ParquetMorselizerBuilder::new()
-                .with_store(Arc::clone(&store))
-                .with_schema(Arc::clone(&schema))
-                .with_projection_indices(&[0])
-                .with_reverse_row_groups(reverse_scan)
-                .build()
-        };
-
-        // Forward scan: RG0(3,4), RG1(5,6,7,8), RG2(9,10)
-        let opener = make_opener(false);
-        let stream = open_file(&opener, file.clone()).await.unwrap();
-        let forward_values = collect_int32_values(stream).await;
-
-        // Forward scan should produce: RG0(3,4), RG1(5,6,7,8), RG2(9,10)
-        assert_eq!(
-            forward_values,
-            vec![3, 4, 5, 6, 7, 8, 9, 10],
-            "Forward scan should select correct rows based on RowSelection"
-        );
-
-        // Reverse scan
-        // CORRECT behavior: reverse row groups AND their corresponding selections
-        // - RG2 is read first, WITH RG2's selection (select 2, skip 2) -> 9, 10
-        // - RG1 is read second, WITH RG1's selection (select all) -> 5, 6, 7, 8
-        // - RG0 is read third, WITH RG0's selection (skip 2, select 2) -> 3, 4
-        let opener = make_opener(true);
-        let stream = open_file(&opener, file).await.unwrap();
-        let reverse_values = collect_int32_values(stream).await;
-
-        // Correct expected result: row groups reversed but each keeps its own selection
-        // RG2 with its selection (9,10), RG1 with its selection (5,6,7,8), RG0 with its selection (3,4)
-        assert_eq!(
-            reverse_values,
-            vec![9, 10, 5, 6, 7, 8, 3, 4],
-            "Reverse scan should reverse row group order while maintaining correct RowSelection for each group"
-        );
+        for progressive_io in [false, true] {
+            for (reverse, expected) in [
+                (false, vec![3, 4, 5, 6, 7, 8, 9, 10]),
+                (true, vec![9, 10, 5, 6, 7, 8, 3, 4]),
+            ] {
+                let mut opener = ParquetMorselizerBuilder::new()
+                    .with_store(Arc::clone(&store))
+                    .with_schema(Arc::clone(&schema))
+                    .with_projection_indices(&[0])
+                    .with_reverse_row_groups(reverse)
+                    .build();
+                opener.progressive_io = progressive_io;
+                let stream = open_file(&opener, file.clone()).await.unwrap();
+                assert_eq!(collect_int32_values(stream).await, expected);
+            }
+        }
     }
 
     #[tokio::test]