pipe column orderings into pruning predicate creation

datafusion-examples/examples/advanced_parquet_index.rs

@@ -46,7 +46,7 @@ use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties}
 use datafusion::parquet::schema::types::ColumnPath;
 use datafusion::physical_expr::utils::{Guarantee, LiteralGuarantee};
 use datafusion::physical_expr::PhysicalExpr;
-use datafusion::physical_optimizer::pruning::PruningPredicate;
+use datafusion::physical_optimizer::pruning::{ColumnOrdering, PruningPredicate};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::*;
@@ -300,8 +300,11 @@ impl IndexTableProvider {
         // In this example, we use the PruningPredicate's literal guarantees to
         // analyze the predicate. In a real system, using
         // `PruningPredicate::prune` would likely be easier to do.
-        let pruning_predicate =
-            PruningPredicate::try_new(Arc::clone(predicate), self.schema())?;
+        let pruning_predicate = PruningPredicate::try_new(
+            Arc::clone(predicate),
+            self.schema(),
+            vec![ColumnOrdering::Unknown; self.schema().fields().len()],
+        )?;
 
         // The PruningPredicate's guarantees must all be satisfied in order for
         // the predicate to possibly evaluate to true.

datafusion-examples/examples/parquet_index.rs

@@ -40,7 +40,7 @@ use datafusion::parquet::arrow::{
     arrow_reader::ParquetRecordBatchReaderBuilder, ArrowWriter,
 };
 use datafusion::physical_expr::PhysicalExpr;
-use datafusion::physical_optimizer::pruning::PruningPredicate;
+use datafusion::physical_optimizer::pruning::{ColumnOrdering, PruningPredicate};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::*;
 use std::any::Any;
@@ -366,8 +366,11 @@ impl ParquetMetadataIndex {
     ) -> Result<Vec<(&str, u64)>> {
         // Use the PruningPredicate API to determine which files can not
         // possibly have any relevant data.
-        let pruning_predicate =
-            PruningPredicate::try_new(predicate, self.schema().clone())?;
+        let pruning_predicate = PruningPredicate::try_new(
+            predicate,
+            self.schema().clone(),
+            vec![ColumnOrdering::Unknown; self.schema().fields().len()],
+        )?;
 
         // Now evaluate the pruning predicate into a boolean mask, one element per
         // file in the index. If the mask is true, the file may have rows that

datafusion-examples/examples/pruning.rs

@@ -24,7 +24,7 @@ use datafusion::common::pruning::PruningStatistics;
 use datafusion::common::{DFSchema, ScalarValue};
 use datafusion::execution::context::ExecutionProps;
 use datafusion::physical_expr::create_physical_expr;
-use datafusion::physical_optimizer::pruning::PruningPredicate;
+use datafusion::physical_optimizer::pruning::{ColumnOrdering, PruningPredicate};
 use datafusion::prelude::*;
 
 /// This example shows how to use  DataFusion's `PruningPredicate` to prove
@@ -190,7 +190,12 @@ fn create_pruning_predicate(expr: Expr, schema: &SchemaRef) -> PruningPredicate
     let df_schema = DFSchema::try_from(schema.as_ref().clone()).unwrap();
     let props = ExecutionProps::new();
     let physical_expr = create_physical_expr(&expr, &df_schema, &props).unwrap();
-    PruningPredicate::try_new(physical_expr, schema.clone()).unwrap()
+    PruningPredicate::try_new(
+        physical_expr,
+        schema.clone(),
+        vec![ColumnOrdering::Unknown; schema.fields().len()],
+    )
+    .unwrap()
 }
 
 fn i32_array<'a>(values: impl Iterator<Item = &'a Option<i32>>) -> ArrayRef {

datafusion/core/tests/parquet/page_pruning.rs

@@ -661,11 +661,15 @@ uint_tests!(64);
 // page-2                         0  0.0                                       4.0
 // page-3                         0  5.0                                       9.0
 async fn prune_f64_lt() {
+    // TODO: because NaN could be present but not accounted for in the statistics, these
+    // expressions should not be pruned at present. When IEEE 754 total order is added to
+    // the Parquet spec this can be revisited.
+    // See https://github.com/apache/parquet-format/pull/221
     test_prune(
         Scenario::Float64,
         "SELECT * FROM t where f < 1",
         Some(0),
-        Some(5),
+        Some(0),
         11,
         5,
     )
@@ -674,7 +678,7 @@ async fn prune_f64_lt() {
         Scenario::Float64,
         "SELECT * FROM t where -f > -1",
         Some(0),
-        Some(5),
+        Some(0),
         11,
         5,
     )
@@ -683,13 +687,17 @@ async fn prune_f64_lt() {
 
 #[tokio::test]
 async fn prune_f64_scalar_fun_and_gt() {
+    // TODO: because NaN could be present but not accounted for in the statistics, this
+    // expression should not be pruned at present. When IEEE 754 total order is added to
+    // the Parquet spec this can be revisited.
+    // See https://github.com/apache/parquet-format/pull/221
     // result of sql "SELECT * FROM t where abs(f - 1) <= 0.000001  and f >= 0.1"
     // only use "f >= 0" to prune
     test_prune(
         Scenario::Float64,
         "SELECT * FROM t where abs(f - 1) <= 0.000001  and f >= 0.1",
         Some(0),
-        Some(10),
+        Some(0),
         1,
         5,
     )

datafusion/core/tests/parquet/row_group_pruning.rs

@@ -635,12 +635,16 @@ async fn prune_uint32_eq_large_in_list() {
 
 #[tokio::test]
 async fn prune_f64_lt() {
+    // TODO: because NaN could be present but not accounted for in the statistics, these
+    // expressions should not be pruned at present. When IEEE 754 total order is added to
+    // the Parquet spec this can be revisited.
+    // See https://github.com/apache/parquet-format/pull/221
     RowGroupPruningTest::new()
         .with_scenario(Scenario::Float64)
         .with_query("SELECT * FROM t where f < 1")
         .with_expected_errors(Some(0))
-        .with_matched_by_stats(Some(3))
-        .with_pruned_by_stats(Some(1))
+        .with_matched_by_stats(Some(0))
+        .with_pruned_by_stats(Some(0))
         .with_matched_by_bloom_filter(Some(0))
         .with_pruned_by_bloom_filter(Some(0))
         .with_expected_rows(11)
@@ -650,8 +654,8 @@ async fn prune_f64_lt() {
         .with_scenario(Scenario::Float64)
         .with_query("SELECT * FROM t where -f > -1")
         .with_expected_errors(Some(0))
-        .with_matched_by_stats(Some(3))
-        .with_pruned_by_stats(Some(1))
+        .with_matched_by_stats(Some(0))
+        .with_pruned_by_stats(Some(0))
         .with_matched_by_bloom_filter(Some(0))
         .with_pruned_by_bloom_filter(Some(0))
         .with_expected_rows(11)
@@ -661,14 +665,19 @@ async fn prune_f64_lt() {
 
 #[tokio::test]
 async fn prune_f64_scalar_fun_and_gt() {
+    // TODO: because NaN could be present but not accounted for in the statistics, this
+    // expression should not be pruned at present. When IEEE 754 total order is added to
+    // the Parquet spec this can be revisited.
+    // See https://github.com/apache/parquet-format/pull/221
+
     // result of sql "SELECT * FROM t where abs(f - 1) <= 0.000001  and f >= 0.1"
     // only use "f >= 0" to prune
     RowGroupPruningTest::new()
         .with_scenario(Scenario::Float64)
         .with_query("SELECT * FROM t where abs(f - 1) <= 0.000001  and f >= 0.1")
         .with_expected_errors(Some(0))
-        .with_matched_by_stats(Some(2))
-        .with_pruned_by_stats(Some(2))
+        .with_matched_by_stats(Some(0))
+        .with_pruned_by_stats(Some(0))
         .with_matched_by_bloom_filter(Some(0))
         .with_pruned_by_bloom_filter(Some(0))
         .with_expected_rows(1)

datafusion/datasource-parquet/src/opener.rs

@@ -33,14 +33,15 @@ use arrow::datatypes::{SchemaRef, TimeUnit};
 use arrow::error::ArrowError;
 use datafusion_common::{exec_err, Result};
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
-use datafusion_physical_optimizer::pruning::PruningPredicate;
+use datafusion_physical_optimizer::pruning::{ColumnOrdering, PruningPredicate};
 use datafusion_physical_plan::metrics::{Count, ExecutionPlanMetricsSet, MetricBuilder};
 
 use futures::{StreamExt, TryStreamExt};
 use log::debug;
 use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
+use parquet::basic::{ColumnOrder, SortOrder};
 use parquet::file::metadata::ParquetMetaDataReader;
 
 /// Implements [`FileOpener`] for a parquet file
@@ -179,10 +180,42 @@ impl FileOpener for ParquetOpener {
                 }
             }
 
+            // Map column ordering from physical schema to logical
+            let ordering: Vec<ColumnOrdering> = if let Some(column_orders) =
+                reader_metadata.metadata().file_metadata().column_orders()
+            {
+                logical_file_schema
+                    .fields()
+                    .iter()
+                    .map(|field| {
+                        match physical_file_schema.index_of(field.name()) {
+                            Ok(idx) => match column_orders[idx] {
+                                ColumnOrder::TYPE_DEFINED_ORDER(sort_order) => {
+                                    match sort_order {
+                                        SortOrder::SIGNED => ColumnOrdering::Signed,
+                                        SortOrder::UNSIGNED => ColumnOrdering::Unsigned,
+                                        _ => ColumnOrdering::Undefined,
+                                    }
+                                }
+                                /* TODO(ets): for future
+                                ColumnOrder::IEEE_754_TOTAL_ORDER => {
+                                    ColumnOrdering::TotalOrder
+                                }*/
+                                ColumnOrder::UNDEFINED => ColumnOrdering::Unknown,
+                            },
+                            _ => ColumnOrdering::Unknown,
+                        }
+                    })
+                    .collect::<Vec<_>>()
+            } else {
+                vec![ColumnOrdering::Unknown; logical_file_schema.fields().len()]
+            };
+
             // Build predicates for this specific file
             let (pruning_predicate, page_pruning_predicate) = build_pruning_predicates(
                 predicate.as_ref(),
                 &logical_file_schema,
+                ordering,
                 &predicate_creation_errors,
             );
 
@@ -363,9 +396,11 @@ fn create_initial_plan(
 pub(crate) fn build_pruning_predicate(
     predicate: Arc<dyn PhysicalExpr>,
     file_schema: &SchemaRef,
+    column_orderings: Vec<ColumnOrdering>,
     predicate_creation_errors: &Count,
 ) -> Option<Arc<PruningPredicate>> {
-    match PruningPredicate::try_new(predicate, Arc::clone(file_schema)) {
+    match PruningPredicate::try_new(predicate, Arc::clone(file_schema), column_orderings)
+    {
         Ok(pruning_predicate) => {
             if !pruning_predicate.always_true() {
                 return Some(Arc::new(pruning_predicate));
@@ -385,16 +420,19 @@ pub(crate) fn build_pruning_predicate(
 pub(crate) fn build_page_pruning_predicate(
     predicate: &Arc<dyn PhysicalExpr>,
     file_schema: &SchemaRef,
+    column_orderings: Vec<ColumnOrdering>,
 ) -> Arc<PagePruningAccessPlanFilter> {
     Arc::new(PagePruningAccessPlanFilter::new(
         predicate,
         Arc::clone(file_schema),
+        column_orderings,
     ))
 }
 
 pub(crate) fn build_pruning_predicates(
     predicate: Option<&Arc<dyn PhysicalExpr>>,
     file_schema: &SchemaRef,
+    column_orderings: Vec<ColumnOrdering>,
     predicate_creation_errors: &Count,
 ) -> (
     Option<Arc<PruningPredicate>>,
@@ -406,9 +444,11 @@ pub(crate) fn build_pruning_predicates(
     let pruning_predicate = build_pruning_predicate(
         Arc::clone(predicate),
         file_schema,
+        column_orderings.clone(),
         predicate_creation_errors,
     );
-    let page_pruning_predicate = build_page_pruning_predicate(predicate, file_schema);
+    let page_pruning_predicate =
+        build_page_pruning_predicate(predicate, file_schema, column_orderings);
     (pruning_predicate, Some(page_pruning_predicate))
 }
 

datafusion/datasource-parquet/src/page_filter.rs

@@ -31,7 +31,7 @@ use arrow::{
 use datafusion_common::pruning::PruningStatistics;
 use datafusion_common::ScalarValue;
 use datafusion_physical_expr::{split_conjunction, PhysicalExpr};
-use datafusion_physical_optimizer::pruning::PruningPredicate;
+use datafusion_physical_optimizer::pruning::{ColumnOrdering, PruningPredicate};
 
 use log::{debug, trace};
 use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
@@ -119,14 +119,19 @@ pub struct PagePruningAccessPlanFilter {
 impl PagePruningAccessPlanFilter {
     /// Create a new [`PagePruningAccessPlanFilter`] from a physical
     /// expression.
-    pub fn new(expr: &Arc<dyn PhysicalExpr>, schema: SchemaRef) -> Self {
+    pub fn new(
+        expr: &Arc<dyn PhysicalExpr>,
+        schema: SchemaRef,
+        column_orderings: Vec<ColumnOrdering>,
+    ) -> Self {
         // extract any single column predicates
         let predicates = split_conjunction(expr)
             .into_iter()
             .filter_map(|predicate| {
                 let pp = match PruningPredicate::try_new(
                     Arc::clone(predicate),
                     Arc::clone(&schema),
+                    column_orderings.clone(),
                 ) {
                     Ok(pp) => pp,
                     Err(e) => {