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) => {