diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 384fbbf3c4450..bda21bab29b9f 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -198,15 +198,16 @@ pub struct FileScanConfig { /// Expression adapter used to adapt filters and projections that are pushed down into the scan /// from the logical schema to the physical schema of the file. pub expr_adapter_factory: Option>, - /// Statistics for the entire table (file schema + partition columns). + /// Statistics for the entire table (file schema + partition columns), before + /// projection or filtering. /// See [`FileScanConfigBuilder::with_statistics`] for more details. /// /// The effective statistics are computed on-demand via /// [`ProjectionExprs::project_statistics`]. /// - /// Note that this field is pub(crate) because accessing it directly from outside - /// would be incorrect if there are filters being applied, thus this should be accessed - /// via [`FileScanConfig::statistics`]. + /// Retained for serialization and conservative bounds used by ordering inference. + /// Use [`FileScanConfig::statistics`] for filter-adjusted statistics; these raw + /// values may no longer be exact for the scan output. pub(crate) statistics: Statistics, /// Declared physical output partitioning for this scan. /// @@ -948,7 +949,9 @@ impl DataSource for FileScanConfig { Arc::clone(schema), self.validated_output_ordering(), ) - .with_constraints(self.constraints.clone()); + .with_constraints(self.constraints.clone()) + // Filters only remove rows, so original exact extrema remain valid bounds. + .with_column_statistics(&self.statistics.column_statistics); // Only a filter that the source applies exactly gives valid equivalences. // A filter that is used only for pruning lets non-matching rows through. diff --git a/datafusion/datasource/src/file_scan_config/proto.rs b/datafusion/datasource/src/file_scan_config/proto.rs index 0e98d89f5e14f..d535a22349534 100644 --- a/datafusion/datasource/src/file_scan_config/proto.rs +++ b/datafusion/datasource/src/file_scan_config/proto.rs @@ -71,9 +71,7 @@ impl FileScanConfig { file_source, batch_size, expr_adapter_factory, - // Serialized through `statistics()` so its pushed-filter policy - // remains centralized. - statistics: _, + statistics, output_partitioning, } = self; @@ -139,11 +137,11 @@ impl FileScanConfig { (*file_compression_type.get_variant()).into(); compression as i32 }); - let statistics = self.statistics(); - + // Serialize the original statistics so decoded scans retain trusted bounds. + // The decoded scan still applies the filter policy in `statistics()`. Ok(protobuf::FileScanExecConf { file_groups: proto_file_groups, - statistics: Some((&statistics).into()), + statistics: Some(statistics.into()), limit: limit .map(|limit| usize_to_wire::(limit, "FileScanConfig", "limit")) .transpose()? diff --git a/datafusion/physical-expr/src/equivalence/properties/dependency.rs b/datafusion/physical-expr/src/equivalence/properties/dependency.rs index b626e644a2a9b..a50851c7a7c08 100644 --- a/datafusion/physical-expr/src/equivalence/properties/dependency.rs +++ b/datafusion/physical-expr/src/equivalence/properties/dependency.rs @@ -1807,6 +1807,75 @@ mod tests { Ok(()) } + #[test] + fn project_shares_ranges_with_trusted_integer_bounds() -> Result<()> { + use datafusion_common::stats::Precision; + use datafusion_common::{ColumnStatistics, ScalarValue}; + + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Int64, true), + Field::new( + "s", + DataType::Struct(Fields::from(vec![Field::new( + "x", + DataType::Int64, + true, + )])), + true, + ), + ])); + let bounds = + Interval::try_new(ScalarValue::Int64(Some(1)), ScalarValue::Int64(Some(3)))?; + let input = EquivalenceProperties::new_with_orderings( + Arc::clone(&schema), + ["a", "s"].map(|name| { + vec![PhysicalSortExpr::new_default(col(name, &schema).unwrap())] + }), + ) + .with_column_statistics(&[ + ColumnStatistics { + min_value: Precision::Exact(bounds.lower().clone()), + max_value: Precision::Exact(bounds.upper().clone()), + ..ColumnStatistics::new_unknown() + }, + ColumnStatistics::new_unknown(), + ]); + let (mapping, seen) = probe_mapping( + vec![ + (col("a", &schema)?, 0, false), + (col("a", &schema)?, 1, false), + (col("s", &schema)?, 2, false), + (col("s", &schema)?, 3, false), + ], + &schema, + )?; + let projected = input.project(&mapping, output_schema(&mapping, &schema)?); + assert_eq!( + sorted_orderings(&projected), + vec!["p0@0 ASC", "p1@1 ASC", "p2@2 ASC", "p3@3 ASC"] + ); + for probe in &seen[..2] { + let ranges = probe.lock().unwrap(); + assert!(ranges.len() >= 3); + assert!(ranges.iter().all(|range| range == &bounds)); + } + // Bound propagation, ordering substitution and ordering derivation all + // reuse the same unbounded struct range, even with trusted integer bounds. + let mut arrays = vec![]; + for probe in &seen[2..] { + let ranges = probe.lock().unwrap(); + assert!(ranges.len() >= 3); + for range in ranges.iter() { + let ScalarValue::Struct(array) = range.lower() else { + panic!("{range:?}"); + }; + arrays.push(Arc::clone(array)); + } + } + assert!(arrays.iter().all(|array| Arc::ptr_eq(array, &arrays[0]))); + Ok(()) + } + #[test] fn project_does_not_reuse_a_column_range_across_calls() -> Result<()> { for data_type in [DataType::Utf8, DataType::Int64] { diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index 1206315bbb274..56d5fb982e8e2 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -30,14 +30,20 @@ use crate::{ }; use arrow::compute::SortOptions; -use arrow::datatypes::SchemaRef; -use datafusion_common::{Constraint, Constraints, HashMap, HashSet, Result, plan_err}; +use arrow::datatypes::{DataType, SchemaRef}; +use datafusion_common::stats::Precision; +use datafusion_common::{ + ColumnStatistics, Constraint, Constraints, HashMap, HashSet, Result, plan_err, +}; use datafusion_expr::interval_arithmetic::Interval; use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; -use indexmap::IndexSet; +use hashbrown::DefaultHashBuilder; +use indexmap::{IndexMap, IndexSet}; use itertools::{Either, Itertools}; +type ColumnRanges = IndexMap; + /// `EquivalenceProperties` stores information about the output of a plan node /// that can be used to optimize the plan. Currently, it keeps track of: /// - Sort expressions (orderings), @@ -144,6 +150,8 @@ pub struct EquivalenceProperties { constraints: Constraints, /// Schema associated with this object. schema: SchemaRef, + /// Trusted bounds for non-null integer values, independent of estimate precision. + column_ranges: Option>, } /// This object serves as a cache for storing equivalent sort expressions @@ -234,7 +242,7 @@ impl EquivalenceProperties { r_expr: Arc, sort_expr: &PhysicalSortExpr, schema: &SchemaRef, - column_ranges: &ColumnRanges, + column_ranges: &ColumnRangeCache, ) -> Option { if r_expr.eq(&sort_expr.expr) { // No point in substituting an expression with itself. @@ -258,6 +266,7 @@ impl EquivalenceProperties { geq_cache: GroupingEquivalenceClass::default(), constraints: Constraints::default(), schema, + column_ranges: None, } } @@ -293,9 +302,28 @@ impl EquivalenceProperties { eq_group, constraints: Constraints::default(), schema, + column_ranges: None, } } + /// Sets exact integer extrema for this output's schema and all partitions. + /// These are bounds on non-null values, not evidence of ordering or constants. + pub fn with_column_statistics( + mut self, + column_statistics: &[ColumnStatistics], + ) -> Self { + let ranges = column_statistics + .iter() + .zip(self.schema.fields()) + .enumerate() + .filter_map(|(index, (statistics, field))| { + integer_range(field.data_type(), statistics).map(|range| (index, range)) + }) + .collect::(); + self.column_ranges = (!ranges.is_empty()).then(|| Arc::new(ranges)); + self + } + /// Returns the associated schema. pub fn schema(&self) -> &SchemaRef { &self.schema @@ -346,6 +374,13 @@ impl EquivalenceProperties { self.constraints.extend(other.constraints); self.add_equivalence_group(other.eq_group)?; self.add_orderings(other.oeq_class); + if let Some(ranges) = other.column_ranges { + match &mut self.column_ranges { + Some(current) => Arc::make_mut(current) + .extend(ranges.iter().map(|(index, range)| (*index, range.clone()))), + None => self.column_ranges = Some(ranges), + } + } self.add_groupings(other.geq_class); Ok(self) } @@ -610,10 +645,13 @@ impl EquivalenceProperties { for arg in args { match sorted.next() { Some(sort_expr) if sort_expr.expr.eq(arg) => { - let data_type = arg.data_type(&self.schema)?; arg_properties.push(ExprProperties { sort_properties: SortProperties::Ordered(sort_expr.options), - range: Interval::make_unbounded(&data_type)?, + range: column_range( + arg, + &self.schema, + self.column_ranges.as_deref(), + )?, preserves_lex_ordering: true, strictly_order_preserving: true, }); @@ -939,7 +977,7 @@ impl EquivalenceProperties { schema: &SchemaRef, mapping: &ProjectionMapping, oeq_class: OrderingEquivalenceClass, - column_ranges: &ColumnRanges, + column_ranges: &ColumnRangeCache, ) -> OrderingEquivalenceClass { let new_orderings = oeq_class.into_iter().flat_map(|order| { // Modify/expand existing orderings by substituting sort @@ -1134,6 +1172,7 @@ impl EquivalenceProperties { &self, mapping: &ProjectionMapping, mut oeq_class: OrderingEquivalenceClass, + column_ranges: &ColumnRangeCache, ) -> Vec { // Normalize source expressions in the mapping: let mapping = self.normalize_mapping(mapping); @@ -1141,9 +1180,8 @@ impl EquivalenceProperties { // orderings examines every source, even one that refers to no ordered column (it then // gets an empty set of dependencies). Sharing one cache across both steps builds the // unbounded range of a wide column referenced by many sources only once. - let column_ranges = ColumnRanges::default(); oeq_class = - Self::substitute_oeq_class(&self.schema, &mapping, oeq_class, &column_ranges); + Self::substitute_oeq_class(&self.schema, &mapping, oeq_class, column_ranges); // Get dependency map for existing orderings: let dependency_map = self.construct_dependency_map(oeq_class, &mapping); let orderings = mapping.iter().flat_map(|(source, targets)| { @@ -1151,7 +1189,7 @@ impl EquivalenceProperties { .into_iter() .filter_map(|deps| { let ep = - get_expr_properties(source, &deps, &self.schema, &column_ranges); + get_expr_properties(source, &deps, &self.schema, column_ranges); let sort_properties = ep.map(|prop| prop.sort_properties); if let Ok(SortProperties::Ordered(options)) = sort_properties { Some((options, deps)) @@ -1320,8 +1358,14 @@ impl EquivalenceProperties { output_schema: SchemaRef, eq_group: EquivalenceGroup, ) -> Self { - let orderings = - self.projected_orderings(mapping, self.oeq_cache.normal_cls.clone()); + let range_cache = ColumnRangeCache::new(self.column_ranges.as_deref()); + let column_ranges = + self.projected_column_ranges(mapping, &output_schema, &range_cache); + let orderings = self.projected_orderings( + mapping, + self.oeq_cache.normal_cls.clone(), + &range_cache, + ); let groupings = self.projected_groupings(mapping); let normal_orderings = orderings .iter() @@ -1335,11 +1379,43 @@ impl EquivalenceProperties { constraints: self.projected_constraints(mapping).unwrap_or_default(), schema: output_schema, eq_group, + column_ranges, }; result.add_groupings(groupings); result } + fn projected_column_ranges( + &self, + mapping: &ProjectionMapping, + output_schema: &SchemaRef, + column_ranges: &ColumnRangeCache, + ) -> Option> { + self.column_ranges.as_ref()?; + let mut ranges = ColumnRanges::default(); + let dependencies = Dependencies::default(); + for (source, targets) in mapping.iter() { + let Ok(properties) = + get_expr_properties(source, &dependencies, &self.schema, column_ranges) + else { + continue; + }; + let range = properties.range; + if !range.data_type().is_integer() || range.is_unbounded() { + continue; + } + for (target, _) in targets.iter() { + if let Some(column) = target.downcast_ref::() + && let Some(field) = output_schema.fields().get(column.index()) + && field.data_type() == &range.data_type() + { + ranges.insert(column.index(), range.clone()); + } + } + } + (!ranges.is_empty()).then(|| Arc::new(ranges)) + } + /// Returns the longest (potentially partial) permutation satisfying the /// existing ordering. For example, if we have the equivalent orderings /// `[a ASC, b ASC]` and `[c DESC]`, with `exprs` containing `[c, b, a, d]`, @@ -1474,7 +1550,7 @@ impl EquivalenceProperties { transformed_expr, &dependencies, &self.schema, - &ColumnRanges::default(), + &ColumnRangeCache::new(self.column_ranges.as_deref()), ) .map(|properties| properties.sort_properties), Ok(SortProperties::Ordered(actual)) if actual == options @@ -1615,7 +1691,7 @@ fn expr_refers( .any(|child| expr_refers(child, referred_expr)) } -/// Unbounded ranges of the columns of one schema, built on first use. +/// Column ranges for one schema, using trusted bounds or built on first use. /// /// Building the unbounded range of a nested type materializes a null array of that type, so /// a wide struct referenced by many projection sources is costly to rebuild per reference. @@ -1623,10 +1699,20 @@ fn expr_refers( /// expressions and failures are computed as before and never cached. Cloning a cached range /// clones its two endpoints, which for nested types share their arrays. #[derive(Default)] -struct ColumnRanges(std::cell::RefCell>); +struct ColumnRangeCache<'a> { + bounds: Option<&'a ColumnRanges>, + ranges: std::cell::RefCell>, +} -impl ColumnRanges { - fn unbounded( +impl<'a> ColumnRangeCache<'a> { + fn new(bounds: Option<&'a ColumnRanges>) -> Self { + Self { + bounds, + ranges: std::cell::RefCell::default(), + } + } + + fn range( &self, expr: &Arc, schema: &SchemaRef, @@ -1634,18 +1720,19 @@ impl ColumnRanges { let Some(column) = expr.downcast_ref::() else { return Interval::make_unbounded(&expr.data_type(schema)?); }; - if let Some(range) = self.0.borrow().get(&column.index()) { + if let Some(range) = self.ranges.borrow().get(&column.index()) { return Ok(range.clone()); } - let range = Interval::make_unbounded(&expr.data_type(schema)?)?; - self.0.borrow_mut().insert(column.index(), range.clone()); + let range = column_range(expr, schema, self.bounds)?; + self.ranges + .borrow_mut() + .insert(column.index(), range.clone()); Ok(range) } } /// This function examines the given expression and its properties to determine -/// the ordering properties of the expression. The range knowledge is not utilized -/// yet in the scope of this function. +/// the ordering properties of the expression, using trusted column ranges. /// /// # Parameters /// @@ -1662,7 +1749,7 @@ fn get_expr_properties( expr: &Arc, dependencies: &Dependencies, schema: &SchemaRef, - column_ranges: &ColumnRanges, + column_ranges: &ColumnRangeCache, ) -> Result { if let Some(column_order) = dependencies.iter().find(|&order| expr.eq(&order.expr)) { // If exact match is found, return its ordering. This is a base case @@ -1673,16 +1760,33 @@ fn get_expr_properties( // one-to-one in its own inputs (e.g. `floor(x)` as a sort key), and // it does not need to: parent expressions are substituted for this // sort key, so their strictness only has to be relative to it. + let range = if expr.is::() || column_ranges.bounds.is_none() { + column_ranges.range(expr, schema)? + } else { + // Ordering is already known; infer bounds without matching dependencies. + let range = get_expr_properties( + expr, + &Dependencies::default(), + schema, + column_ranges, + ) + .map(|properties| properties.range) + .ok(); + match range { + Some(range) if range.data_type() == expr.data_type(schema)? => range, + _ => Interval::make_unbounded(&expr.data_type(schema)?)?, + } + }; Ok(ExprProperties { sort_properties: SortProperties::Ordered(column_order.options), - range: column_ranges.unbounded(expr, schema)?, + range, preserves_lex_ordering: false, strictly_order_preserving: true, }) } else if expr.downcast_ref::().is_some() { Ok(ExprProperties { sort_properties: SortProperties::Unordered, - range: column_ranges.unbounded(expr, schema)?, + range: column_ranges.range(expr, schema)?, preserves_lex_ordering: false, // A base case of the recursion: a column is the identity mapping // of itself, which is trivially one-to-one. @@ -1708,6 +1812,39 @@ fn get_expr_properties( } } +fn column_range( + expr: &Arc, + schema: &SchemaRef, + column_ranges: Option<&ColumnRanges>, +) -> Result { + let data_type = expr.data_type(schema)?; + if let Some(column) = expr.downcast_ref::() + && let Some(range) = column_ranges.and_then(|ranges| ranges.get(&column.index())) + && range.data_type() == data_type + { + return Ok(range.clone()); + } + Interval::make_unbounded(&data_type) +} + +fn integer_range( + data_type: &DataType, + statistics: &ColumnStatistics, +) -> Option { + if data_type.is_integer() + && let (Precision::Exact(min), Precision::Exact(max)) = + (&statistics.min_value, &statistics.max_value) + && !min.is_null() + && !max.is_null() + && &min.data_type() == data_type + && &max.data_type() == data_type + && min <= max + { + return Interval::try_new(min.clone(), max.clone()).ok(); + } + None +} + #[cfg(test)] mod tests { use super::*; diff --git a/datafusion/physical-expr/src/equivalence/properties/union.rs b/datafusion/physical-expr/src/equivalence/properties/union.rs index a3d5c1c454a47..b82b5ddef7d96 100644 --- a/datafusion/physical-expr/src/equivalence/properties/union.rs +++ b/datafusion/physical-expr/src/equivalence/properties/union.rs @@ -5,7 +5,7 @@ use std::iter::Peekable; use std::sync::Arc; -use super::EquivalenceProperties; +use super::{ColumnRanges, EquivalenceProperties}; use crate::equivalence::GroupingEquivalenceClass; use crate::equivalence::class::AcrossPartitions; use crate::{ConstExpr, PhysicalSortExpr}; @@ -61,7 +61,25 @@ fn calculate_union_binary( orderings.add_satisfied_orderings(rhs, &lhs)?; let orderings = orderings.build(); + // A union's bounds must cover every branch, including branches with no bounds. + let column_ranges = lhs + .column_ranges + .as_ref() + .zip(rhs.column_ranges.as_ref()) + .map(|(left, right)| { + left.iter() + .filter_map(|(index, left)| { + right + .get(index) + .and_then(|right| left.union(right).ok()) + .map(|range| (*index, range)) + }) + .collect::() + }) + .filter(|ranges| !ranges.is_empty()) + .map(Arc::new); let mut eq_properties = EquivalenceProperties::new(lhs.schema); + eq_properties.column_ranges = column_ranges; eq_properties.add_constants(constants)?; eq_properties.add_orderings(orderings); Ok(eq_properties) diff --git a/datafusion/physical-expr/src/equivalence/properties/walk.rs b/datafusion/physical-expr/src/equivalence/properties/walk.rs index e0487cdd23dfa..1c1779e88bb55 100644 --- a/datafusion/physical-expr/src/equivalence/properties/walk.rs +++ b/datafusion/physical-expr/src/equivalence/properties/walk.rs @@ -4,7 +4,7 @@ use std::sync::Arc; -use super::EquivalenceProperties; +use super::{EquivalenceProperties, column_range}; use crate::equivalence::ordering::is_partial_const; use crate::expressions::{Column, Literal}; use crate::{LexOrdering, PhysicalExpr, PhysicalSortExpr}; @@ -216,8 +216,8 @@ impl<'a> SortKeyWalk<'a> { /// /// - An intermediate node takes its order and range from its children, /// through [`PhysicalExpr::get_properties`]. A `Literal` leaf is a - /// singleton with a point range. A `Column` leaf is unordered, with an - /// unbounded range of its data type. + /// singleton with a point range. A `Column` leaf is unordered, with its + /// trusted range or an unbounded range of its data type. /// - Then, whatever its children said: a node that is fixed or partially /// constant is a singleton, and a node that heads an ordering is ordered /// like the first ordering it heads. @@ -244,8 +244,11 @@ impl<'a> SortKeyWalk<'a> { node.data = node.expr.get_properties(&[])?; } else if node.expr.is::() { // We have a Column, which is the other possible leaf node type: - node.data.range = - Interval::make_unbounded(&node.expr.data_type(self.props.schema())?)?; + node.data.range = column_range( + &node.expr, + self.props.schema(), + self.props.column_ranges.as_deref(), + )?; // A column is the identity mapping of itself, which is trivially // strict: node.data.strictly_order_preserving = true; @@ -322,7 +325,8 @@ mod tests { use arrow::compute::SortOptions; use arrow::datatypes::{DataType, Field, Schema, TimeUnit}; use datafusion_common::config::ConfigOptions; - use datafusion_common::{Result, ScalarValue}; + use datafusion_common::stats::Precision; + use datafusion_common::{ColumnStatistics, Result, ScalarValue}; use datafusion_expr::physical_planning_context::{ ScalarSubqueryResults, SubqueryIndex, }; @@ -360,6 +364,42 @@ mod tests { Ok(()) } + /// Exact bounds allow arithmetic keys to preserve ordering after an + /// earlier key is pinned. Potential overflow still prevents inference. + #[test] + fn bounded_arithmetic_after_pinning() -> Result<()> { + let schema = Arc::new(Schema::new( + ["k", "a"] + .map(|name| Field::new(name, DataType::Int32, true)) + .to_vec(), + )); + let [k, a] = ["k", "a"].map(|name| col(name, &schema).unwrap()); + let a_plus_one = + Arc::new(BinaryExpr::new(Arc::clone(&a), Operator::Plus, lit(1i32))) + as Arc; + let sort = |expr: &Arc| { + PhysicalSortExpr::new_default(Arc::clone(expr)) + }; + + for max in [3, i32::MAX] { + let statistics = [ + ColumnStatistics::new_unknown(), + ColumnStatistics { + min_value: Precision::Exact(ScalarValue::Int32(Some(1))), + max_value: Precision::Exact(ScalarValue::Int32(Some(max))), + ..ColumnStatistics::new_unknown() + }, + ]; + let mut properties = EquivalenceProperties::new(Arc::clone(&schema)); + properties.add_ordering([sort(&k), sort(&a)]); + let required = [sort(&k), sort(&a_plus_one)]; + assert!(!properties.ordering_satisfy(required.clone())?); + let properties = properties.with_column_statistics(&statistics); + assert_eq!(properties.ordering_satisfy(required)?, max == 3); + } + Ok(()) + } + /// An ordering found after a pin can imply another one. With ordering /// `[k, x, y, z]`, `x = CAST(y AS BIGINT)` and `y = CAST(z AS INT)`, /// pinning `k` leaves `[x, y, z]`, which implies `[y, z]`, which implies diff --git a/datafusion/proto/tests/cases/plans/sources.rs b/datafusion/proto/tests/cases/plans/sources.rs index 32909afddf83b..7ac1d622d4ec3 100644 --- a/datafusion/proto/tests/cases/plans/sources.rs +++ b/datafusion/proto/tests/cases/plans/sources.rs @@ -39,7 +39,9 @@ use datafusion::prelude::SessionContext; use datafusion::scalar::ScalarValue; use datafusion_common::config::TableParquetOptions; use datafusion_common::stats::Precision; -use datafusion_common::{DataFusionError, Result, internal_datafusion_err, internal_err}; +use datafusion_common::{ + ColumnStatistics, DataFusionError, Result, internal_datafusion_err, internal_err, +}; use datafusion_datasource::file_compression_type::FileCompressionType; use datafusion_datasource::{TableSchema, TableSchemaBuilder}; use datafusion_expr::ColumnarValue; @@ -95,12 +97,12 @@ fn parquet_schema_provider_requires_extension_codec() -> Result<()> { #[test] fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> { let file_schema = - Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)])); + Arc::new(Schema::new(vec![Field::new("col", DataType::Int64, false)])); let predicate = Arc::new(BinaryExpr::new( Arc::new(Column::new("col", 0)), Operator::Eq, - lit("1"), + lit(1_i64), )); let mut options = TableParquetOptions::new(); @@ -125,9 +127,13 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> { 1024, )])]) .with_statistics(Statistics { - num_rows: Precision::Inexact(100), - total_byte_size: Precision::Inexact(1024), - column_statistics: Statistics::unknown_column(&file_schema), + num_rows: Precision::Exact(100), + total_byte_size: Precision::Exact(1024), + column_statistics: vec![ + ColumnStatistics::new_unknown() + .with_min_value(Precision::Exact(ScalarValue::from(1_i64))) + .with_max_value(Precision::Exact(ScalarValue::from(100_i64))), + ], }) .build(); diff --git a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt index 27303d6cc5022..599ac6f8874a2 100644 --- a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt @@ -542,13 +542,288 @@ DROP TABLE nulls_first_bad; statement ok DROP TABLE nulls_first_ok; -# Config reset - # The SLT runner sets `target_partitions` to 4 instead of using the default, so # reset it explicitly. statement ok set datafusion.execution.target_partitions = 4; +# Exact integer bounds prove that these projections cannot wrap (#25220). +statement ok +SET datafusion.explain.physical_plan_only = true; + +statement ok +CREATE TABLE arithmetic_bounds_source (a BIGINT, b BIGINT, u BIGINT UNSIGNED) AS VALUES +(1, 9223372036854775805, 0), +(2, 9223372036854775806, 1), +(3, 9223372036854775807, 2), +(NULL, NULL, NULL); + +query I +COPY (SELECT * FROM arithmetic_bounds_source ORDER BY a) +TO 'test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet' +STORED AS PARQUET; +---- +4 + +statement ok +CREATE EXTERNAL TABLE arithmetic_bounds +STORED AS PARQUET +WITH ORDER (a ASC NULLS LAST) +WITH ORDER (b ASC NULLS LAST) +WITH ORDER (u ASC NULLS LAST) +LOCATION 'test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet'; + +query TT +EXPLAIN SELECT 1 - a AS x FROM arithmetic_bounds ORDER BY x DESC NULLS LAST; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[1 - a@0 as x], file_type=parquet + +query I +SELECT 1 - a AS x FROM arithmetic_bounds ORDER BY x DESC NULLS LAST; +---- +0 +-1 +-2 +NULL + +# Window outputs retain the input bounds needed by the outer projection. +query TT +EXPLAIN SELECT a + 1 AS x, rn +FROM (SELECT a, row_number() OVER (ORDER BY a) AS rn FROM arithmetic_bounds) +ORDER BY x; +---- +physical_plan +01)ProjectionExec: expr=[a@0 + 1 as x, row_number() ORDER BY [arithmetic_bounds.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@1 as rn] +02)--BoundedWindowAggExec: wdw=[row_number() ORDER BY [arithmetic_bounds.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() ORDER BY [arithmetic_bounds.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +query II +SELECT a + 1 AS x, rn +FROM (SELECT a, row_number() OVER (ORDER BY a) AS rn FROM arithmetic_bounds) +ORDER BY x; +---- +2 1 +3 2 +4 3 +NULL 4 + +# Computed bounds survive consecutive projections and column reordering. +query TT +EXPLAIN SELECT u, a + 1 AS x, (a + 1) - 2 AS y FROM arithmetic_bounds ORDER BY y; +---- +physical_plan +01)ProjectionExec: expr=[u@1 as u, __common_expr_1@0 as x, __common_expr_1@0 - 2 as y] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a@0 + 1 as __common_expr_1, u], output_ordering=[u@1 ASC NULLS LAST], file_type=parquet + +query III +SELECT u, a + 1 AS x, (a + 1) - 2 AS y FROM arithmetic_bounds ORDER BY y; +---- +0 2 0 +1 3 1 +2 4 2 +NULL NULL NULL + +# Reordered bounds must still detect signed overflow. +query TT +EXPLAIN SELECT a, b - 1 AS x, (b - 1) + 2 AS y FROM arithmetic_bounds ORDER BY y; +---- +physical_plan +01)SortExec: expr=[y@2 ASC NULLS LAST], preserve_partitioning=[false] +02)--ProjectionExec: expr=[a@1 as a, __common_expr_1@0 as x, __common_expr_1@0 + 2 as y] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[b@1 - 1 as __common_expr_1, a], output_ordering=[a@1 ASC NULLS LAST], file_type=parquet + +query III +SELECT a, b - 1 AS x, (b - 1) + 2 AS y FROM arithmetic_bounds ORDER BY y; +---- +3 9223372036854775806 -9223372036854775808 +1 9223372036854775804 9223372036854775806 +2 9223372036854775805 9223372036854775807 +NULL NULL NULL + +statement ok +CREATE EXTERNAL TABLE expression_ordered_bounds +STORED AS PARQUET +WITH ORDER (a + 1 ASC NULLS LAST) +LOCATION 'test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet'; + +query TT +EXPLAIN SELECT (a + 1) - 2 AS x FROM expression_ordered_bounds ORDER BY x; +---- +physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a@0 + 1 - 2 as x], file_type=parquet + +statement ok +DROP TABLE expression_ordered_bounds; + +# The same range information is available to grouping-expression analysis. +query TT +EXPLAIN SELECT a + 1 AS x, count(*) FROM arithmetic_bounds GROUP BY x ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--ProjectionExec: expr=[arithmetic_bounds.a + Int64(1)@0 as x, count(Int64(1))@1 as count(*)] +03)----AggregateExec: mode=FinalPartitioned, gby=[arithmetic_bounds.a + Int64(1)@0 as arithmetic_bounds.a + Int64(1)], aggr=[count(Int64(1))], group_clustering_mode=Full +04)------RepartitionExec: partitioning=Hash([arithmetic_bounds.a + Int64(1)@0], 4), input_partitions=1, maintains_sort_order=true +05)--------AggregateExec: mode=Partial, gby=[a@0 + 1 as arithmetic_bounds.a + Int64(1)], aggr=[count(Int64(1))], group_clustering_mode=Full +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +# Exact extrema do not justify removing a sort when the arithmetic wraps. +query TT +EXPLAIN SELECT u - 1::BIGINT UNSIGNED AS x FROM arithmetic_bounds ORDER BY x; +---- +physical_plan +01)SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[u@2 - 1 as x], file_type=parquet + +query I +SELECT u - 1::BIGINT UNSIGNED AS x FROM arithmetic_bounds ORDER BY x; +---- +0 +1 +18446744073709551615 +NULL + +# Filtering preserves the original bounds even when result statistics are estimates. +query TT +EXPLAIN SELECT a + 1 AS x FROM arithmetic_bounds WHERE a % 2 = 1 ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--ProjectionExec: expr=[a@0 + 1 as x] +03)----FilterExec: a@0 % 2 = 1 +04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true +05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet, predicate=a@0 % 2 = 1 + +# Bounds alone do not establish an ordering. +statement ok +CREATE EXTERNAL TABLE arithmetic_unordered +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_sorted_statistics/cast_bounds/2.parquet'; + +query TT +EXPLAIN SELECT a + 1 AS x FROM arithmetic_unordered ORDER BY x; +---- +physical_plan +01)SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[false] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/cast_bounds/2.parquet]]}, projection=[a@0 + 1 as x], file_type=parquet + +statement ok +DROP TABLE arithmetic_unordered; + +# Late dynamic filters must preserve ordering derived from file bounds. +statement ok +CREATE EXTERNAL TABLE dynamic_bounds (a BIGINT) +STORED AS PARQUET +WITH ORDER (a ASC NULLS LAST) +LOCATION 'test_files/scratch/parquet_sorted_statistics/cast_bounds'; + +query I +SELECT p.x FROM (VALUES (2), (4)) AS s(b) +JOIN (SELECT a + 1 AS x FROM dynamic_bounds) p ON p.x = s.b +ORDER BY p.x; +---- +2 +4 + +statement ok +DROP TABLE dynamic_bounds; + +# A known ordering without column statistics still needs the sort. +statement ok +SET datafusion.execution.collect_statistics = false; + +query TT +EXPLAIN SELECT a + 1 AS x FROM arithmetic_bounds ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[true] +03)----ProjectionExec: expr=[a@0 + 1 as x] +04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true +05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +statement ok +SET datafusion.execution.collect_statistics = true; + +# Union bounds cover every branch. A missing range is not a bounded branch. +query TT +EXPLAIN SELECT a + 1 AS x +FROM (SELECT a FROM arithmetic_bounds UNION ALL SELECT a FROM arithmetic_bounds) t +ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--ProjectionExec: expr=[a@0 + 1 as x] +03)----UnionExec +04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet +05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +query TT +EXPLAIN SELECT a + 1 AS x +FROM (SELECT a FROM arithmetic_bounds UNION ALL SELECT b AS a FROM arithmetic_bounds) t +ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[true] +03)----ProjectionExec: expr=[a@0 + 1 as x] +04)------UnionExec +05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet +06)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[b@1 as a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +query I +COPY (SELECT b AS a FROM arithmetic_bounds_source ORDER BY b) +TO 'test_files/scratch/parquet_sorted_statistics/bounds_missing.parquet' +STORED AS PARQUET OPTIONS ('format.statistics_enabled' 'none'); +---- +4 + +statement ok +CREATE EXTERNAL TABLE bounds_missing +STORED AS PARQUET +WITH ORDER (a ASC NULLS LAST) +LOCATION 'test_files/scratch/parquet_sorted_statistics/bounds_missing.parquet'; + +query TT +EXPLAIN SELECT a + 1 AS x +FROM (SELECT a FROM arithmetic_bounds UNION ALL SELECT a FROM bounds_missing) t +ORDER BY x; +---- +physical_plan +01)SortPreservingMergeExec: [x@0 ASC NULLS LAST] +02)--SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[true] +03)----ProjectionExec: expr=[a@0 + 1 as x] +04)------UnionExec +05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/arithmetic_bounds.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet +06)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/bounds_missing.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet + +query I +SELECT a + 1 AS x +FROM (SELECT a FROM arithmetic_bounds UNION ALL SELECT a FROM bounds_missing) t +ORDER BY x; +---- +-9223372036854775808 +2 +3 +4 +9223372036854775806 +9223372036854775807 +NULL +NULL + +statement ok +DROP TABLE bounds_missing; + +statement ok +DROP TABLE arithmetic_bounds; + +statement ok +DROP TABLE arithmetic_bounds_source; + +statement ok +RESET datafusion.explain.physical_plan_only; + +# Config reset + statement ok reset datafusion.execution.collect_statistics;