From 3d5c3b0becf299c1e9b141a981d9819c0eddc021 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 26 Oct 2026 23:19:44 +0800 Subject: [PATCH 1/3] perf: use exact integer statistics for ordering inference --- .../datasource/src/file_scan_config/mod.rs | 4 +- .../datasource/src/file_scan_config/proto.rs | 10 +- .../src/equivalence/properties/mod.rs | 162 +++++++++- .../src/equivalence/properties/union.rs | 20 +- .../test_files/parquet_sorted_statistics.slt | 302 ++++++++++++++++++ 5 files changed, 474 insertions(+), 24 deletions(-) diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index c94979a472969..cb1b8682dc4d1 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -961,7 +961,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 706784a170688..8d100278f14aa 100644 --- a/datafusion/datasource/src/file_scan_config/proto.rs +++ b/datafusion/datasource/src/file_scan_config/proto.rs @@ -84,9 +84,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; @@ -152,11 +150,11 @@ impl FileScanConfig { (*file_compression_type.get_variant()).into(); compression as i32 }); - let statistics = self.statistics(); - + // Preserve original exact bounds used by ordering inference. The decoded + // scan applies the filter precision policy when statistics are requested. 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/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index ed824e5712f96..0a3d8e5606f6c 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -40,9 +40,12 @@ use crate::{ }; use arrow::compute::SortOptions; -use arrow::datatypes::SchemaRef; +use arrow::datatypes::{DataType, SchemaRef}; +use datafusion_common::stats::Precision; use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; -use datafusion_common::{Constraint, Constraints, HashMap, Result, plan_err}; +use datafusion_common::{ + ColumnStatistics, Constraint, Constraints, HashMap, Result, plan_err, +}; use datafusion_expr::interval_arithmetic::Interval; use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use datafusion_physical_expr_common::sort_expr::options_compatible; @@ -147,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 @@ -243,7 +248,8 @@ impl EquivalenceProperties { return None; } let dependencies = Dependencies::new(std::iter::once(sort_expr.clone())); - let properties = get_expr_properties(&r_expr, &dependencies, schema).ok()?; + let properties = + get_expr_properties(&r_expr, &dependencies, schema, None).ok()?; (properties.strictly_order_preserving && properties.sort_properties == SortProperties::Ordered(sort_expr.options)) .then(|| PhysicalSortExpr::new(r_expr, sort_expr.options)) @@ -257,6 +263,7 @@ impl EquivalenceProperties { oeq_cache: OrderingEquivalenceCache::default(), constraints: Constraints::default(), schema, + column_ranges: None, } } @@ -290,9 +297,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 @@ -338,6 +364,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), + } + } Ok(self) } @@ -510,10 +543,13 @@ impl EquivalenceProperties { if !next.expr.eq(child) { break 'exprs; } - let data_type = child.data_type(&self.schema)?; child_properties.push(ExprProperties { sort_properties: SortProperties::Ordered(next.options), - range: Interval::make_unbounded(&data_type)?, + range: column_range( + child, + &self.schema, + self.column_ranges.as_deref(), + )?, preserves_lex_ordering: true, strictly_order_preserving: true, }); @@ -1126,7 +1162,12 @@ impl EquivalenceProperties { referred_dependencies(&dependency_map, source) .into_iter() .filter_map(|deps| { - let ep = get_expr_properties(source, &deps, &self.schema); + let ep = get_expr_properties( + source, + &deps, + &self.schema, + self.column_ranges.as_deref(), + ); let sort_properties = ep.map(|prop| prop.sort_properties); if let Ok(SortProperties::Ordered(options)) = sort_properties { Some((options, deps)) @@ -1278,6 +1319,7 @@ impl EquivalenceProperties { output_schema: SchemaRef, eq_group: EquivalenceGroup, ) -> Self { + let column_ranges = self.projected_column_ranges(mapping, &output_schema); let orderings = self.projected_orderings(mapping, self.oeq_cache.normal_cls.clone()); let normal_orderings = orderings @@ -1290,9 +1332,43 @@ impl EquivalenceProperties { constraints: self.projected_constraints(mapping).unwrap_or_default(), schema: output_schema, eq_group, + column_ranges, } } + fn projected_column_ranges( + &self, + mapping: &ProjectionMapping, + output_schema: &SchemaRef, + ) -> Option>> { + self.column_ranges.as_ref()?; + let mut ranges = HashMap::new(); + let dependencies = Dependencies::default(); + for (source, targets) in mapping.iter() { + let Ok(properties) = get_expr_properties( + source, + &dependencies, + &self.schema, + self.column_ranges.as_deref(), + ) 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]`, @@ -1426,7 +1502,7 @@ impl EquivalenceProperties { options, ))); matches!( - get_expr_properties(transformed_expr, &dependencies, &self.schema) + get_expr_properties(transformed_expr, &dependencies, &self.schema, self.column_ranges.as_deref()) .map(|properties| properties.sort_properties), Ok(SortProperties::Ordered(actual)) if actual == options ) @@ -1546,8 +1622,7 @@ impl Display for EquivalenceProperties { /// /// Range information is calculated as: /// - If it is a `Literal` node, we set the range as a point value. If it is a -/// `Column` node, we set the datatype of the range, but cannot give an interval -/// for the range, yet. +/// `Column` node, we use its trusted range or fall back to a typed unbounded range. /// - If it is an intermediate node, the children states matter. Each `PhysicalExpr` /// and operator has its own rules on how to propagate the children range. fn update_properties( @@ -1564,8 +1639,11 @@ fn update_properties( 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(eq_properties.schema())?)?; + node.data.range = column_range( + &node.expr, + eq_properties.schema(), + eq_properties.column_ranges.as_deref(), + )?; // A column is the identity mapping of itself, which is trivially // strict: node.data.strictly_order_preserving = true; @@ -1610,8 +1688,7 @@ fn expr_refers( } /// 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 /// @@ -1628,6 +1705,7 @@ fn get_expr_properties( expr: &Arc, dependencies: &Dependencies, schema: &SchemaRef, + column_ranges: Option<&HashMap>, ) -> 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 @@ -1638,16 +1716,35 @@ 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.is_none() { + column_range(expr, schema, column_ranges)? + } else { + let children = expr + .children() + .iter() + .map(|child| { + get_expr_properties(child, dependencies, schema, column_ranges) + }) + .collect::>>(); + let range = children + .and_then(|children| expr.get_properties(&children)) + .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: Interval::make_unbounded(&expr.data_type(schema)?)?, + range, preserves_lex_ordering: false, strictly_order_preserving: true, }) } else if expr.downcast_ref::().is_some() { Ok(ExprProperties { sort_properties: SortProperties::Unordered, - range: Interval::make_unbounded(&expr.data_type(schema)?)?, + range: column_range(expr, schema, column_ranges)?, preserves_lex_ordering: false, // A base case of the recursion: a column is the identity mapping // of itself, which is trivially one-to-one. @@ -1666,13 +1763,46 @@ fn get_expr_properties( let child_states = expr .children() .iter() - .map(|child| get_expr_properties(child, dependencies, schema)) + .map(|child| get_expr_properties(child, dependencies, schema, column_ranges)) .collect::>>()?; // Calculate expression ordering using ordering of its children. expr.get_properties(&child_states) } } +fn column_range( + expr: &Arc, + schema: &SchemaRef, + column_ranges: Option<&HashMap>, +) -> 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 ea4094e75159a..0b11efad6d7d9 100644 --- a/datafusion/physical-expr/src/equivalence/properties/union.rs +++ b/datafusion/physical-expr/src/equivalence/properties/union.rs @@ -23,7 +23,7 @@ use crate::equivalence::class::AcrossPartitions; use crate::{ConstExpr, PhysicalSortExpr}; use arrow::datatypes::SchemaRef; -use datafusion_common::{Result, internal_err}; +use datafusion_common::{HashMap, Result, internal_err}; use datafusion_physical_expr_common::sort_expr::LexOrdering; /// Computes the union (in the sense of `UnionExec`) `EquivalenceProperties` @@ -74,7 +74,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/sqllogictest/test_files/parquet_sorted_statistics.slt b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt index 6da368998865e..9a5229ae76e1e 100644 --- a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt @@ -555,6 +555,308 @@ DROP TABLE nulls_first_bad; statement ok DROP TABLE nulls_first_ok; +# Exact integer bounds prove that these projections cannot wrap (#25220). +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.explain.physical_plan_only = true; + +statement ok +SET datafusion.optimizer.enable_sort_pushdown = false; + +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) +WHERE rn <= 2 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)--FilterExec: row_number() ORDER BY [arithmetic_bounds.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@1 <= 2 +03)----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] +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 + +query II +SELECT a + 1 AS x, rn +FROM (SELECT a, row_number() OVER (ORDER BY a) AS rn FROM arithmetic_bounds) +WHERE rn <= 2 ORDER BY x; +---- +2 1 +3 2 + +# 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)ProjectionExec: expr=[arithmetic_bounds.a + Int64(1)@0 as x, count(Int64(1))@1 as count(*)] +02)--AggregateExec: mode=Single, gby=[a@0 + 1 as arithmetic_bounds.a + Int64(1)], aggr=[count(Int64(1))], group_clustering_mode=Full +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 + +# 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)ProjectionExec: expr=[a@0 + 1 as x] +02)--FilterExec: a@0 % 2 = 1 +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, predicate=a@0 % 2 = 1 + +# Bounds alone do not establish an ordering. +query I +COPY (SELECT * FROM (VALUES (3), (1), (2)) AS t(a)) +TO 'test_files/scratch/parquet_sorted_statistics/arithmetic_unordered.parquet' +STORED AS PARQUET; +---- +3 + +statement ok +CREATE EXTERNAL TABLE arithmetic_unordered +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_sorted_statistics/arithmetic_unordered.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/arithmetic_unordered.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 +SET datafusion.execution.target_partitions = 2; + +query I +COPY (SELECT * FROM (VALUES (1), (2), (3)) AS t(a) ORDER BY a) +TO 'test_files/scratch/parquet_sorted_statistics/dynamic_bounds/1.parquet' +STORED AS PARQUET; +---- +3 + +query I +COPY (SELECT * FROM (VALUES (4), (5), (6)) AS t(a) ORDER BY a) +TO 'test_files/scratch/parquet_sorted_statistics/dynamic_bounds/2.parquet' +STORED AS PARQUET; +---- +3 + +statement ok +CREATE EXTERNAL TABLE dynamic_bounds +STORED AS PARQUET +WITH ORDER (a ASC NULLS LAST) +LOCATION 'test_files/scratch/parquet_sorted_statistics/dynamic_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; + +statement ok +SET datafusion.execution.target_partitions = 1; + +# 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)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=[a@0 + 1 as x], 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; + +statement ok +RESET datafusion.optimizer.enable_sort_pushdown; + # Config reset # The SLT runner sets `target_partitions` to 4 instead of using the default, so From 4be1c1b0adb83a1800421a100a32eb3ca5ae282f Mon Sep 17 00:00:00 2001 From: Huaijin Date: Wed, 28 Oct 2026 00:43:55 +0800 Subject: [PATCH 2/3] fix: stabilize range diagnostics and simplify ordering tests --- .../datasource/src/file_scan_config/mod.rs | 9 +- .../datasource/src/file_scan_config/proto.rs | 4 +- .../src/equivalence/properties/mod.rs | 41 ++++++--- datafusion/proto/tests/cases/plans/sources.rs | 18 ++-- .../test_files/parquet_sorted_statistics.slt | 87 +++++++------------ 5 files changed, 78 insertions(+), 81 deletions(-) diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index cb1b8682dc4d1..69ac3dc3d6eb8 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -211,15 +211,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. /// diff --git a/datafusion/datasource/src/file_scan_config/proto.rs b/datafusion/datasource/src/file_scan_config/proto.rs index 8d100278f14aa..0299ba3aeef39 100644 --- a/datafusion/datasource/src/file_scan_config/proto.rs +++ b/datafusion/datasource/src/file_scan_config/proto.rs @@ -150,8 +150,8 @@ impl FileScanConfig { (*file_compression_type.get_variant()).into(); compression as i32 }); - // Preserve original exact bounds used by ordering inference. The decoded - // scan applies the filter precision policy when statistics are requested. + // 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()), diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index 0a3d8e5606f6c..7a59744bad9fc 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -22,6 +22,7 @@ mod union; // Submodule containing calculate_union pub use joins::*; pub use union::*; +use std::collections::BTreeMap; use std::fmt::{self, Display}; use std::mem; use std::sync::Arc; @@ -136,7 +137,7 @@ use itertools::Itertools; /// "order: [[a@0 ASC, c@2 DESC]], eq: [{members: [b@1], constant: (heterogeneous)}]" /// ); /// ``` -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct EquivalenceProperties { /// Distinct equivalence classes (i.e. expressions with the same value). eq_group: EquivalenceGroup, @@ -154,6 +155,24 @@ pub struct EquivalenceProperties { column_ranges: Option>>, } +impl fmt::Debug for EquivalenceProperties { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Keep range diagnostics stable when properties are reconstructed. + let column_ranges = self + .column_ranges + .as_ref() + .map(|ranges| ranges.iter().collect::>()); + f.debug_struct("EquivalenceProperties") + .field("eq_group", &self.eq_group) + .field("oeq_class", &self.oeq_class) + .field("oeq_cache", &self.oeq_cache) + .field("constraints", &self.constraints) + .field("schema", &self.schema) + .field("column_ranges", &column_ranges) + .finish() + } +} + /// This object serves as a cache for storing equivalent sort expressions /// in normal form, and a map associating leading sort expressions with /// full lexicographical orderings. With this information, DataFusion can @@ -1719,17 +1738,15 @@ fn get_expr_properties( let range = if expr.is::() || column_ranges.is_none() { column_range(expr, schema, column_ranges)? } else { - let children = expr - .children() - .iter() - .map(|child| { - get_expr_properties(child, dependencies, schema, column_ranges) - }) - .collect::>>(); - let range = children - .and_then(|children| expr.get_properties(&children)) - .map(|properties| properties.range) - .ok(); + // 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)?)?, diff --git a/datafusion/proto/tests/cases/plans/sources.rs b/datafusion/proto/tests/cases/plans/sources.rs index 40821583678d1..b06d605284a01 100644 --- a/datafusion/proto/tests/cases/plans/sources.rs +++ b/datafusion/proto/tests/cases/plans/sources.rs @@ -52,7 +52,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; @@ -108,12 +110,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(); @@ -138,9 +140,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 9a5229ae76e1e..8c7dec49f09d4 100644 --- a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt @@ -555,16 +555,15 @@ DROP TABLE nulls_first_bad; statement ok DROP TABLE nulls_first_ok; -# Exact integer bounds prove that these projections cannot wrap (#25220). +# The SLT runner sets `target_partitions` to 4 instead of using the default, so +# reset it explicitly. statement ok -SET datafusion.execution.target_partitions = 1; +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 -SET datafusion.optimizer.enable_sort_pushdown = false; - statement ok CREATE TABLE arithmetic_bounds_source (a BIGINT, b BIGINT, u BIGINT UNSIGNED) AS VALUES (1, 9223372036854775805, 0), @@ -604,21 +603,22 @@ NULL query TT EXPLAIN SELECT a + 1 AS x, rn FROM (SELECT a, row_number() OVER (ORDER BY a) AS rn FROM arithmetic_bounds) -WHERE rn <= 2 ORDER BY x; +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)--FilterExec: row_number() ORDER BY [arithmetic_bounds.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@1 <= 2 -03)----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] -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 +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) -WHERE rn <= 2 ORDER BY x; +ORDER BY x; ---- 2 1 3 2 +4 3 +NULL 4 # Computed bounds survive consecutive projections and column reordering. query TT @@ -672,9 +672,12 @@ query TT EXPLAIN SELECT a + 1 AS x, count(*) FROM arithmetic_bounds GROUP BY x ORDER BY x; ---- physical_plan -01)ProjectionExec: expr=[arithmetic_bounds.a + Int64(1)@0 as x, count(Int64(1))@1 as count(*)] -02)--AggregateExec: mode=Single, gby=[a@0 + 1 as arithmetic_bounds.a + Int64(1)], aggr=[count(Int64(1))], group_clustering_mode=Full -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 +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 @@ -697,56 +700,34 @@ query TT EXPLAIN SELECT a + 1 AS x FROM arithmetic_bounds WHERE a % 2 = 1 ORDER BY x; ---- physical_plan -01)ProjectionExec: expr=[a@0 + 1 as x] -02)--FilterExec: a@0 % 2 = 1 -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, predicate=a@0 % 2 = 1 +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. -query I -COPY (SELECT * FROM (VALUES (3), (1), (2)) AS t(a)) -TO 'test_files/scratch/parquet_sorted_statistics/arithmetic_unordered.parquet' -STORED AS PARQUET; ----- -3 - statement ok CREATE EXTERNAL TABLE arithmetic_unordered STORED AS PARQUET -LOCATION 'test_files/scratch/parquet_sorted_statistics/arithmetic_unordered.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/arithmetic_unordered.parquet]]}, projection=[a@0 + 1 as x], file_type=parquet +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 -SET datafusion.execution.target_partitions = 2; - -query I -COPY (SELECT * FROM (VALUES (1), (2), (3)) AS t(a) ORDER BY a) -TO 'test_files/scratch/parquet_sorted_statistics/dynamic_bounds/1.parquet' -STORED AS PARQUET; ----- -3 - -query I -COPY (SELECT * FROM (VALUES (4), (5), (6)) AS t(a) ORDER BY a) -TO 'test_files/scratch/parquet_sorted_statistics/dynamic_bounds/2.parquet' -STORED AS PARQUET; ----- -3 - -statement ok -CREATE EXTERNAL TABLE dynamic_bounds +CREATE EXTERNAL TABLE dynamic_bounds (a BIGINT) STORED AS PARQUET WITH ORDER (a ASC NULLS LAST) -LOCATION 'test_files/scratch/parquet_sorted_statistics/dynamic_bounds'; +LOCATION 'test_files/scratch/parquet_sorted_statistics/cast_bounds'; query I SELECT p.x FROM (VALUES (2), (4)) AS s(b) @@ -759,9 +740,6 @@ ORDER BY p.x; statement ok DROP TABLE dynamic_bounds; -statement ok -SET datafusion.execution.target_partitions = 1; - # A known ordering without column statistics still needs the sort. statement ok SET datafusion.execution.collect_statistics = false; @@ -770,8 +748,11 @@ query TT EXPLAIN SELECT a + 1 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=[a@0 + 1 as x], file_type=parquet +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; @@ -854,16 +835,8 @@ DROP TABLE arithmetic_bounds_source; statement ok RESET datafusion.explain.physical_plan_only; -statement ok -RESET datafusion.optimizer.enable_sort_pushdown; - # 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; - statement ok reset datafusion.execution.collect_statistics; From 1009b7725f23a033492bbff105181b734ad15989 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Wed, 28 Oct 2026 09:46:27 +0800 Subject: [PATCH 3/3] refactor: use ordered column ranges for stable debug output --- .../src/equivalence/properties/mod.rs | 38 ++++++------------- .../src/equivalence/properties/union.rs | 6 +-- 2 files changed, 14 insertions(+), 30 deletions(-) diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index 7a59744bad9fc..d6d085aef5ef2 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -22,7 +22,6 @@ mod union; // Submodule containing calculate_union pub use joins::*; pub use union::*; -use std::collections::BTreeMap; use std::fmt::{self, Display}; use std::mem; use std::sync::Arc; @@ -52,9 +51,12 @@ use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use datafusion_physical_expr_common::sort_expr::options_compatible; use datafusion_physical_expr_common::utils::ExprPropertiesNode; -use indexmap::IndexSet; +use hashbrown::DefaultHashBuilder; +use indexmap::{IndexMap, IndexSet}; use itertools::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), @@ -137,7 +139,7 @@ use itertools::Itertools; /// "order: [[a@0 ASC, c@2 DESC]], eq: [{members: [b@1], constant: (heterogeneous)}]" /// ); /// ``` -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct EquivalenceProperties { /// Distinct equivalence classes (i.e. expressions with the same value). eq_group: EquivalenceGroup, @@ -152,25 +154,7 @@ pub struct EquivalenceProperties { /// Schema associated with this object. schema: SchemaRef, /// Trusted bounds for non-null integer values, independent of estimate precision. - column_ranges: Option>>, -} - -impl fmt::Debug for EquivalenceProperties { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - // Keep range diagnostics stable when properties are reconstructed. - let column_ranges = self - .column_ranges - .as_ref() - .map(|ranges| ranges.iter().collect::>()); - f.debug_struct("EquivalenceProperties") - .field("eq_group", &self.eq_group) - .field("oeq_class", &self.oeq_class) - .field("oeq_cache", &self.oeq_cache) - .field("constraints", &self.constraints) - .field("schema", &self.schema) - .field("column_ranges", &column_ranges) - .finish() - } + column_ranges: Option>, } /// This object serves as a cache for storing equivalent sort expressions @@ -333,7 +317,7 @@ impl EquivalenceProperties { .filter_map(|(index, (statistics, field))| { integer_range(field.data_type(), statistics).map(|range| (index, range)) }) - .collect::>(); + .collect::(); self.column_ranges = (!ranges.is_empty()).then(|| Arc::new(ranges)); self } @@ -1359,9 +1343,9 @@ impl EquivalenceProperties { &self, mapping: &ProjectionMapping, output_schema: &SchemaRef, - ) -> Option>> { + ) -> Option> { self.column_ranges.as_ref()?; - let mut ranges = HashMap::new(); + let mut ranges = ColumnRanges::default(); let dependencies = Dependencies::default(); for (source, targets) in mapping.iter() { let Ok(properties) = get_expr_properties( @@ -1724,7 +1708,7 @@ fn get_expr_properties( expr: &Arc, dependencies: &Dependencies, schema: &SchemaRef, - column_ranges: Option<&HashMap>, + column_ranges: Option<&ColumnRanges>, ) -> 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 @@ -1790,7 +1774,7 @@ fn get_expr_properties( fn column_range( expr: &Arc, schema: &SchemaRef, - column_ranges: Option<&HashMap>, + column_ranges: Option<&ColumnRanges>, ) -> Result { let data_type = expr.data_type(schema)?; if let Some(column) = expr.downcast_ref::() diff --git a/datafusion/physical-expr/src/equivalence/properties/union.rs b/datafusion/physical-expr/src/equivalence/properties/union.rs index 0b11efad6d7d9..f52aa730fc99a 100644 --- a/datafusion/physical-expr/src/equivalence/properties/union.rs +++ b/datafusion/physical-expr/src/equivalence/properties/union.rs @@ -18,12 +18,12 @@ use std::iter::Peekable; use std::sync::Arc; -use super::EquivalenceProperties; +use super::{ColumnRanges, EquivalenceProperties}; use crate::equivalence::class::AcrossPartitions; use crate::{ConstExpr, PhysicalSortExpr}; use arrow::datatypes::SchemaRef; -use datafusion_common::{HashMap, Result, internal_err}; +use datafusion_common::{Result, internal_err}; use datafusion_physical_expr_common::sort_expr::LexOrdering; /// Computes the union (in the sense of `UnionExec`) `EquivalenceProperties` @@ -87,7 +87,7 @@ fn calculate_union_binary( .and_then(|right| left.union(right).ok()) .map(|range| (*index, range)) }) - .collect::>() + .collect::() }) .filter(|ranges| !ranges.is_empty()) .map(Arc::new);