asolimando commented on code in PR #27148:
URL: https://github.com/apache/datafusion/pull/27148#discussion_r4513669441
##########
datafusion/physical-expr/src/intervals/utils.rs:
##########
@@ -44,9 +82,10 @@ pub fn check_support(expr: &Arc<dyn PhysicalExpr>, schema:
&SchemaRef) -> bool {
// interval endpoints to a type with a different ordering (e.g.
integers
// to strings) does not preserve the interval, so such expressions are
// left unanalyzed rather than producing bogus bounds.
- is_datatype_supported(cast.cast_type()) && check_support(cast.expr(),
schema)
+ is_type_supported(cast.cast_type())
Review Comment:
`CAST(d AS DOUBLE) > 0.05` is estimated at 1100 rows instead of 220 on
`main` (500 actual, reproducer in my comment in `statistics_helper`). Decimal
to `Float64` and narrowing decimal casts are lossy, so the bound is not
propagated back and the selectivity is 1.0.
I think this branch also requires
`can_propagate_cast_constraints(input_type, cast.cast_type())`, can you double
check?
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -389,7 +389,14 @@ impl FilterExec {
} else {
let null_rejecting_columns =
collect_null_rejecting_columns(predicate);
- if check_support(predicate, schema) {
+ if check_statistics_support(predicate, schema)
Review Comment:
On a 1100-row table (`i` INT in 0..99, `w` DECIMAL(38, 18) in 0..99, `d`
DECIMAL(15, 2) in 0.00..0.10, reproducer below), these predicates are estimated
at 1100 rows, while `main` gives the 20% default (220):
- `i < CAST(2.5 AS DECIMAL(5, 1))`: 33 actual
- `HAVING sum(i) > CAST(300.5 AS DECIMAL(10, 1))`: 72 actual, the Q18 shape
- `w < 10`: 110 actual
- `CAST(d AS DOUBLE) > 0.05`: 500 actual
In all of them interval analysis can't narrow the column, so the selectivity
is 1.0, which is worse than the default. The two new checks miss these causes:
- INT to decimal is lossless here, but `can_propagate_cast_constraints` has
no rule for it, and `decimal_columns_have_bounds` skips INT columns
- `w`'s range has 99 × 10^18 values, more than the `u64` returned by
`Interval::cardinality`, so the cardinality is `None`
- decimal to `Float64` is lossy, so the bound is rightly not propagated, but
`check_statistics_support` accepts the cast
`HAVING sum(i) > 300` is also 1100 on `main`, so this is not
decimal-specific. I wonder if `cardinality_ratio` should return `None` when a
cardinality is unknown or propagation doesn't reach a column, so that
`statistics_helper` falls back to the default for any type, and
`decimal_columns_have_bounds` isn't needed (`has_decimal_equality` is a
separate issue and should stay). WDYT?
<details><summary>Reproducer</summary>
```
# Reproducers for PR #27148. Paste at the end of
filter_statistics_decimal.slt.
#
# t has 1100 rows: i cycles through 0..99 (INT), w through 0..99 as
# DECIMAL(38, 18), and d through 0.00..0.10 as DECIMAL(15, 2).
query I
COPY (
SELECT
CAST(value % 100 AS INT) AS i,
CAST(value % 100 AS DECIMAL(38, 18)) AS w,
CAST((value % 11) / 100.0 AS DECIMAL(15, 2)) AS d
FROM range(1100)
)
TO 'test_files/scratch/filter_statistics_decimal/t.parquet'
STORED AS PARQUET;
----
1100
statement ok
set datafusion.explain.physical_plan_only = true;
statement ok
set datafusion.explain.show_statistics = true;
statement ok
set datafusion.execution.target_partitions = 1;
statement ok
CREATE EXTERNAL TABLE t
STORED AS PARQUET
LOCATION 'test_files/scratch/filter_statistics_decimal/t.parquet';
# INT column compared with a decimal literal
query TT
EXPLAIN SELECT i FROM t WHERE i < CAST(2.5 AS DECIMAL(5, 1));
----
query I
SELECT count(*) FROM t WHERE i < CAST(2.5 AS DECIMAL(5, 1));
----
# Aggregate with INT output compared with a decimal literal (TPC-H Q18 shape)
query TT
EXPLAIN SELECT i FROM t GROUP BY i HAVING sum(i) > CAST(300.5 AS DECIMAL(10,
1));
----
query I
SELECT count(*) FROM (SELECT i FROM t GROUP BY i HAVING sum(i) > CAST(300.5
AS DECIMAL(10, 1)));
----
# Range filter on a DECIMAL(38, 18) column
query TT
EXPLAIN SELECT w FROM t WHERE w < 10;
----
query I
SELECT count(*) FROM t WHERE w < 10;
----
# Decimal column under a cast to DOUBLE
query TT
EXPLAIN SELECT d FROM t WHERE CAST(d AS DOUBLE) > 0.05;
----
query I
SELECT count(*) FROM t WHERE CAST(d AS DOUBLE) > 0.05;
----
# Aggregate with INT output compared with an INT literal: no decimal involved
query TT
EXPLAIN SELECT i FROM t GROUP BY i HAVING sum(i) > 300;
----
statement ok
DROP TABLE t;
statement ok
set datafusion.execution.target_partitions = 4;
statement ok
RESET datafusion.explain.physical_plan_only;
statement ok
RESET datafusion.explain.show_statistics;
```
</details>
##########
datafusion/physical-expr/src/expressions/cast.rs:
##########
@@ -699,6 +699,28 @@ fn can_propagate_cast_constraints(source: &DataType,
target: &DataType) -> bool
|| (source.is_integer() && target.is_integer())
// NaN bounds are unbounded; finite Float32 values widen exactly.
|| (*source == Float32 && *target == Float64)
+ || is_decimal_widening(source, target)
+}
+
+/// Whether a decimal-to-decimal cast keeps every value exactly, i.e. the
target
+/// has at least as many fractional and integer digits as the source. Casting
+/// bounds back to the source then rounds to a neighboring representable value,
+/// which never excludes a valid input.
+fn is_decimal_widening(source: &DataType, target: &DataType) -> bool {
Review Comment:
`builder.rs` has the same rule in
[`recursive_type_can_be_coerced`](https://github.com/apache/datafusion/blob/9a5a7ad39c658656d7582a9fb504e0cde4185974/datafusion/expr/src/logical_plan/builder.rs#L133-L168),
which also handles INT to decimal (`Int32` as `(10, 0)`, `Int64` as `(19,
0)`). Sharing one helper would let bounds propagate through `CAST(int AS
Decimal)` too, fixing the `i` and `sum(i)` predicates from my comment in
`statistics_helper`. wdyt?
##########
datafusion/physical-expr/src/intervals/utils.rs:
##########
@@ -23,19 +23,57 @@ use datafusion_expr::interval_arithmetic::Interval;
/// will relax as more types of `PhysicalExpr`s and `Operator`s are supported.
/// Currently, [`CastExpr`], [`NegativeExpr`], [`BinaryExpr`], [`Column`] and
[`Literal`] are supported.
pub fn check_support(expr: &Arc<dyn PhysicalExpr>, schema: &SchemaRef) -> bool
{
+ check_support_with(expr, schema, is_datatype_supported)
+}
+
+/// Like [`check_support`], but additionally accepts decimal types. This is
+/// used when estimating filter statistics, where decimal predicates can be
+/// analyzed with interval arithmetic. Other users of [`check_support`], such
+/// as symmetric hash join pruning, do not support decimals yet.
+pub fn check_statistics_support(
+ expr: &Arc<dyn PhysicalExpr>,
+ schema: &SchemaRef,
+) -> bool {
+ check_support_with(expr, schema, |data_type| {
+ is_datatype_supported(data_type)
+ || matches!(
+ data_type,
+ DataType::Decimal32(_, _)
Review Comment:
Nit: `data_type.is_decimal()`, as in `filter.rs`, would be shorter and
consistent.
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1249,6 +1256,56 @@ fn compute_fallback_selectivity(
selectivity
}
+/// Returns true if every decimal column in `predicate` has both a min and a
+/// max statistic.
+///
+/// Interval analysis cannot narrow a column without bounds, so it estimates a
+/// selectivity of 1.0 for it. Filters on aggregate outputs, such as
+/// `HAVING sum(l_quantity) > 300` in TPC-H Q18, usually have no bounds. These
+/// decimal predicates used the default selectivity before interval analysis
+/// supported decimals, and join planning depends on that estimate, so they
+/// keep using it.
+fn decimal_columns_have_bounds(
+ predicate: &Arc<dyn PhysicalExpr>,
+ schema: &SchemaRef,
+ column_statistics: &[ColumnStatistics],
+) -> bool {
+ collect_columns(predicate).iter().all(|column| {
+ let is_decimal = schema
+ .fields()
+ .get(column.index())
+ .is_some_and(|field| field.data_type().is_decimal());
+ !is_decimal
+ || column_statistics.get(column.index()).is_some_and(|stats| {
+ stats.min_value.get_value().is_some()
Review Comment:
Nit: a null min/max scalar (`Precision::Exact(Decimal128(None, 15, 2))`)
passes this check, as `get_value().is_some()` is true, and gives a selectivity
of 1.0. I couldn't reach it from SQL, but checking `!v.is_null()` would cover
it. This test fails on the PR head (`left: Inexact(1100), right: Inexact(220)`):
```rust
#[tokio::test]
async fn test_filter_statistics_decimal_null_bounds() -> Result<()> {
// Table:
// a: Decimal128(15, 2), min=NULL, max=NULL
let schema =
Schema::new(vec![Field::new("a", DataType::Decimal128(15, 2),
true)]);
let null_decimal = ScalarValue::Decimal128(None, 15, 2);
let input = Arc::new(StatisticsExec::new(
Statistics {
num_rows: Precision::Inexact(1100),
total_byte_size: Precision::Absent,
column_statistics: vec![ColumnStatistics {
min_value: Precision::Exact(null_decimal.clone()),
max_value: Precision::Exact(null_decimal),
..Default::default()
}],
},
schema.clone(),
));
// WHERE a > 0.03
let predicate = binary(
col("a", &schema)?,
Operator::Gt,
lit(ScalarValue::Decimal128(Some(3), 15, 2)),
&schema,
)?;
let filter: Arc<dyn ExecutionPlan> =
Arc::new(FilterExec::try_new(predicate, input)?);
let statistics =
StatisticsContext::new().compute(filter.as_ref(),
&StatisticsArgs::new())?;
// Default selectivity (20%), as for a column without min/max
assert_eq!(statistics.num_rows, Precision::Inexact(220));
Ok(())
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
