You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/16 01:19:44 UTC
(datafusion) 01/01: fix: prune row groups when file statistics collapse the predicate to a constant (#24770)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24770-2c1f74f10b2d6bd3506fad3fc37e5361621f6a82
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit be5a96e76df9e0111cd4f21998b60a8db7212ba3
Author: Jens Holdgaard Pedersen <je...@holdgaard.org>
AuthorDate: Wed Sep 16 01:19:23 2026 +0000
fix: prune row groups when file statistics collapse the predicate to a constant (#24770)
> [!NOTE]
> This fix was investigated and written with AI assistance (Claude
Code), posted with the account owner's review and consent.
## Which issue does this PR close?
- Closes #24769.
## Rationale for this change
A query like `SELECT ... WHERE col = 'x'` over a Parquet file whose
statistics say `col` is entirely NULL cannot match any row, and
DataFusion 54 pruned such row groups from the footer without reading
them. On 55 the same query reads them.
The cause is an interaction rather than a bug in pruning itself.
`constant_columns_from_stats` substitutes columns that file statistics
prove constant, and its all-NULL branch folds such a column to a NULL
literal. Once substituted, the predicate simplifies to a bare constant —
`NULL` here — at which point `build_pruning_predicates` returns `None`
(there are no column references left to build a pruning predicate over),
and `prune_row_groups` falls through with no pruning at all.
So for exactly the files where the statistics carry the *most*
information, the substitution is strictly counterproductive: before it,
the pruning predicate's own `col_null_count != row_count` conjunct
proved the row group empty and skipped it.
This surfaced as a regression when bisecting a real workload (a Parquet
log store where a `body` column is NULL for the vast majority of rows)
from 54 → 55. The bisect lands on #22969, which removed
`ListingOptions::collect_stat` in favour of the session's
`execution.collect_statistics` — default `true`. That change is correct
in itself; it simply began feeding per-file statistics to the
substitution on paths that previously had none, exposing the gap.
Setting `execution.collect_statistics = false` restores pruning on 55,
which is a useful confirmation but obviously not a fix.
Results were never wrong — a filter drops NULL and false rows alike —
but the scan work is real: row groups that used to be skipped from the
footer are now decoded in full.
## What changes are included in this PR?
`prune_row_groups` now recognises the collapsed-to-constant case: if the
(post-substitution, post-simplification) predicate is a `false` or NULL
literal, every remaining row group is skipped and the skip is credited
to `row_groups_pruned_statistics`. A small
`RowGroupAccessPlanFilter::skip_all` helper is added alongside the
existing `prune_by_*` methods.
The check is deliberately narrow — a downcast to `Literal` plus a
NULL/`false` value test — so it cannot affect predicates that still
reference columns; those take the existing path unchanged.
## Are these changes tested?
Yes: `test_prune_all_null_column_equality_from_file_statistics` in
`opener/mod.rs`, modelled on the neighbouring
`test_prune_on_partition_values_and_file_statistics`. It fails on
current `main` (3 rows scanned, 0 pruned) and passes with this change.
One note on how it asserts, in case it saves a reviewer time: it checks
the `row_groups_pruned_statistics` metric rather than the returned row
count. A row-count assertion cannot distinguish "pruned" from "scanned,
then row-filtered" — both give zero rows — and an earlier draft of this
test passed against the unfixed code for exactly that reason.
Verified `cargo test -p datafusion-datasource-parquet --lib opener::` —
47 passed. The 16 pre-existing `bloom_filter::tests` failures on this
crate are unrelated and reproduce identically on an untouched checkout.
## Are there any user-facing changes?
No API or result changes. Queries that were already correct stay
correct; affected scans read less data. Plans may show more row groups
pruned, and `row_groups_pruned_statistics` increases correspondingly.
---------
Co-authored-by: Jens Holdgaard Pedersen <je...@MacBook-Pro-tilhrende-sysadm.local>
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
---
datafusion/datasource-parquet/src/opener/mod.rs | 477 ++++++++++++++++++++-
.../datasource-parquet/src/row_group_filter.rs | 11 +
2 files changed, 485 insertions(+), 3 deletions(-)
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index ec24462db0..d430d81e44 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -58,8 +58,9 @@ use datafusion_common::{
ColumnStatistics, HashSet, Result, ScalarValue, Statistics, exec_err, internal_err,
};
use datafusion_datasource::{PartitionedFile, TableSchema};
-use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking};
+use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking, Literal};
use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
+use datafusion_physical_expr::utils::collect_columns;
use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use datafusion_physical_expr_common::sort_expr::LexOrdering;
@@ -460,6 +461,27 @@ struct PreparedParquetOpen {
enable_page_index: bool,
enable_bloom_filter: bool,
enable_row_group_stats_pruning: bool,
+ /// True when substituting the columns `constant_columns_from_stats`
+ /// proved constant for this file is on its own enough to collapse the
+ /// predicate to a constant no row can satisfy — see `prune_row_groups`.
+ ///
+ /// Provisional until `prepare_filters` has the physical file schema: see
+ /// `stats_constant_columns_in_predicate`.
+ stats_prove_unsatisfiable: bool,
+ /// The stats-derived constant columns the predicate actually references,
+ /// recorded only when `stats_prove_unsatisfiable` is set.
+ ///
+ /// Collected file statistics cannot distinguish a column that is
+ /// physically absent from the Parquet file from one that is present and
+ /// all NULL: `statistics_from_parquet_metadata` documents that a column
+ /// in the Arrow schema but not the Parquet schema gets
+ /// `Precision::Exact(null)` min/max and an `Exact(num_rows)` null count —
+ /// exactly what a present all-NULL column produces. So an absent column
+ /// can enter the constants as a NULL and collapse the predicate, which
+ /// would prune the very missing-column case this flag exists to leave
+ /// alone. `prepare_filters` clears the flag once the physical schema
+ /// shows whether these columns are really in the file.
+ stats_constant_columns_in_predicate: Vec<String>,
limit: Option<usize>,
coerce_int96: Option<TimeUnit>,
coerce_int96_tz: Option<Arc<str>>,
@@ -794,10 +816,42 @@ impl ParquetMorselizer {
// Note that if there are statistics for partition columns there will be overlap,
// but since we use a HashMap, we'll just overwrite the partition values with the
// constant values from statistics (which should be the same).
- literal_columns.extend(constant_columns_from_stats(
+ let stats_constants = constant_columns_from_stats(
partitioned_file.statistics.as_deref(),
&logical_file_schema,
- ));
+ );
+ // Whether the statistics substitution *itself* proves the predicate
+ // unsatisfiable has to be decided here, before the partition-value
+ // and missing-column rewriting below: a predicate can collapse to a
+ // constant through either of those too, and only a collapse the
+ // file's own statistics produced is proof for `prune_row_groups`.
+ // The referenced columns are recorded alongside the flag because this
+ // is also too early to know whether they exist in the file at all —
+ // `prepare_filters` re-checks them against the physical file schema.
+ let stats_referenced: Vec<String> =
+ match (&self.predicate, stats_constants.is_empty()) {
+ (Some(predicate), false) => collect_columns(predicate)
+ .iter()
+ .filter(|c| stats_constants.contains_key(c.name()))
+ .map(|c| c.name().to_string())
+ .collect(),
+ _ => Vec::new(),
+ };
+ // `stats_referenced` is the cheap short-circuit so the simplifier only
+ // runs when a stats-derived constant is actually referenced.
+ let stats_prove_unsatisfiable = !stats_referenced.is_empty()
+ && self.predicate.as_ref().is_some_and(|p| {
+ stats_alone_unsatisfiable(
+ p,
+ &stats_constants,
+ self.table_schema.table_schema(),
+ )
+ });
+ let stats_constant_columns_in_predicate = match stats_prove_unsatisfiable {
+ true => stats_referenced,
+ false => Vec::new(),
+ };
+ literal_columns.extend(stats_constants);
let mut projection = self.projection.clone();
let mut predicate = self.predicate.clone();
@@ -867,6 +921,8 @@ impl ParquetMorselizer {
enable_page_index: self.enable_page_index,
enable_bloom_filter: self.enable_bloom_filter,
enable_row_group_stats_pruning: self.enable_row_group_stats_pruning,
+ stats_prove_unsatisfiable,
+ stats_constant_columns_in_predicate,
limit: self.limit,
coerce_int96: self.coerce_int96,
coerce_int96_tz: self.coerce_int96_tz.clone(),
@@ -1073,6 +1129,24 @@ impl MetadataLoadedParquetOpen {
}
prepared.physical_file_schema = Arc::clone(&physical_file_schema);
+ // The physical schema is only known here, so this is the first point
+ // that can tell a stats-derived NULL constant for a column genuinely
+ // present in the file from one synthesized for a column the file does
+ // not have (see `stats_constant_columns_in_predicate`). Withdraw the
+ // proof if any referenced column turns out to be absent: pruning on a
+ // synthesized NULL would skip the missing-column case this flag is
+ // meant to leave alone, and a custom `PhysicalExprAdapter` may fill a
+ // missing column with a non-NULL default, so the skip could drop rows
+ // that default would have matched.
+ if prepared.stats_prove_unsatisfiable
+ && prepared
+ .stats_constant_columns_in_predicate
+ .iter()
+ .any(|name| physical_file_schema.field_with_name(name).is_err())
+ {
+ prepared.stats_prove_unsatisfiable = false;
+ }
+
// Build predicates for this specific file
let pruning_predicate = build_pruning_predicates(
prepared.predicate.as_ref(),
@@ -1128,6 +1202,37 @@ impl FiltersPreparedParquetOpen {
row_groups.prune_by_range(rg_metadata, range);
}
+ // Substituting the columns file statistics proved constant
+ // (`constant_columns_from_stats`) collapsed the predicate to a
+ // constant that can never be true (`false`, or NULL — filters treat
+ // NULL as false), so the file's own statistics have proven that no
+ // row can match: skip every remaining row group, credited to
+ // statistics pruning. Without this, the substitution is strictly
+ // counterproductive for such files: it *removes* the column
+ // references the pruning predicate would need
+ // (`build_pruning_predicates` returns `None` for a bare literal), so
+ // a file that was previously pruned via its `null_count` statistics
+ // is instead scanned in full.
+ //
+ // The flag is deliberately computed from the statistics substitution
+ // alone (see `ParquetMorselizer::prepare`). A predicate can also
+ // collapse via the missing-column adapter or partition-value
+ // folding, and those paths keep their existing behaviour (files
+ // pruned by partition values are `FilePruner`'s job); a mixed
+ // predicate that only collapses once one of them has run is not
+ // treated as proof here.
+ if prepared.enable_row_group_stats_pruning && prepared.stats_prove_unsatisfiable {
+ prepared
+ .file_metrics
+ .row_groups_pruned_statistics
+ .add_pruned(row_groups.remaining_row_group_count());
+ row_groups.skip_all();
+ return Ok(RowGroupsPrunedParquetOpen {
+ prepared: self,
+ row_groups,
+ });
+ }
+
// If there is a predicate that can be evaluated against the metadata
if let Some(predicate) = self.pruning_predicate.as_ref().map(|p| p.as_ref()) {
if prepared.enable_row_group_stats_pruning {
@@ -1720,6 +1825,38 @@ fn row_group_bytes(rg_meta: &RowGroupMetaData) -> u64 {
type ConstantColumns = HashMap<String, ScalarValue>;
+/// True when substituting `stats_constants` — the columns this file's
+/// statistics prove constant — is on its own enough to collapse `predicate`
+/// to a constant no row can satisfy (`false`, or NULL, which filters treat
+/// as false).
+///
+/// Only the statistics substitution is applied, so a predicate that needs
+/// the missing-column adapter or partition-value folding to collapse does
+/// not qualify: those are separate proofs with their own pruning paths.
+///
+/// `schema` must resolve every column the predicate can reference — pass the
+/// full table schema, since the simplifier types each node as it walks and
+/// the substitution leaves non-constant columns in place. Substitution or
+/// simplification failing is not proof, so both answer `false` and leave the
+/// caller's normal pruning paths untouched.
+fn stats_alone_unsatisfiable(
+ predicate: &Arc<dyn PhysicalExpr>,
+ stats_constants: &ConstantColumns,
+ schema: &Schema,
+) -> bool {
+ let Ok(substituted) =
+ replace_columns_with_literals(Arc::clone(predicate), stats_constants)
+ else {
+ return false;
+ };
+ let Ok(simplified) = PhysicalExprSimplifier::new(schema).simplify(substituted) else {
+ return false;
+ };
+ simplified.downcast_ref::<Literal>().is_some_and(|l| {
+ l.value().is_null() || l.value() == &ScalarValue::Boolean(Some(false))
+ })
+}
+
/// Extract constant column values from statistics, keyed by column name in the logical file schema.
fn constant_columns_from_stats(
statistics: Option<&Statistics>,
@@ -1886,6 +2023,7 @@ async fn load_page_index<T: AsyncFileReader>(
mod test {
use super::*;
use super::{ConstantColumns, ParquetMorselizer, constant_columns_from_stats};
+ use crate::metadata::DFParquetMetadata;
use crate::{
CachedParquetFileReaderFactory, DefaultParquetFileReaderFactory,
ParquetFileReaderFactory, ParquetRowSelection, RowGroupAccess,
@@ -2512,6 +2650,43 @@ mod test {
write_parquet_batches(store, filename, vec![batch], None).await
}
+ /// Write `batch` and return the statistics a collecting `ListingTable`
+ /// would attach for `logical_file_schema`, produced by the same
+ /// `statistics_from_parquet_metadata` the read path uses.
+ ///
+ /// Tests about schema evolution have to go through this rather than
+ /// hand-build `Statistics`: the shape that function synthesizes for a
+ /// logical column the file does not physically contain is the whole
+ /// subject, so a hand-built approximation can silently stop matching it.
+ async fn write_parquet_with_collected_statistics(
+ store: Arc<dyn ObjectStore>,
+ filename: &str,
+ batch: RecordBatch,
+ logical_file_schema: &SchemaRef,
+ ) -> (usize, Statistics) {
+ use parquet::file::metadata::ParquetMetaDataReader;
+
+ let mut out = BytesMut::new().writer();
+ {
+ let mut writer =
+ ArrowWriter::try_new(&mut out, batch.schema(), None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.finish().unwrap();
+ }
+ let data = out.into_inner().freeze();
+ let data_len = data.len();
+ let metadata = ParquetMetaDataReader::new()
+ .parse_and_finish(&data)
+ .unwrap();
+ let statistics = DFParquetMetadata::statistics_from_parquet_metadata(
+ &metadata,
+ logical_file_schema,
+ )
+ .unwrap();
+ store.put(&Path::from(filename), data.into()).await.unwrap();
+ (data_len, statistics)
+ }
+
/// Write multiple batches to a parquet file with optional writer properties
async fn write_parquet_batches(
store: Arc<dyn ObjectStore>,
@@ -3040,6 +3215,302 @@ mod test {
assert_eq!(num_rows, 0);
}
+ /// Row groups this scan skipped through statistics pruning.
+ ///
+ /// The row count alone cannot tell "pruned" from "scanned, then
+ /// row-filtered" — both yield zero rows — so the tests around
+ /// constant-column substitution assert on this metric instead.
+ fn pruned_row_groups_statistics(metrics: &ExecutionPlanMetricsSet) -> usize {
+ use datafusion_physical_plan::metrics::MetricValue;
+ metrics
+ .clone_inner()
+ .iter()
+ .find_map(|m| match m.value() {
+ MetricValue::PruningMetrics {
+ name,
+ pruning_metrics,
+ } if name == "row_groups_pruned_statistics" => {
+ Some(pruning_metrics.pruned())
+ }
+ _ => None,
+ })
+ .expect("row_groups_pruned_statistics metric is emitted")
+ }
+
+ #[tokio::test]
+ async fn test_prune_all_null_column_equality_from_file_statistics() {
+ // Regression: a column whose file statistics say every value is
+ // NULL cannot satisfy `col = <literal>`, so the file's row groups
+ // must be pruned. Constant-column substitution folds the all-NULL
+ // column to a NULL literal, collapsing the predicate to a
+ // constant; the opener has to recognise that and skip the row
+ // groups, rather than lose the pruning because the substituted
+ // predicate no longer references any column.
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+
+ let batch = record_batch!(
+ ("a", Int32, vec![Some(1), Some(2), Some(3)]),
+ ("b", Utf8, vec![None::<&str>, None, None])
+ )
+ .unwrap();
+ let data_size =
+ write_parquet(Arc::clone(&store), "file.parquet", batch.clone()).await;
+ let file_schema = batch.schema();
+ let mut file = PartitionedFile::new(
+ "file.parquet".to_string(),
+ u64::try_from(data_size).unwrap(),
+ );
+ // Statistics as a collecting `ListingTable` would supply them: the
+ // `b` column's null count equals the row count, i.e. every value is
+ // NULL. (`new_unknown` pre-fills one entry per field, so overwrite
+ // rather than append.)
+ let mut statistics = Statistics::new_unknown(&file_schema);
+ statistics.num_rows = Precision::Exact(3);
+ statistics.column_statistics[1].null_count = Precision::Exact(3);
+ file.statistics = Some(Arc::new(statistics));
+ let table_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Utf8, true),
+ ]));
+ let table_schema_for_opener = TableSchemaBuilder::from(&file_schema).build();
+ let make_opener = |predicate, metrics: ExecutionPlanMetricsSet| {
+ ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_table_schema(table_schema_for_opener.clone())
+ .with_projection_indices(&[0])
+ .with_predicate(predicate)
+ .with_row_group_stats_pruning(true)
+ .with_metrics(metrics)
+ .build()
+ };
+
+ // `b = 'x'` cannot match: every `b` is NULL, so the file's one row
+ // group is pruned from statistics rather than read and filtered.
+ let metrics = ExecutionPlanMetricsSet::new();
+ let expr = col("b").eq(lit("x"));
+ let predicate = logical2physical(&expr, &table_schema);
+ let opener = make_opener(predicate, metrics.clone());
+ let stream = open_file(&opener, file.clone()).await.unwrap();
+ let (_, num_rows) = count_batches_and_rows(stream).await;
+ assert_eq!(num_rows, 0);
+ assert_eq!(
+ pruned_row_groups_statistics(&metrics),
+ 1,
+ "an all-NULL column cannot satisfy equality: the row group must be pruned"
+ );
+
+ // A predicate the statistics cannot disprove still reads the file,
+ // so the skip above is not simply "prune everything".
+ let metrics = ExecutionPlanMetricsSet::new();
+ let expr = col("a").eq(lit(2));
+ let predicate = logical2physical(&expr, &table_schema);
+ let opener = make_opener(predicate, metrics.clone());
+ let stream = open_file(&opener, file).await.unwrap();
+ let (num_batches, num_rows) = count_batches_and_rows(stream).await;
+ assert_eq!(num_batches, 1);
+ assert_eq!(num_rows, 3);
+ assert_eq!(pruned_row_groups_statistics(&metrics), 0);
+ }
+
+ #[tokio::test]
+ async fn test_prune_exact_constant_column_false_predicate_from_file_statistics() {
+ // The `false` half of the branch above, which the all-NULL case
+ // only covers for NULL: exact min == max statistics prove `a`
+ // constant, so `a = <other literal>` folds to `false` and the row
+ // group is pruned on that proof.
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+
+ let batch = record_batch!(("a", Int32, vec![Some(7), Some(7), Some(7)])).unwrap();
+ let data_size =
+ write_parquet(Arc::clone(&store), "file.parquet", batch.clone()).await;
+ let file_schema = batch.schema();
+ let mut file = PartitionedFile::new(
+ "file.parquet".to_string(),
+ u64::try_from(data_size).unwrap(),
+ );
+ let mut statistics = Statistics::new_unknown(&file_schema);
+ statistics.num_rows = Precision::Exact(3);
+ statistics.column_statistics[0].null_count = Precision::Exact(0);
+ statistics.column_statistics[0].min_value =
+ Precision::Exact(ScalarValue::Int32(Some(7)));
+ statistics.column_statistics[0].max_value =
+ Precision::Exact(ScalarValue::Int32(Some(7)));
+ file.statistics = Some(Arc::new(statistics));
+ let table_schema =
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let table_schema_for_opener = TableSchemaBuilder::from(&file_schema).build();
+
+ let metrics = ExecutionPlanMetricsSet::new();
+ let expr = col("a").eq(lit(8));
+ let predicate = logical2physical(&expr, &table_schema);
+ let opener = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_table_schema(table_schema_for_opener)
+ .with_projection_indices(&[0])
+ .with_predicate(predicate)
+ .with_row_group_stats_pruning(true)
+ .with_metrics(metrics.clone())
+ .build();
+ let stream = open_file(&opener, file).await.unwrap();
+ let (_, num_rows) = count_batches_and_rows(stream).await;
+ assert_eq!(num_rows, 0);
+ assert_eq!(
+ pruned_row_groups_statistics(&metrics),
+ 1,
+ "a column statistics prove constant cannot equal a different literal: \
+ the row group must be pruned"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_no_prune_when_missing_column_collapses_mixed_predicate() {
+ // A stats-derived constant appearing in the predicate is not proof
+ // that the statistics are what made it unsatisfiable. Here `a = 1`
+ // is proven true by statistics and folds away, and the collapse to
+ // NULL comes from `b = 2` — a column this file does not have. That
+ // is the schema-evolution path, which deliberately scans and
+ // row-filters rather than pruning, so the statistics branch must not
+ // fire.
+ //
+ // The statistics come from `statistics_from_parquet_metadata` rather
+ // than being hand-built, because the shape that matters is the one it
+ // synthesizes for a missing column: `Exact(null)` min/max and an
+ // `Exact(num_rows)` null count, indistinguishable from a present
+ // all-NULL column. Hand-building anything weaker (leaving `b`
+ // unknown, say) stops `b` from ever becoming a constant, and the test
+ // then passes without exercising this path at all.
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+
+ // The file has only `a`; `b` is in the table schema but missing
+ // from this file.
+ let batch = record_batch!(("a", Int32, vec![Some(1), Some(1), Some(1)])).unwrap();
+ let physical_schema = batch.schema();
+ let logical_file_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, true),
+ ]));
+ let (data_size, statistics) = write_parquet_with_collected_statistics(
+ Arc::clone(&store),
+ "file.parquet",
+ batch,
+ &logical_file_schema,
+ )
+ .await;
+ let mut file = PartitionedFile::new(
+ "file.parquet".to_string(),
+ u64::try_from(data_size).unwrap(),
+ );
+ // Assert the premise instead of trusting it: if collection ever stops
+ // representing the absent `b` as an all-NULL column, this test is no
+ // longer covering the case it is named for and should say so loudly.
+ assert_eq!(
+ statistics.column_statistics[1].null_count,
+ Precision::Exact(3),
+ "collection must still report the absent column as all NULL for \
+ this test to exercise the missing-column path"
+ );
+ file.statistics = Some(Arc::new(statistics));
+ let table_schema_for_opener =
+ TableSchemaBuilder::from(&logical_file_schema).build();
+
+ let metrics = ExecutionPlanMetricsSet::new();
+ let expr = col("a").eq(lit(1)).and(col("b").eq(lit(2)));
+ let predicate = logical2physical(&expr, &logical_file_schema);
+ let opener = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_table_schema(table_schema_for_opener)
+ .with_projection_indices(&[0])
+ .with_predicate(predicate)
+ .with_row_group_stats_pruning(true)
+ .with_metrics(metrics.clone())
+ .build();
+ let stream = open_file(&opener, file).await.unwrap();
+ let (_, num_rows) = count_batches_and_rows(stream).await;
+ // The row group is read and its rows handed up for the filter above
+ // to apply, which is the missing-column path this test protects.
+ assert_eq!(
+ num_rows, 3,
+ "the file must be scanned, not pruned, on the missing-column path"
+ );
+ assert_eq!(
+ pruned_row_groups_statistics(&metrics),
+ 0,
+ "the collapse came from the missing column, not from file \
+ statistics: the statistics branch must not claim the prune"
+ );
+ // Guard against the file schema drifting: `b` really is absent.
+ assert!(physical_schema.field_with_name("b").is_err());
+ }
+
+ #[tokio::test]
+ async fn test_prune_when_present_all_null_column_collapses_mixed_predicate() {
+ // The mirror of the test above, and the reason its fix cannot simply
+ // be "never trust a NULL constant". Identical predicate, identical
+ // statistics source, one difference: `b` is physically in the file
+ // and genuinely all NULL. The collapse is then real proof about this
+ // file's contents, so the row group must still be pruned.
+ //
+ // Collected statistics describe these two files identically — an
+ // `Exact(null)` min/max and an `Exact(num_rows)` null count for `b`
+ // either way — which is exactly why the decision has to wait for the
+ // physical file schema.
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+
+ let batch = record_batch!(
+ ("a", Int32, vec![Some(1), Some(1), Some(1)]),
+ ("b", Int32, vec![None::<i32>, None, None])
+ )
+ .unwrap();
+ let physical_schema = batch.schema();
+ let logical_file_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, true),
+ ]));
+ let (data_size, statistics) = write_parquet_with_collected_statistics(
+ Arc::clone(&store),
+ "file.parquet",
+ batch,
+ &logical_file_schema,
+ )
+ .await;
+ let mut file = PartitionedFile::new(
+ "file.parquet".to_string(),
+ u64::try_from(data_size).unwrap(),
+ );
+ assert_eq!(
+ statistics.column_statistics[1].null_count,
+ Precision::Exact(3),
+ "the present `b` must collect as all NULL, so that it is \
+ indistinguishable from the absent `b` in the test above"
+ );
+ file.statistics = Some(Arc::new(statistics));
+ let table_schema_for_opener =
+ TableSchemaBuilder::from(&logical_file_schema).build();
+
+ let metrics = ExecutionPlanMetricsSet::new();
+ let expr = col("a").eq(lit(1)).and(col("b").eq(lit(2)));
+ let predicate = logical2physical(&expr, &logical_file_schema);
+ let opener = ParquetMorselizerBuilder::new()
+ .with_store(Arc::clone(&store))
+ .with_table_schema(table_schema_for_opener)
+ .with_projection_indices(&[0])
+ .with_predicate(predicate)
+ .with_row_group_stats_pruning(true)
+ .with_metrics(metrics.clone())
+ .build();
+ let stream = open_file(&opener, file).await.unwrap();
+ let (_, num_rows) = count_batches_and_rows(stream).await;
+ assert_eq!(num_rows, 0);
+ assert_eq!(
+ pruned_row_groups_statistics(&metrics),
+ 1,
+ "the all-NULL column is really in the file: the statistics do \
+ prove the predicate unsatisfiable and must prune"
+ );
+ // Guard against the file schema drifting: `b` really is present.
+ assert!(physical_schema.field_with_name("b").is_ok());
+ }
+
#[tokio::test]
async fn test_prune_on_partition_value_and_data_value() {
let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
index ab690e345d..7d4048df75 100644
--- a/datafusion/datasource-parquet/src/row_group_filter.rs
+++ b/datafusion/datasource-parquet/src/row_group_filter.rs
@@ -69,6 +69,17 @@ impl RowGroupAccessPlanFilter {
Self { access_plan }
}
+ /// Skip every remaining row group: used when the predicate has been
+ /// proven unsatisfiable for the whole file (for example, when
+ /// constant-column substitution from file statistics collapsed it to a
+ /// constant `false`/NULL), so no row group can contain a matching row.
+ pub fn skip_all(&mut self) {
+ let indexes: Vec<usize> = self.access_plan.row_group_indexes();
+ for idx in indexes {
+ self.access_plan.skip(idx);
+ }
+ }
+
/// Return true if there are no row groups
pub fn is_empty(&self) -> bool {
self.access_plan.is_empty()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org