WIP: DO NOT REVIEW: Benchmark enabling parquet filter pushdown by default with one-shot row group I/O

datafusion-examples/examples/data_io/json_shredding.rs

@@ -23,6 +23,7 @@ use arrow::array::{RecordBatch, StringArray};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 
 use datafusion::assert_batches_eq;
+use datafusion::common::config::ParquetPushdownFilterMode;
 use datafusion::common::tree_node::{
     Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
 };
@@ -92,6 +93,11 @@ pub async fn json_shredding() -> Result<()> {
     // Set up query execution
     let mut cfg = SessionConfig::new();
     cfg.options_mut().execution.parquet.pushdown_filters = true;
+    // This example needs the filter pushed into the scan so the JSON
+    // shredding rewriter can rewrite it into direct shredded-column
+    // access. Force pushdown regardless of the projection width.
+    cfg.options_mut().execution.parquet.pushdown_filter_mode =
+        ParquetPushdownFilterMode::Always;
     let ctx = SessionContext::new_with_config(cfg);
     ctx.runtime_env().register_object_store(
         ObjectStoreUrl::parse("memory://")?.as_ref(),

datafusion/common/src/config.rs

@@ -583,6 +583,72 @@ impl Display for SpillCompression {
     }
 }
 
+/// Strategy for filter pushdown in Parquet scan when
+/// `datafusion.execution.parquet.pushdown_filters`
+/// (*[`ParquetOptions::pushdown_filters`]) is enabled
+///
+/// Different strategies are better depending on how data is stored in the
+/// Parquet files and what rows predicates select (e.g. their selectivity and
+/// how many contiguous rows they select).
+///
+/// Note: This option has no effect unless `pushdown_filters` is also enabled.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum ParquetPushdownFilterMode {
+    /// Let DataFusion pick the best available strategy.
+    ///
+    /// Note: currently  identical to [`Self::Heuristic`], but as we implement
+    /// more sophisticated pushdown strategies (e.g. runtime-adaptive placement
+    /// in <https://github.com/apache/datafusion/issues/22883>), this may change.
+    #[default]
+    Auto,
+    /// Always push filters into the scan.
+    Always,
+    /// Use plan-time heuristics to decide which filters to push.
+    ///
+    /// The current heuristic skips pushdown when the projection contains fewer
+    /// than 3 non-filter columns, which avoid narrow-projection queries such as
+    /// `SELECT col2 FROM t WHERE col1 <> ''`, where `RowFilter` overhead
+    /// tends to dominate the decode it would save.
+    Heuristic,
+}
+
+impl FromStr for ParquetPushdownFilterMode {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_ascii_lowercase().as_str() {
+            "auto" | "" => Ok(Self::Auto),
+            "always" => Ok(Self::Always),
+            "heuristic" => Ok(Self::Heuristic),
+            other => Err(DataFusionError::Configuration(format!(
+                "Invalid pushdown filter mode: {other}. Expected one of: auto, always, heuristic"
+            ))),
+        }
+    }
+}
+
+impl ConfigField for ParquetPushdownFilterMode {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = ParquetPushdownFilterMode::from_str(value)?;
+        Ok(())
+    }
+}
+
+impl Display for ParquetPushdownFilterMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let str = match self {
+            Self::Auto => "auto",
+            Self::Always => "always",
+            Self::Heuristic => "heuristic",
+        };
+        write!(f, "{str}")
+    }
+}
+
 /// A `usize` configuration value that rejects zero when set from strings.
 ///
 /// Use this for options where zero is never a meaningful runtime value.
