This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 30d4368127 return absent stats when filters are pushed down (#12471)
30d4368127 is described below
commit 30d4368127db80c125a1adea22a955a496a516df
Author: waruto <[email protected]>
AuthorDate: Mon Sep 23 18:21:50 2024 +0800
return absent stats when filters are pushed down (#12471)
* do not pushdown filters that can be resolved only using partition cols
return absent stats when filters are pushed down
* fix and add test
* Update datafusion/core/src/datasource/physical_plan/parquet/mod.rs
Co-authored-by: Andrew Lamb <[email protected]>
* add test for partition pruning filters
---------
Co-authored-by: Andrew Lamb <[email protected]>
---
datafusion/core/src/datasource/listing/table.rs | 14 ++++
.../src/datasource/physical_plan/parquet/mod.rs | 13 +++-
datafusion/core/tests/parquet/file_statistics.rs | 21 ++++++
datafusion/core/tests/sql/path_partition.rs | 78 ++++++++++++++++++++--
datafusion/physical-plan/src/execution_plan.rs | 3 +
5 files changed, 122 insertions(+), 7 deletions(-)
diff --git a/datafusion/core/src/datasource/listing/table.rs
b/datafusion/core/src/datasource/listing/table.rs
index d2c2bd7a12..9142e922eb 100644
--- a/datafusion/core/src/datasource/listing/table.rs
+++ b/datafusion/core/src/datasource/listing/table.rs
@@ -819,6 +819,20 @@ impl TableProvider for ListingTable {
.map(|col| Ok(self.table_schema.field_with_name(&col.0)?.clone()))
.collect::<Result<Vec<_>>>()?;
+ // If the filters can be resolved using only partition cols, there is
no need to
+ // pushdown it to TableScan, otherwise, `unhandled` pruning predicates
will be generated
+ let table_partition_col_names = table_partition_cols
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>();
+ let filters = filters
+ .iter()
+ .filter(|filter| {
+ !expr_applicable_for_cols(&table_partition_col_names, filter)
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+
let filters = conjunction(filters.to_vec())
.map(|expr| -> Result<_> {
// NOTE: Use the table schema (NOT file schema) here because
`expr` may contain references to partition columns.
diff --git a/datafusion/core/src/datasource/physical_plan/parquet/mod.rs
b/datafusion/core/src/datasource/physical_plan/parquet/mod.rs
index 7aacf5cf5c..ce679bfa76 100644
--- a/datafusion/core/src/datasource/physical_plan/parquet/mod.rs
+++ b/datafusion/core/src/datasource/physical_plan/parquet/mod.rs
@@ -743,7 +743,18 @@ impl ExecutionPlan for ParquetExec {
}
fn statistics(&self) -> Result<Statistics> {
- Ok(self.projected_statistics.clone())
+ // When filters are pushed down, we have no way of knowing the exact
statistics.
+ // Note that pruning predicate is also a kind of filter pushdown.
+ // (bloom filters use `pruning_predicate` too)
+ let stats = if self.pruning_predicate.is_some()
+ || self.page_pruning_predicate.is_some()
+ || (self.predicate.is_some() && self.pushdown_filters())
+ {
+ self.projected_statistics.clone().to_inexact()
+ } else {
+ self.projected_statistics.clone()
+ };
+ Ok(stats)
}
fn fetch(&self) -> Option<usize> {
diff --git a/datafusion/core/tests/parquet/file_statistics.rs
b/datafusion/core/tests/parquet/file_statistics.rs
index cd62c3bf42..18d8300fb2 100644
--- a/datafusion/core/tests/parquet/file_statistics.rs
+++ b/datafusion/core/tests/parquet/file_statistics.rs
@@ -36,8 +36,29 @@ use datafusion_execution::config::SessionConfig;
use datafusion_execution::runtime_env::RuntimeEnvBuilder;
use datafusion::execution::session_state::SessionStateBuilder;
+use datafusion_expr::{col, lit, Expr};
use tempfile::tempdir;
+#[tokio::test]
+async fn check_stats_precision_with_filter_pushdown() {
+ let testdata = datafusion::test_util::parquet_test_data();
+ let filename = format!("{}/{}", testdata, "alltypes_plain.parquet");
+ let table_path = ListingTableUrl::parse(filename).unwrap();
+
+ let opt = ListingOptions::new(Arc::new(ParquetFormat::default()));
+ let table = get_listing_table(&table_path, None, &opt).await;
+ let (_, _, state) = get_cache_runtime_state();
+ // Scan without filter, stats are exact
+ let exec = table.scan(&state, None, &[], None).await.unwrap();
+ assert_eq!(exec.statistics().unwrap().num_rows, Precision::Exact(8));
+
+ // Scan with filter pushdown, stats are inexact
+ let filter = Expr::gt(col("id"), lit(1));
+
+ let exec = table.scan(&state, None, &[filter], None).await.unwrap();
+ assert_eq!(exec.statistics().unwrap().num_rows, Precision::Inexact(8));
+}
+
#[tokio::test]
async fn load_table_stats_with_session_level_cache() {
let testdata = datafusion::test_util::parquet_test_data();
diff --git a/datafusion/core/tests/sql/path_partition.rs
b/datafusion/core/tests/sql/path_partition.rs
index 7e7544bdb7..919054e833 100644
--- a/datafusion/core/tests/sql/path_partition.rs
+++ b/datafusion/core/tests/sql/path_partition.rs
@@ -25,6 +25,7 @@ use std::sync::Arc;
use arrow::datatypes::DataType;
use datafusion::datasource::listing::ListingTableUrl;
+use datafusion::datasource::physical_plan::ParquetExec;
use datafusion::{
assert_batches_sorted_eq,
datasource::{
@@ -36,6 +37,7 @@ use datafusion::{
prelude::SessionContext,
test_util::{self, arrow_test_data, parquet_test_data},
};
+use datafusion_catalog::TableProvider;
use datafusion_common::stats::Precision;
use datafusion_common::ScalarValue;
use datafusion_execution::config::SessionConfig;
@@ -43,6 +45,9 @@ use datafusion_execution::config::SessionConfig;
use async_trait::async_trait;
use bytes::Bytes;
use chrono::{TimeZone, Utc};
+use datafusion_expr::{col, lit, Expr, Operator};
+use datafusion_physical_expr::expressions::{BinaryExpr, Column, Literal};
+use datafusion_physical_expr::PhysicalExpr;
use futures::stream::{self, BoxStream};
use object_store::{
path::Path, GetOptions, GetResult, GetResultPayload, ListResult,
ObjectMeta,
@@ -51,6 +56,52 @@ use object_store::{
use object_store::{Attributes, MultipartUpload, PutMultipartOpts, PutPayload};
use url::Url;
+#[tokio::test]
+async fn parquet_partition_pruning_filter() -> Result<()> {
+ let ctx = SessionContext::new();
+
+ let table = create_partitioned_alltypes_parquet_table(
+ &ctx,
+ &[
+ "year=2021/month=09/day=09/file.parquet",
+ "year=2021/month=10/day=09/file.parquet",
+ "year=2021/month=10/day=28/file.parquet",
+ ],
+ &[
+ ("year", DataType::Int32),
+ ("month", DataType::Int32),
+ ("day", DataType::Int32),
+ ],
+ "mirror:///",
+ "alltypes_plain.parquet",
+ )
+ .await;
+
+ // The first three filters can be resolved using only the partition
columns.
+ let filters = [
+ Expr::eq(col("year"), lit(2021)),
+ Expr::eq(col("month"), lit(10)),
+ Expr::eq(col("day"), lit(28)),
+ Expr::gt(col("id"), lit(1)),
+ ];
+ let exec = table.scan(&ctx.state(), None, &filters, None).await?;
+ let parquet_exec = exec.as_any().downcast_ref::<ParquetExec>().unwrap();
+ let pred = parquet_exec.predicate().unwrap();
+ // Only the last filter should be pushdown to TableScan
+ let expected = Arc::new(BinaryExpr::new(
+ Arc::new(Column::new_with_schema("id", &exec.schema()).unwrap()),
+ Operator::Gt,
+ Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+ ));
+
+ assert!(pred.as_any().is::<BinaryExpr>());
+ let pred = pred.as_any().downcast_ref::<BinaryExpr>().unwrap();
+
+ assert_eq!(pred, expected.as_any());
+
+ Ok(())
+}
+
#[tokio::test]
async fn parquet_distinct_partition_col() -> Result<()> {
let ctx = SessionContext::new();
@@ -491,7 +542,7 @@ async fn parquet_statistics() -> Result<()> {
// stats for the first col are read from the parquet file
assert_eq!(stat_cols[0].null_count, Precision::Exact(1));
// TODO assert partition column stats once implemented (#1186)
- assert_eq!(stat_cols[1], ColumnStatistics::new_unknown(),);
+ assert_eq!(stat_cols[1], ColumnStatistics::new_unknown());
Ok(())
}
@@ -563,6 +614,25 @@ async fn register_partitioned_alltypes_parquet(
table_path: &str,
source_file: &str,
) {
+ let table = create_partitioned_alltypes_parquet_table(
+ ctx,
+ store_paths,
+ partition_cols,
+ table_path,
+ source_file,
+ )
+ .await;
+ ctx.register_table("t", table)
+ .expect("registering listing table failed");
+}
+
+async fn create_partitioned_alltypes_parquet_table(
+ ctx: &SessionContext,
+ store_paths: &[&str],
+ partition_cols: &[(&str, DataType)],
+ table_path: &str,
+ source_file: &str,
+) -> Arc<dyn TableProvider> {
let testdata = parquet_test_data();
let parquet_file_path = format!("{testdata}/{source_file}");
let url = Url::parse("mirror://").unwrap();
@@ -591,11 +661,7 @@ async fn register_partitioned_alltypes_parquet(
let config = ListingTableConfig::new(table_path)
.with_listing_options(options)
.with_schema(file_schema);
-
- let table = ListingTable::try_new(config).unwrap();
-
- ctx.register_table("t", Arc::new(table))
- .expect("registering listing table failed");
+ Arc::new(ListingTable::try_new(config).unwrap())
}
#[derive(Debug)]
diff --git a/datafusion/physical-plan/src/execution_plan.rs
b/datafusion/physical-plan/src/execution_plan.rs
index 40d18ebece..542861688d 100644
--- a/datafusion/physical-plan/src/execution_plan.rs
+++ b/datafusion/physical-plan/src/execution_plan.rs
@@ -379,6 +379,9 @@ pub trait ExecutionPlan: Debug + DisplayAs + Send + Sync {
/// Returns statistics for this `ExecutionPlan` node. If statistics are not
/// available, should return [`Statistics::new_unknown`] (the default), not
/// an error.
+ ///
+ /// For TableScan executors, which supports filter pushdown, special
attention
+ /// needs to be paid to whether the stats returned by this method are
exact or not
fn statistics(&self) -> Result<Statistics> {
Ok(Statistics::new_unknown(&self.schema()))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
