From 3064ea932be473ed1f811b96860867541142cd9b Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 4 Oct 2023 09:28:26 -0400
Subject: [PATCH 1/4] Minor: Improve PruningPredicate documentation

---
 .../core/src/physical_optimizer/pruning.rs    | 119 +++++++++++-------
 1 file changed, 71 insertions(+), 48 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/pruning.rs b/datafusion/core/src/physical_optimizer/pruning.rs
index 2987ec6d6552a..98f4dd617a8e1 100644
--- a/datafusion/core/src/physical_optimizer/pruning.rs
+++ b/datafusion/core/src/physical_optimizer/pruning.rs
@@ -15,19 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! This module contains code to prune "containers" of row groups
-//! based on statistics prior to execution. This can lead to
-//! significant performance improvements by avoiding the need
-//! to evaluate a plan on entire containers (e.g. an entire file)
-//!
-//! For example, DataFusion uses this code to prune (skip) row groups
-//! while reading parquet files if it can be determined from the
-//! predicate that nothing in the row group can match.
-//!
-//! This code can also be used by other systems to prune other
-//! entities (e.g. entire files) if the statistics are known via some
-//! other source (e.g. a catalog)
-
+//! [`PruningPredicate`] to apply filter [`Expr`] to prune "containers"
+//! based on statistics (e.g. Parquet Row Groups)
 use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::sync::Arc;
@@ -53,18 +42,21 @@ use datafusion_physical_expr::utils::collect_columns;
 use datafusion_physical_expr::{expressions as phys_expr, PhysicalExprRef};
 use log::trace;
 
-/// Interface to pass statistics information to [`PruningPredicate`]
+/// Interface to pass statistics (min/max/nulls) information to [`PruningPredicate`].
 ///
-/// Returns statistics for containers / files of data in Arrays.
+/// Returns statistics for containers / files as Arrow [`ArrayRefs`], so the
+/// evaluation happens once on a single `RecordBatch`, amortizing the overhead
+/// of evaluating of the predicate. This is important when pruning 1000s of
+/// containers which often happens in analytic systems.
 ///
-/// For example, for the following three files with a single column
+/// For example, for the following three files with a single column `a`:
 /// ```text
 /// file1: column a: min=5, max=10
 /// file2: column a: No stats
 /// file2: column a: min=20, max=30
 /// ```
 ///
-/// PruningStatistics should return:
+/// PruningStatistics would return:
 ///
 /// ```text
 /// min_values("a") -> Some([5, Null, 20])
@@ -91,10 +83,44 @@ pub trait PruningStatistics {
     fn null_counts(&self, column: &Column) -> Option<ArrayRef>;
 }
 
-/// Evaluates filter expressions on statistics in order to
-/// prune data containers (e.g. parquet row group)
+/// Evaluates filter expressions on statistics, rather than the actual data. If
+/// no rows could possibly pass the filter entire containers can be "pruned"
+/// (skipped), without reading any actual data, leading to significant
+/// performance improvements.
+///
+/// [`PruningPredicate`]s are used to prune (avoid scanning) Parquet Row Groups
+/// based on the min/max values found in the Parquet metadata. If the
+/// `PruningPredicate` can guarantee that no rows in the Row Group match the
+/// filter, the entire Row Group is skipped during query execution.
+///
+/// Note that this API is designed to be general, as it works:
+///
+/// 1. Arbitrary expressions expressions (including user defined functions)
+///
+/// 2. Anything that implements the [`PruningStatistics`] interface, not just
+/// Parquet metadata, allowing it to be used by other systems to prune entities
+/// (e.g. entire files) if the statistics are known via some other source, such
+/// as a catalog.
+///
+/// # Example
+///
+/// Given an expression like `x = 5` and statistics for 3 containers (Row
+/// Groups, files, etc) `A`, `B`, and `C`:
+///
+/// ```text
+///   A: {x_min = 0, x_max = 4}
+///   B: {x_min = 2, x_max = 10}
+///   C: {x_min = 5, x_max = 8}
+/// ```
 ///