@@ -1228,7 +1294,12 @@ config_namespace! {
 
         /// (reading) If true, filter expressions are be applied during the parquet decoding operation to
         /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
-        pub pushdown_filters: bool, default = false
+        pub pushdown_filters: bool, default = true
+
+        /// (reading) When `pushdown_filters` is enabled, determines how DataFusion
+        /// pushes each filter into the Parquet scan. Options are `auto` (the default)
+        ///  `always`, and `heurstic` (plan time heuristic).
+        pub pushdown_filter_mode: ParquetPushdownFilterMode, default = ParquetPushdownFilterMode::Auto
 
         /// (reading) If true, filter expressions evaluated during the parquet decoding operation
         /// will be reordered heuristically to minimize the cost of evaluation. If false,
@@ -1241,6 +1312,19 @@ config_namespace! {
         /// pattern of selected rows.
         pub force_filter_selections: bool, default = false
 
+        /// (reading) Controls the I/O pattern used when `pushdown_filters` is
+        /// enabled. If false (the default), all data pages needed to read a row
+        /// group (for both filter evaluation and output projection) are fetched
+        /// in a single request, the same I/O pattern used when
+        /// `pushdown_filters` is disabled. If true, data is fetched
+        /// progressively: first the columns needed by each filter, then, after
+        /// the filters are evaluated, the remaining projected columns for the
+        /// rows that passed. Progressive fetching can reduce the total bytes
+        /// read when the file has a Parquet offset index, at the cost of
+        /// additional I/O requests per row group; files without an offset index
+        /// are always read with a single request per row group.
+        pub progressive_io: bool, default = false
+
         /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
         /// and `Binary/BinaryLarge` with `BinaryView`.
         pub schema_force_view_types: bool, default = true

datafusion/common/src/file_options/parquet_writer.rs

@@ -236,8 +236,10 @@ impl ParquetOptions {
             skip_metadata: _,
             metadata_size_hint: _,
             pushdown_filters: _,
+            pushdown_filter_mode: _, // reads-only, not used for writer props
             reorder_filters: _,
             force_filter_selections: _, // not used for writer props
+            progressive_io: _,          // not used for writer props
             allow_single_file_parallelism: _,
             maximum_parallel_row_group_writers: _,
             maximum_buffered_record_batches_per_stream: _,
@@ -494,8 +496,10 @@ mod tests {
             skip_metadata: defaults.skip_metadata,
             metadata_size_hint: defaults.metadata_size_hint,
             pushdown_filters: defaults.pushdown_filters,
+            pushdown_filter_mode: defaults.pushdown_filter_mode,
             reorder_filters: defaults.reorder_filters,
             force_filter_selections: defaults.force_filter_selections,
+            progressive_io: defaults.progressive_io,
             allow_single_file_parallelism: defaults.allow_single_file_parallelism,
             maximum_parallel_row_group_writers: defaults
                 .maximum_parallel_row_group_writers,
@@ -614,8 +618,10 @@ mod tests {
                 skip_metadata: global_options_defaults.skip_metadata,
                 metadata_size_hint: global_options_defaults.metadata_size_hint,
                 pushdown_filters: global_options_defaults.pushdown_filters,
+                pushdown_filter_mode: global_options_defaults.pushdown_filter_mode,
                 reorder_filters: global_options_defaults.reorder_filters,
                 force_filter_selections: global_options_defaults.force_filter_selections,
+                progressive_io: global_options_defaults.progressive_io,
                 allow_single_file_parallelism: global_options_defaults
                     .allow_single_file_parallelism,
                 maximum_parallel_row_group_writers: global_options_defaults

datafusion/core/tests/datasource/object_store_access.rs

@@ -920,6 +920,85 @@ async fn query_single_parquet_file_multi_row_groups_multiple_predicates() {
     );
 }
 
+#[tokio::test]
+async fn query_single_parquet_file_pushdown_filters_one_shot_io() {
+    let test = Test::new().with_single_file_parquet().await;
+    test.set("datafusion.execution.parquet.pushdown_filters", "true")
+        .await;
+    // Opt out of the narrow-projection pushdown gate: these tests exist to
+    // demonstrate the pushdown I/O pattern, which requires a RowFilter to
+    // actually be installed.
+    test.set(
+        "datafusion.execution.parquet.pushdown_filter_mode",
+        "always",
+    )
+    .await;
+
+    // With filter pushdown enabled and `progressive_io` disabled (the
+    // default), all data needed for each row group (columns needed by the
+    // filter and the projection) is fetched with a single request per row
+    // group: the same I/O pattern as when pushdown_filters is disabled.
+    assert_snapshot!(
+        test.query("select min(b) from parquet_table WHERE a > 50").await,
+        @r"
+    ------- Query Output (1 rows) -------
+    +----------------------+
+    | min(parquet_table.b) |
+    +----------------------+
+    | 1051                 |
+    +----------------------+
+    ------- Object Store Request Summary -------
+    RequestCountingObjectStore()
+    Total Requests: 3
+    - GET  (opts) path=parquet_table.parquet head=true
+    - GET  (ranges) path=parquet_table.parquet ranges=4-534,534-1064
+    - GET  (ranges) path=parquet_table.parquet ranges=1064-1594,1594-2124
+    "
+    );
+}
+
+#[tokio::test]
+async fn query_single_parquet_file_pushdown_filters_progressive_io() {
+    let test = Test::new().with_single_file_parquet().await;
+    test.set("datafusion.execution.parquet.pushdown_filters", "true")
+        .await;
+    // Opt out of the narrow-projection pushdown gate: these tests exist to
+    // demonstrate the pushdown I/O pattern, which requires a RowFilter to
+    // actually be installed.
+    test.set(
+        "datafusion.execution.parquet.pushdown_filter_mode",
+        "always",
+    )
+    .await;
+    test.set("datafusion.execution.parquet.progressive_io", "true")
+        .await;
+
+    // With `progressive_io` enabled (and an offset index present in the
+    // file), each row group is fetched progressively: first the columns
+    // needed to evaluate the filter (`a`), then the remaining projected
+    // columns (`b`) for the rows that passed, resulting in multiple
+    // requests per row group.
+    assert_snapshot!(
+        test.query("select min(b) from parquet_table WHERE a > 50").await,
+        @r"
+    ------- Query Output (1 rows) -------
+    +----------------------+
+    | min(parquet_table.b) |
+    +----------------------+
+    | 1051                 |
+    +----------------------+
+    ------- Object Store Request Summary -------
+    RequestCountingObjectStore()
+    Total Requests: 5
+    - GET  (opts) path=parquet_table.parquet head=true
+    - GET  (ranges) path=parquet_table.parquet ranges=4-534
+    - GET  (ranges) path=parquet_table.parquet ranges=534-951,951-1064
+    - GET  (ranges) path=parquet_table.parquet ranges=1064-1594
+    - GET  (ranges) path=parquet_table.parquet ranges=1594-2124
+    "
+    );
+}
+
 /// Runs tests with a request counting object store
 struct Test {
     object_store: Arc<RequestCountingObjectStore>,
@@ -957,6 +1036,17 @@ impl Test {
         format!("{}", self.object_store)
     }
 
+    /// Set a session configuration value via SQL `SET`
+    async fn set(&self, key: &str, value: &str) {
+        self.session_context
+            .sql(&format!("set {key} = {value}"))
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+    }
+
     /// Store the specified bytes at the given path
     async fn with_bytes(self, path: &str, bytes: impl Into<Bytes>) -> Self {
         let path = Path::from(path);

datafusion/core/tests/parquet/mod.rs

@@ -37,6 +37,7 @@ use datafusion::{
     physical_plan::metrics::MetricsSet,
     prelude::{ParquetReadOptions, SessionConfig, SessionContext},
 };
+use datafusion_common::config::ParquetPushdownFilterMode;
 use datafusion_expr::{Expr, LogicalPlan, LogicalPlanBuilder};
 use datafusion_physical_plan::metrics::MetricValue;
 use parquet::arrow::ArrowWriter;
@@ -323,6 +324,10 @@ impl ContextWithParquet {
             Unit::RowGroup(row_per_group) => {
                 config = config.with_parquet_bloom_filter_pruning(true);
                 config.options_mut().execution.parquet.pushdown_filters = true;
+                // force unconditional pushdown to test so the filters are
+                // applied for TopK dynamic RG pruning
+                config.options_mut().execution.parquet.pushdown_filter_mode =
+                    ParquetPushdownFilterMode::Always;
                 make_test_file_rg(
                     scenario,
                     row_per_group,
@@ -339,6 +344,10 @@ impl ContextWithParquet {
                 config = config.with_parquet_bloom_filter_pruning(true);
                 config = config.with_parquet_page_index_pruning(true);
                 config.options_mut().execution.parquet.pushdown_filters = true;
+                // force unconditional pushdown to test so the filters are
+                // applied for TopK dynamic RG pruning
+                config.options_mut().execution.parquet.pushdown_filter_mode =
+                    ParquetPushdownFilterMode::Always;
                 make_test_file_rg(
                     scenario,
                     row_per_group,

datafusion/datasource-parquet/src/file_format.rs

@@ -716,8 +716,21 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
                 parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64)
             }),
             pushdown_filters: global_options.global.pushdown_filters,
+            pushdown_filter_mode: match global_options.global.pushdown_filter_mode {
+                datafusion_common::config::ParquetPushdownFilterMode::Auto => {
+                    parquet_options::PushdownFilterMode::Auto
+                }
+                datafusion_common::config::ParquetPushdownFilterMode::Always => {
+                    parquet_options::PushdownFilterMode::Always
+                }
+                datafusion_common::config::ParquetPushdownFilterMode::Heuristic => {
+                    parquet_options::PushdownFilterMode::Heuristic
+                }
+            }
+            .into(),
             reorder_filters: global_options.global.reorder_filters,
             force_filter_selections: global_options.global.force_filter_selections,
+            progressive_io: global_options.global.progressive_io,
             data_pagesize_limit: global_options.global.data_pagesize_limit as u64,
             write_batch_size: global_options.global.write_batch_size as u64,
             writer_version: global_options.global.writer_version.to_string(),