## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/physical-expr/src/analysis.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    //! Interval and selectivity in [`AnalysisContext`]
    19
    20
    use std::fmt::Debug;
    21
    use std::sync::Arc;
    22
    23
    use crate::expressions::Column;
    24
    use crate::intervals::cp_solver::{ExprIntervalGraph, PropagationResult};
    25
    use crate::utils::collect_columns;
    26
    use crate::PhysicalExpr;
    27
    28
    use arrow::datatypes::Schema;
    29
    use datafusion_common::stats::Precision;
    30
    use datafusion_common::{
    31
        assert_or_internal_err, internal_datafusion_err, internal_err, ColumnStatistics,
    32
        DataFusionError, Result, ScalarValue,
    33
    };
    34
    use datafusion_expr::interval_arithmetic::{cardinality_ratio, Interval};
    35
    36
    /// The shared context used during the analysis of an expression. Includes
    37
    /// the boundaries for all known columns.
    38
    #[derive(Clone, Debug, PartialEq)]
    39
    pub struct AnalysisContext {
    40
        // A list of known column boundaries, ordered by the index
    41
        // of the column in the current schema.
    42
        pub boundaries: Vec<ExprBoundaries>,
    43
        /// The estimated percentage of rows that this expression would select, if
    44
        /// it were to be used as a boolean predicate on a filter. The value will be
    45
        /// between 0.0 (selects nothing) and 1.0 (selects everything).
    46
        pub selectivity: Option<f64>,
    47
    }
    48
    49
    impl AnalysisContext {
    50
    0
        pub fn new(boundaries: Vec<ExprBoundaries>) -> Self {
    51
    0
            Self {
    52
    0
                boundaries,
    53
    0
                selectivity: None,
    54
    0
            }
    55
    0
        }
    56
    57
    0
        pub fn with_selectivity(mut self, selectivity: f64) -> Self {
    58
    0
            self.selectivity = Some(selectivity);
    59
    0
            self
    60
    0
        }
    61
    62
        /// Create a new analysis context from column statistics.
    63
    0
        pub fn try_from_statistics(
    64
    0
            input_schema: &Schema,
    65
    0
            statistics: &[ColumnStatistics],
    66
    0
        ) -> Result<Self> {
    67
    0
            statistics
    68
    0
                .iter()
    69
    0
                .enumerate()
    70
    0
                .map(|(idx, stats)| ExprBoundaries::try_from_column(input_schema, stats, idx))
    71
    0
                .collect::<Result<Vec<_>>>()
    72
    0
                .map(Self::new)
    73
    0
        }
    74
    }
    75
    76
    /// Represents the boundaries (e.g. min and max values) of a particular column
    77
    ///
    78
    /// This is used range analysis of expressions, to determine if the expression
    79
    /// limits the value of particular columns (e.g. analyzing an expression such as
    80
    /// `time < 50` would result in a boundary interval for `time` having a max
    81
    /// value of `50`).
    82
    #[derive(Clone, Debug, PartialEq)]
    83
    pub struct ExprBoundaries {
    84
        pub column: Column,
    85
        /// Minimum and maximum values this expression can have. A `None` value
    86
        /// indicates that evaluating the given column results in an empty set.
    87
        /// For example, if the column `a` has values in the range [10, 20],
    88
        /// and there is a filter asserting that `a > 50`, then the resulting interval
    89
        /// range of `a` will be `None`.
    90
        pub interval: Option<Interval>,
    91
        /// Maximum number of distinct values this expression can produce, if known.
    92
        pub distinct_count: Precision<usize>,
    93
    }
    94
    95
    impl ExprBoundaries {
    96
        /// Create a new `ExprBoundaries` object from column level statistics.
    97
    0
        pub fn try_from_column(
    98
    0
            schema: &Schema,
    99
    0
            col_stats: &ColumnStatistics,
    100
    0
            col_index: usize,
    101
    0
        ) -> Result<Self> {
    102
    0
            let field = schema.fields().get(col_index).ok_or_else(|| {
    103
    0
                internal_datafusion_err!(
    104
    0
                    "Could not create `ExprBoundaries`: in `try_from_column` `col_index`
    105
    0
                    has gone out of bounds with a value of {col_index}, the schema has {} columns.",
    106
    0
                    schema.fields.len()
    107
                )
    108
    0
            })?;
    109
    0
            let empty_field =
    110
    0
                ScalarValue::try_from(field.data_type()).unwrap_or(ScalarValue::Null);
    111
    0
            let interval = Interval::try_new(
    112
    0
                col_stats
    113
    0
                    .min_value
    114
    0
                    .get_value()
    115
    0
                    .cloned()
    116
    0
                    .unwrap_or_else(|| empty_field.clone()),
    117
    0
                col_stats
    118
    0
                    .max_value
    119
    0
                    .get_value()
    120
    0
                    .cloned()
    121
    0
                    .unwrap_or(empty_field),
    122
    0
            )?;
    123
    0
            let column = Column::new(field.name(), col_index);
    124
    0
            Ok(ExprBoundaries {
    125
    0
                column,
    126
    0
                interval: Some(interval),
    127
    0
                distinct_count: col_stats.distinct_count,
    128
    0
            })
    129
    0
        }
    130
    131
        /// Create `ExprBoundaries` that represent no known bounds for all the
    132
        /// columns in `schema`
    133
    0
        pub fn try_new_unbounded(schema: &Schema) -> Result<Vec<Self>> {
    134
    0
            schema
    135
    0
                .fields()
    136
    0
                .iter()
    137
    0
                .enumerate()
    138
    0
                .map(|(i, field)| {
    139
                    Ok(Self {
    140
    0
                        column: Column::new(field.name(), i),
    141
    0
                        interval: Some(Interval::make_unbounded(field.data_type())?),
    142
    0
                        distinct_count: Precision::Absent,
    143
                    })
    144
    0
                })
    145
    0
                .collect()
    146
    0
        }
    147
    }
    148
    149
    /// Attempts to refine column boundaries and compute a selectivity value.
    150
    ///
    151
    /// The function accepts boundaries of the input columns in the `context` parameter.
    152
    /// It then tries to tighten these boundaries based on the provided `expr`.
    153
    /// The resulting selectivity value is calculated by comparing the initial and final boundaries.
    154
    /// The computation assumes that the data within the column is uniformly distributed and not sorted.
    155
    ///
    156
    /// # Arguments
    157
    ///
    158
    /// * `context` - The context holding input column boundaries.
    159
    /// * `expr` - The expression used to shrink the column boundaries.
    160
    ///
    161
    /// # Returns
    162
    ///
    163
    /// * `AnalysisContext` constructed by pruned boundaries and a selectivity value.
    164
    0
    pub fn analyze(
    165
    0
        expr: &Arc<dyn PhysicalExpr>,
    166
    0
        context: AnalysisContext,
    167
    0
        schema: &Schema,
    168
    0
    ) -> Result<AnalysisContext> {
    169
    0
        let initial_boundaries = &context.boundaries;
    170
    0
        if initial_boundaries
    171
    0
            .iter()
    172
    0
            .all(|bound| bound.interval.is_none())
    173
        {
    174
    0
            assert_or_internal_err!(
    175
    0
                !initial_boundaries
    176
    0
                    .iter()
    177
    0
                    .any(|bound| bound.distinct_count != Precision::Exact(0)),
    178
    0
                "ExprBoundaries has a non-zero distinct count although it represents an empty table"
    179
            );
    180
    0
            assert_or_internal_err!(
    181
    0
                context.selectivity == Some(0.0),
    182
    0
                "AnalysisContext has a non-zero selectivity although it represents an empty table"
    183
            );
    184
    0
            Ok(context)
    185
    0
        } else if initial_boundaries
    186
    0
            .iter()
    187
    0
            .any(|bound| bound.interval.is_none())
    188
        {
    189
    0
            internal_err!(
    190
    0
                    "AnalysisContext is an inconsistent state. Some columns represent empty table while others don't"
    191
                )
    192
        } else {
    193
    0
            let mut target_boundaries = context.boundaries;
    194
    0
            let mut graph = ExprIntervalGraph::try_new(Arc::clone(expr), schema)?;
    195
    0
            let columns = collect_columns(expr)
    196
    0
                .into_iter()
    197
    0
                .map(|c| Arc::new(c) as _)
    198
    0
                .collect::<Vec<_>>();
    199
    200
    0
            let mut target_indices_and_boundaries = vec![];
    201
    0
            let target_expr_and_indices = graph.gather_node_indices(columns.as_slice());
    202
    203
    0
            for (expr, index) in &target_expr_and_indices {
    204
    0
                if let Some(column) = expr.as_any().downcast_ref::<Column>() {
    205
    0
                    if let Some(bound) =
    206
    0
                        target_boundaries.iter().find(|b| b.column == *column)
    207
    0
                    {
    208
    0
                        // Now, it's safe to unwrap
    209
    0
                        target_indices_and_boundaries
    210
    0
                            .push((*index, bound.interval.as_ref().unwrap().clone()));
    211
    0
                    }
    212
    0
                }
    213
            }
    214
    215
    0
            match graph.update_ranges(&mut target_indices_and_boundaries, Interval::TRUE)? {
    216
                PropagationResult::Success => {
    217
    0
                    shrink_boundaries(&graph, target_boundaries, &target_expr_and_indices)
    218
                }
    219
                PropagationResult::Infeasible => {
    220
                    // If the propagation result is infeasible, set intervals to None
    221
    0
                    target_boundaries
    222
    0
                        .iter_mut()
    223
    0
                        .for_each(|bound| bound.interval = None);
    224
    0
                    Ok(AnalysisContext::new(target_boundaries).with_selectivity(0.0))
    225
                }
    226
                PropagationResult::CannotPropagate => {
    227
    0
                    Ok(AnalysisContext::new(target_boundaries).with_selectivity(1.0))
    228
                }
    229
            }
    230
        }
    231
    0
    }
    232
    233
    /// If the `PropagationResult` indicates success, this function calculates the
    234
    /// selectivity value by comparing the initial and final column boundaries.
    235
    /// Following this, it constructs and returns a new `AnalysisContext` with the
    236
    /// updated parameters.
    237
    0
    fn shrink_boundaries(
    238
    0
        graph: &ExprIntervalGraph,
    239
    0
        mut target_boundaries: Vec<ExprBoundaries>,
    240
    0
        target_expr_and_indices: &[(Arc<dyn PhysicalExpr>, usize)],
    241
    0
    ) -> Result<AnalysisContext> {
    242
    0
        let initial_boundaries = target_boundaries.clone();
    243
    0
        target_expr_and_indices.iter().for_each(|(expr, i)| {
    244
    0
            if let Some(column) = expr.as_any().downcast_ref::<Column>() {
    245
    0
                if let Some(bound) = target_boundaries
    246
    0
                    .iter_mut()
    247
    0
                    .find(|bound| bound.column.eq(column))
    248
    0
                {
    249
    0
                    bound.interval = Some(graph.get_interval(*i));
    250
    0
                };
    251
    0
            }
    252
    0
        });
    253
    254
    0
        let selectivity = calculate_selectivity(&target_boundaries, &initial_boundaries)?;
    255
    256
    0
        assert_or_internal_err!(
    257
    0
            (0.0..=1.0).contains(&selectivity),
    258
    0
            "Selectivity is out of limit: {selectivity}",
    259
        );
    260
    261
    0
        Ok(AnalysisContext::new(target_boundaries).with_selectivity(selectivity))
    262
    0
    }
    263
    264
    /// This function calculates the filter predicate's selectivity by comparing
    265
    /// the initial and pruned column boundaries. Selectivity is defined as the
    266
    /// ratio of rows in a table that satisfy the filter's predicate.
    267
    0
    fn calculate_selectivity(
    268
    0
        target_boundaries: &[ExprBoundaries],
    269
    0
        initial_boundaries: &[ExprBoundaries],
    270
    0
    ) -> Result<f64> {
    271
        // Since the intervals are assumed uniform and the values
    272
        // are not correlated, we need to multiply the selectivities
    273
        // of multiple columns to get the overall selectivity.
    274
    0
        if target_boundaries.len() != initial_boundaries.len() {
    275
    0
            return Err(internal_datafusion_err!(
    276
    0
                "The number of columns in the initial and target boundaries should be the same"
    277
    0
            ));
    278
    0
        }
    279
    0
        let mut acc: f64 = 1.0;
    280
    0
        for (initial, target) in initial_boundaries.iter().zip(target_boundaries) {
    281
    0
            match (initial.interval.as_ref(), target.interval.as_ref()) {
    282
    0
                (Some(initial), Some(target)) => {
    283
    0
                    acc *= cardinality_ratio(initial, target);
    284
    0
                }
    285
                (None, Some(_)) => {
    286
    0
                    return internal_err!(
    287
    0
                    "Initial boundary cannot be None while having a Some() target boundary"
    288
                );
    289
                }
    290
    0
                _ => return Ok(0.0),
    291
            }
    292
        }
    293
    294
    0
        Ok(acc)
    295
    0
    }
    296
    297
    #[cfg(test)]
    298
    mod tests {
    299
        use std::sync::Arc;
    300
    301
        use arrow::datatypes::{DataType, Field, Schema};
    302
        use datafusion_common::{assert_contains, DFSchema};
    303
        use datafusion_expr::{
    304
            col, execution_props::ExecutionProps, interval_arithmetic::Interval, lit, Expr,
    305
        };
    306
    307
        use crate::{create_physical_expr, AnalysisContext};
    308
    309
        use super::{analyze, ExprBoundaries};
    310
    311
    0
        fn make_field(name: &str, data_type: DataType) -> Field {
    312
    0
            let nullable = false;
    313
    0
            Field::new(name, data_type, nullable)
    314
    0
        }
    315
    316
        #[test]
    317
    0
        fn test_analyze_boundary_exprs() {
    318
    0
            let schema = Arc::new(Schema::new(vec![make_field("a", DataType::Int32)]));
    319
    320
            /// Test case containing (expression tree, lower bound, upper bound)
    321
            type TestCase = (Expr, Option<i32>, Option<i32>);
    322
    323
    0
            let test_cases: Vec<TestCase> = vec![
    324
                // a > 10
    325
    0
                (col("a").gt(lit(10)), Some(11), None),
    326
                // a < 20
    327
    0
                (col("a").lt(lit(20)), None, Some(19)),
    328
                // a > 10 AND a < 20
    329
    0
                (
    330
    0
                    col("a").gt(lit(10)).and(col("a").lt(lit(20))),
    331
    0
                    Some(11),
    332
    0
                    Some(19),
    333
    0
                ),
    334
                // a >= 10
    335
    0
                (col("a").gt_eq(lit(10)), Some(10), None),
    336
                // a <= 20
    337
    0
                (col("a").lt_eq(lit(20)), None, Some(20)),
    338
                // a >= 10 AND a <= 20
    339
    0
                (
    340
    0
                    col("a").gt_eq(lit(10)).and(col("a").lt_eq(lit(20))),
    341
    0
                    Some(10),
    342
    0
                    Some(20),
    343
    0
                ),
    344
                // a > 10 AND a < 20 AND a < 15
    345
    0
                (
    346
    0
                    col("a")
    347
    0
                        .gt(lit(10))
    348
    0
                        .and(col("a").lt(lit(20)))
    349
    0
                        .and(col("a").lt(lit(15))),
    350
    0
                    Some(11),
    351
    0
                    Some(14),
    352
    0
                ),
    353
                // (a > 10 AND a < 20) AND (a > 15 AND a < 25)
    354
    0
                (
    355
    0
                    col("a")
    356
    0
                        .gt(lit(10))
    357
    0
                        .and(col("a").lt(lit(20)))
    358
    0
                        .and(col("a").gt(lit(15)))
    359
    0
                        .and(col("a").lt(lit(25))),
    360
    0
                    Some(16),
    361
    0
                    Some(19),
    362
    0
                ),
    363
            ];
    364
    0
            for (expr, lower, upper) in test_cases {
    365
    0
                let boundaries = ExprBoundaries::try_new_unbounded(&schema).unwrap();
    366
    0
                let df_schema = DFSchema::try_from(Arc::clone(&schema)).unwrap();
    367
    0
                let physical_expr =
    368
    0
                    create_physical_expr(&expr, &df_schema, &ExecutionProps::new()).unwrap();
    369
    0
                let analysis_result = analyze(
    370
    0
                    &physical_expr,
    371
    0
                    AnalysisContext::new(boundaries),
    372
    0
                    df_schema.as_ref(),
    373
                )
    374
    0
                .unwrap();
    375
    0
                let Some(actual) = &analysis_result.boundaries[0].interval else {
    376
    0
                    panic!("The analysis result should contain non-empty intervals for all columns");
    377
                };
    378
    0
                let expected = Interval::make(lower, upper).unwrap();
    379
    0
                assert_eq!(
    380
    0
                    &expected, actual,
    381
    0
                    "did not get correct interval for SQL expression: {expr:?}"
    382
                );
    383
            }
    384
    0
        }
    385
    386
        #[test]
    387
    0
        fn test_analyze_empty_set_boundary_exprs() {
    388
    0
            let schema = Arc::new(Schema::new(vec![make_field("a", DataType::Int32)]));
    389
    390
    0
            let test_cases: Vec<Expr> = vec![
    391
                // a > 10 AND a < 10
    392
    0
                col("a").gt(lit(10)).and(col("a").lt(lit(10))),
    393
                // a > 5 AND (a < 20 OR a > 20)
    394
                // a > 10 AND a < 20
    395
                // (a > 10 AND a < 20) AND (a > 20 AND a < 30)
    396
    0
                col("a")
    397
    0
                    .gt(lit(10))
    398
    0
                    .and(col("a").lt(lit(20)))
    399
    0
                    .and(col("a").gt(lit(20)))
    400
    0
                    .and(col("a").lt(lit(30))),
    401
            ];
    402
    403
    0
            for expr in test_cases {
    404
    0
                let boundaries = ExprBoundaries::try_new_unbounded(&schema).unwrap();
    405
    0
                let df_schema = DFSchema::try_from(Arc::clone(&schema)).unwrap();
    406
    0
                let physical_expr =
    407
    0
                    create_physical_expr(&expr, &df_schema, &ExecutionProps::new()).unwrap();
    408
    0
                let analysis_result = analyze(
    409
    0
                    &physical_expr,
    410
    0
                    AnalysisContext::new(boundaries),
    411
    0
                    df_schema.as_ref(),
    412
                )
    413
    0
                .unwrap();
    414
    415
    0
                for boundary in analysis_result.boundaries {
    416
    0
                    assert!(boundary.interval.is_none());
    417
                }
    418
            }
    419
    0
        }
    420
    421
        #[test]
    422
    0
        fn test_analyze_invalid_boundary_exprs() {
    423
    0
            let schema = Arc::new(Schema::new(vec![make_field("a", DataType::Int32)]));
    424
    0
            let expr = col("a").lt(lit(10)).or(col("a").gt(lit(20)));
    425
    0
            let expected_error = "OR operator cannot yet propagate true intervals";
    426
    0
            let boundaries = ExprBoundaries::try_new_unbounded(&schema).unwrap();
    427
    0
            let df_schema = DFSchema::try_from(Arc::clone(&schema)).unwrap();
    428
    0
            let physical_expr =
    429
    0
                create_physical_expr(&expr, &df_schema, &ExecutionProps::new()).unwrap();
    430
    0
            let analysis_error = analyze(
    431
    0
                &physical_expr,
    432
    0
                AnalysisContext::new(boundaries),
    433
    0
                df_schema.as_ref(),
    434
            )
    435
    0
            .unwrap_err();
    436
    0
            assert_contains!(analysis_error.to_string(), expected_error);
    437
    0
        }
    438
    }