-/// See [`PruningPredicate::try_new`] for more information.
+/// Applying the `PruningPredicate` will concludes that `A` can be pruned:
+///
+/// ```text
+/// A: false (no rows could possibly match x = 5)
+/// B: true  (rows might match x = 5)
+/// C: true  (rows might match x = 5)
+/// ```
+/// See [`PruningPredicate::try_new`] and [`PruningPredicate::prune`] for more information.
 #[derive(Debug, Clone)]
 pub struct PruningPredicate {
     /// The input schema against which the predicate will be evaluated
@@ -146,17 +172,14 @@ impl PruningPredicate {
     ///
     /// `true`: There MAY be rows that match the predicate
     ///
-    /// `false`: There are no rows that could match the predicate
+    /// `false`: There are no rows that could possible match the predicate
     ///
-    /// Note this function takes a slice of statistics as a parameter
-    /// to amortize the cost of the evaluation of the predicate
-    /// against a single record batch.
-    ///
-    /// Note: the predicate passed to `prune` should be simplified as
+    /// Note: the predicate passed to `prune` should already be simplified as
     /// much as possible (e.g. this pass doesn't handle some
     /// expressions like `b = false`, but it does handle the
-    /// simplified version `b`. The predicates are simplified via the
-    /// ConstantFolding optimizer pass
+    /// simplified version `b`. See [`ExprSimplifier`] to simplify expressions.
+    ///
+    /// [`ExprSimplifier`]: datafusion::optimizer::simplify_expressions::ExprSimplifier
     pub fn prune<S: PruningStatistics>(&self, statistics: &S) -> Result<Vec<bool>> {
         // build a RecordBatch that contains the min/max values in the
         // appropriate statistics columns
@@ -258,7 +281,7 @@ impl RequiredStatColumns {
     /// `self.columns` for details)
     pub(crate) fn iter(
         &self,
-    ) -> impl Iterator<Item = &(phys_expr::Column, StatisticsType, Field)> {
+    ) -> impl Iterator<Item=&(phys_expr::Column, StatisticsType, Field)> {
         self.columns.iter()
     }
 
@@ -807,10 +830,10 @@ fn build_predicate_expression(
             (left, Operator::And, _) if is_always_true(left) => right_expr,
             (_, Operator::And, right) if is_always_true(right) => left_expr,
             (left, Operator::Or, right)
-                if is_always_true(left) || is_always_true(right) =>
-            {
-                unhandled
-            }
+            if is_always_true(left) || is_always_true(right) =>
+                {
+                    unhandled
+                }
             _ => Arc::new(phys_expr::BinaryExpr::new(left_expr, op, right_expr)),
         };
         return expr;
@@ -909,7 +932,7 @@ fn build_statistics_expr(
         _ => {
             return plan_err!(
                 "expressions other than (neq, eq, gt, gteq, lt, lteq) are not supported"
-            )
+            );
         }
     };
     Ok(statistics_expr)
@@ -957,8 +980,8 @@ mod tests {
 
     impl ContainerStats {
         fn new_decimal128(
-            min: impl IntoIterator<Item = Option<i128>>,
-            max: impl IntoIterator<Item = Option<i128>>,
+            min: impl IntoIterator<Item=Option<i128>>,
+            max: impl IntoIterator<Item=Option<i128>>,
             precision: u8,
             scale: i8,
         ) -> Self {
@@ -980,8 +1003,8 @@ mod tests {
         }
 
         fn new_i64(
-            min: impl IntoIterator<Item = Option<i64>>,
-            max: impl IntoIterator<Item = Option<i64>>,
+            min: impl IntoIterator<Item=Option<i64>>,
+            max: impl IntoIterator<Item=Option<i64>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<Int64Array>()),
@@ -991,8 +1014,8 @@ mod tests {
         }
 
         fn new_i32(
-            min: impl IntoIterator<Item = Option<i32>>,
-            max: impl IntoIterator<Item = Option<i32>>,
+            min: impl IntoIterator<Item=Option<i32>>,
+            max: impl IntoIterator<Item=Option<i32>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<Int32Array>()),
@@ -1002,8 +1025,8 @@ mod tests {
         }
 
         fn new_utf8<'a>(
-            min: impl IntoIterator<Item = Option<&'a str>>,
-            max: impl IntoIterator<Item = Option<&'a str>>,
+            min: impl IntoIterator<Item=Option<&'a str>>,
+            max: impl IntoIterator<Item=Option<&'a str>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<StringArray>()),
@@ -1013,8 +1036,8 @@ mod tests {
         }
 
         fn new_bool(
-            min: impl IntoIterator<Item = Option<bool>>,
-            max: impl IntoIterator<Item = Option<bool>>,
+            min: impl IntoIterator<Item=Option<bool>>,
+            max: impl IntoIterator<Item=Option<bool>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<BooleanArray>()),
@@ -1044,7 +1067,7 @@ mod tests {
         /// there are containers
         fn with_null_counts(
             mut self,
-            counts: impl IntoIterator<Item = Option<i64>>,
+            counts: impl IntoIterator<Item=Option<i64>>,
         ) -> Self {
             // take stats out and update them
             let null_counts: ArrayRef =
@@ -1083,7 +1106,7 @@ mod tests {
         fn with_null_counts(
             mut self,
             name: impl Into<String>,
-            counts: impl IntoIterator<Item = Option<i64>>,
+            counts: impl IntoIterator<Item=Option<i64>>,
         ) -> Self {
             let col = Column::from_name(name.into());
 
@@ -2378,7 +2401,7 @@ mod tests {
             &right_input,
             df_schema.clone(),
         )
-        .unwrap();
+            .unwrap();
         assert_eq!(result_left.to_string(), left_input.to_string());
         assert_eq!(result_right.to_string(), right_input.to_string());
 
@@ -2393,7 +2416,7 @@ mod tests {
             &right_input,
             df_schema.clone(),
         )
-        .unwrap();
+            .unwrap();
         assert_eq!(result_left.to_string(), left_input.to_string());
         assert_eq!(result_right.to_string(), right_input.to_string());
 

From 674bf21d1168079cb9a155462e088b7129840881 Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 4 Oct 2023 09:30:23 -0400
Subject: [PATCH 2/4] fmt

---
 .../core/src/physical_optimizer/pruning.rs    | 38 +++++++++----------
 1 file changed, 19 insertions(+), 19 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/pruning.rs b/datafusion/core/src/physical_optimizer/pruning.rs
index 98f4dd617a8e1..20e37e1271e15 100644
--- a/datafusion/core/src/physical_optimizer/pruning.rs
+++ b/datafusion/core/src/physical_optimizer/pruning.rs
@@ -281,7 +281,7 @@ impl RequiredStatColumns {
     /// `self.columns` for details)
     pub(crate) fn iter(
         &self,
-    ) -> impl Iterator<Item=&(phys_expr::Column, StatisticsType, Field)> {
+    ) -> impl Iterator<Item = &(phys_expr::Column, StatisticsType, Field)> {
         self.columns.iter()
     }
 
@@ -830,10 +830,10 @@ fn build_predicate_expression(
             (left, Operator::And, _) if is_always_true(left) => right_expr,
             (_, Operator::And, right) if is_always_true(right) => left_expr,
             (left, Operator::Or, right)
-            if is_always_true(left) || is_always_true(right) =>
-                {
-                    unhandled
-                }
+                if is_always_true(left) || is_always_true(right) =>
+            {
+                unhandled
+            }
             _ => Arc::new(phys_expr::BinaryExpr::new(left_expr, op, right_expr)),
         };
         return expr;
@@ -980,8 +980,8 @@ mod tests {
 
     impl ContainerStats {
         fn new_decimal128(
-            min: impl IntoIterator<Item=Option<i128>>,
-            max: impl IntoIterator<Item=Option<i128>>,
+            min: impl IntoIterator<Item = Option<i128>>,
+            max: impl IntoIterator<Item = Option<i128>>,
             precision: u8,
             scale: i8,
         ) -> Self {
@@ -1003,8 +1003,8 @@ mod tests {
         }
 
         fn new_i64(
-            min: impl IntoIterator<Item=Option<i64>>,
-            max: impl IntoIterator<Item=Option<i64>>,
+            min: impl IntoIterator<Item = Option<i64>>,
+            max: impl IntoIterator<Item = Option<i64>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<Int64Array>()),
@@ -1014,8 +1014,8 @@ mod tests {
         }
 
         fn new_i32(
-            min: impl IntoIterator<Item=Option<i32>>,
-            max: impl IntoIterator<Item=Option<i32>>,
+            min: impl IntoIterator<Item = Option<i32>>,
+            max: impl IntoIterator<Item = Option<i32>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<Int32Array>()),
@@ -1025,8 +1025,8 @@ mod tests {
         }
 
         fn new_utf8<'a>(
-            min: impl IntoIterator<Item=Option<&'a str>>,
-            max: impl IntoIterator<Item=Option<&'a str>>,
+            min: impl IntoIterator<Item = Option<&'a str>>,
+            max: impl IntoIterator<Item = Option<&'a str>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<StringArray>()),
@@ -1036,8 +1036,8 @@ mod tests {
         }
 
         fn new_bool(
-            min: impl IntoIterator<Item=Option<bool>>,
-            max: impl IntoIterator<Item=Option<bool>>,
+            min: impl IntoIterator<Item = Option<bool>>,
+            max: impl IntoIterator<Item = Option<bool>>,
         ) -> Self {
             Self {
                 min: Arc::new(min.into_iter().collect::<BooleanArray>()),
@@ -1067,7 +1067,7 @@ mod tests {
         /// there are containers
         fn with_null_counts(
             mut self,
-            counts: impl IntoIterator<Item=Option<i64>>,
+            counts: impl IntoIterator<Item = Option<i64>>,
         ) -> Self {
             // take stats out and update them
             let null_counts: ArrayRef =
@@ -1106,7 +1106,7 @@ mod tests {
         fn with_null_counts(
             mut self,
             name: impl Into<String>,
-            counts: impl IntoIterator<Item=Option<i64>>,
+            counts: impl IntoIterator<Item = Option<i64>>,
         ) -> Self {
             let col = Column::from_name(name.into());
 
@@ -2401,7 +2401,7 @@ mod tests {
             &right_input,
             df_schema.clone(),
         )
-            .unwrap();
+        .unwrap();
         assert_eq!(result_left.to_string(), left_input.to_string());
         assert_eq!(result_right.to_string(), right_input.to_string());
 
@@ -2416,7 +2416,7 @@ mod tests {
             &right_input,
             df_schema.clone(),
         )
-            .unwrap();
+        .unwrap();
         assert_eq!(result_left.to_string(), left_input.to_string());
         assert_eq!(result_right.to_string(), right_input.to_string());
 

From b8e86f89368e9e77e1b932f59cab499e97916179 Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 4 Oct 2023 10:13:54 -0400
Subject: [PATCH 3/4] Apply suggestions from code review

---
 datafusion/core/src/physical_optimizer/pruning.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/pruning.rs b/datafusion/core/src/physical_optimizer/pruning.rs
index 20e37e1271e15..b5f07d941a40f 100644
--- a/datafusion/core/src/physical_optimizer/pruning.rs
+++ b/datafusion/core/src/physical_optimizer/pruning.rs
@@ -97,7 +97,7 @@ pub trait PruningStatistics {
 ///
 /// 1. Arbitrary expressions expressions (including user defined functions)
 ///
-/// 2. Anything that implements the [`PruningStatistics`] interface, not just
+/// 2. Anything that implements the [`PruningStatistics`] trait, not just
 /// Parquet metadata, allowing it to be used by other systems to prune entities
 /// (e.g. entire files) if the statistics are known via some other source, such
 /// as a catalog.
@@ -172,7 +172,7 @@ impl PruningPredicate {
     ///
     /// `true`: There MAY be rows that match the predicate
     ///
-    /// `false`: There are no rows that could possible match the predicate
+    /// `false`: There are no rows that could possibly match the predicate
     ///
     /// Note: the predicate passed to `prune` should already be simplified as
     /// much as possible (e.g. this pass doesn't handle some

From c5632936b008c66e5d9b3602ed8096f0d38b7757 Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 4 Oct 2023 11:29:01 -0400
Subject: [PATCH 4/4] fix doc links

---
 datafusion/core/src/physical_optimizer/pruning.rs | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/pruning.rs b/datafusion/core/src/physical_optimizer/pruning.rs
index 20e37e1271e15..5afcc9c63ea58 100644
--- a/datafusion/core/src/physical_optimizer/pruning.rs
+++ b/datafusion/core/src/physical_optimizer/pruning.rs
@@ -17,6 +17,8 @@
 
 //! [`PruningPredicate`] to apply filter [`Expr`] to prune "containers"
 //! based on statistics (e.g. Parquet Row Groups)
+//!
+//! [`Expr`]: crate::prelude::Expr
 use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::sync::Arc;
@@ -44,7 +46,7 @@ use log::trace;
 
 /// Interface to pass statistics (min/max/nulls) information to [`PruningPredicate`].
 ///
-/// Returns statistics for containers / files as Arrow [`ArrayRefs`], so the
+/// Returns statistics for containers / files as Arrow [`ArrayRef`], so the
 /// evaluation happens once on a single `RecordBatch`, amortizing the overhead
 /// of evaluating of the predicate. This is important when pruning 1000s of
 /// containers which often happens in analytic systems.
@@ -179,7 +181,7 @@ impl PruningPredicate {
     /// expressions like `b = false`, but it does handle the
     /// simplified version `b`. See [`ExprSimplifier`] to simplify expressions.
     ///
-    /// [`ExprSimplifier`]: datafusion::optimizer::simplify_expressions::ExprSimplifier
+    /// [`ExprSimplifier`]: crate::optimizer::simplify_expressions::ExprSimplifier
     pub fn prune<S: PruningStatistics>(&self, statistics: &S) -> Result<Vec<bool>> {
         // build a RecordBatch that contains the min/max values in the
         // appropriate statistics columns
