DataFusion's Interval Analysis: How `BETWEEN` Becomes a Scalar Range and a Selectivity Number

TL;DR: a BETWEEN 10 AND 20 is desugared into a >= 10 AND a <= 20 by the expression simplifier, then re-absorbed into a single Interval (two ScalarValue endpoints) by the constraint-propagation solver. That interval is intersected with the column’s min/max from statistics, and the ratio of the two interval cardinalities becomes the filter selectivity. No histogram, no sampling, no join — just two numbers per column and a divide.

Every cost-based optimizer eventually hits the same wall: to choose between a hash join and a nested loop join, you need to know how many rows will survive a filter. Most systems solve this with histograms, sampling, or a learned model. All three are expensive, all three are approximate, and all three are a source of embarrassing production incidents.

Apache DataFusion takes a much cheaper route. It has exactly two pieces of information per column — a minimum and a maximum — and it derives a filter’s selectivity from the geometric relationship between the range implied by the predicate and the range the data actually occupies.

This article walks through the whole path: how BETWEEN becomes an interval, how the interval is propagated, and where the resulting number ends up.

The vocabulary: `Precision<T>`

Before anything else, the tri-state that everything else is built on:

rust
// datafusion/common/src/stats.rs:31-60
pub enum Precision<T: Debug + Clone + PartialEq + Eq + PartialOrd> {
    /// The exact value is known. Used for guaranteeing correctness.
    Exact(T),
    /// The value is not known exactly, but is likely close to this value.
    /// Used for cost-based optimizations.
    Inexact(T),
    /// Nothing is known about the value. This is the default state.
    #[default]
    Absent,
}

Exact is not “approximately right”. The doc comment is explicit: exact values are “used for guaranteeing correctness” — they answer COUNT(*), MIN and MAX without scanning and let the optimizer delete limits. Inexact is the cost-based-optimization tier. Absent is an absorbing element: any arithmetic involving Absent yields Absent.

This distinction matters later, because the pruning layer (covered in a companion piece) refuses to use anything but Exact bounds.

The container is small:

rust
// datafusion/common/src/stats.rs:368-383
pub struct Statistics {
    pub num_rows: Precision<usize>,
    pub total_byte_size: Precision<usize>,
    pub column_statistics: Vec<ColumnStatistics>,
}

and the per-column part is where the interesting fields live:

rust
// datafusion/common/src/stats.rs:1080-1114
pub struct ColumnStatistics {
    pub null_count: Precision<usize>,
    pub max_value: Precision<ScalarValue>,
    pub min_value: Precision<ScalarValue>,
    pub sum_value: Precision<ScalarValue>,
    pub distinct_count: Precision<usize>,
    pub byte_size: Precision<usize>,
}

Note the shape: min_value and max_value are ScalarValues, not typed numbers. The same struct covers Int64, Utf8, Timestamp and Decimal256 without a type parameter, at the cost of runtime type dispatch everywhere.

Step 1: `BETWEEN` is not special

The first thing DataFusion does with a BETWEEN 10 AND 20 is throw away the Between node entirely. The expression simplifier has a rule for it:

rust
// datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs:1654-1669
//
// Rules for Between
//
// a between 3 and 5  -->  a >= 3 AND a <=5
// a not between 3 and 5  -->  a < 3 OR a > 5
Expr::Between(between) => Transformed::yes(if between.negated {
    let l = *between.expr.clone();
    let r = *between.expr;
    or(l.lt(*between.low), r.gt(*between.high))
} else {
    and(
        between.expr.clone().gt_eq(*between.low),
        between.expr.lt_eq(*between.high),
    )
}),

This looks like a loss of information — we now have two independent conjuncts instead of one node. It is not. Two comparisons against the same column is exactly the shape the interval solver wants, and the solver is the thing that puts the range back together.

Step 2: intervals over `ScalarValue`

The interval type is deliberately minimal:

rust
// datafusion/expr-common/src/interval_arithmetic.rs:177-181
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Interval {
    lower: ScalarValue,
    upper: ScalarValue,
}

A NULL endpoint means “unbounded on this side”. The constructor is strict about ordering, because an inverted interval is always a bug:

rust
// datafusion/expr-common/src/interval_arithmetic.rs:267-285
pub fn try_new(lower: ScalarValue, upper: ScalarValue) -> Result<Self> {
    assert_eq_or_internal_err!(
        lower.data_type(), upper.data_type(),
        "Endpoints of an Interval should have the same type"
    );
    let interval = Self::new(lower, upper);
    assert_or_internal_err!(
        interval.lower.is_null()
            || interval.upper.is_null()
            || interval.lower <= interval.upper,
        "Interval's lower bound {} is greater than the upper bound {}",
        interval.lower, interval.upper
    );
    Ok(interval)
}

Self::new also standardizes the awkward cases: NaN becomes unbounded, +INF becomes f64::MAX for a lower bound and NULL for an upper bound, and unsigned lower bounds default to zero rather than -inf.

Step 3: the column’s interval comes from statistics

The solver’s input is one interval per column, built straight out of ColumnStatistics:

rust
// datafusion/physical-expr/src/analysis.rs:97-129
pub fn try_from_column(
    schema: &Schema,
    col_stats: &ColumnStatistics,
    col_index: usize,
) -> Result<Self> {
    let field = schema.fields().get(col_index).ok_or_else(|| { ... })?;
    let empty_field =
        ScalarValue::try_from(field.data_type()).unwrap_or(ScalarValue::Null);
    let interval = Interval::try_new(
        col_stats.min_value.get_value().cloned().unwrap_or_else(|| empty_field.clone()),
        col_stats.max_value.get_value().cloned().unwrap_or(empty_field),
    )?;
    ...
}

So ColumnStatistics { min_value: Exact(Int32(0)), max_value: Exact(Int32(999)) } becomes Interval { lower: Int32(0), upper: Int32(999) }, and a column with absent statistics becomes an unbounded interval.

Step 4: propagating the predicate backwards

analyze() is the entry point. It builds an expression DAG, seeds the leaf nodes with the intervals from statistics, and pushes the predicate’s constraints down into them:

rust
// datafusion/physical-expr/src/analysis.rs:164-232 (excerpt)
pub fn analyze(
    expr: &Arc<dyn PhysicalExpr>,
    context: AnalysisContext,
    schema: &Schema,
) -> Result<AnalysisContext> {
    ...
    let mut target_boundaries = context.boundaries;
    let mut graph = ExprIntervalGraph::try_new(Arc::clone(expr), schema)?;
    ...
    match graph.update_ranges(&mut target_indices_and_boundaries, Interval::TRUE)? {
        PropagationResult::Success => {
            shrink_boundaries(&graph, target_boundaries, &target_expr_and_indices)
        }
        PropagationResult::Infeasible => {
            // If the propagation result is infeasible, set intervals to None
            for bound in &mut target_boundaries { bound.interval = None; }
            Ok(AnalysisContext::new(target_boundaries).with_selectivity(0.0))
        }
        PropagationResult::CannotPropagate => {
            Ok(AnalysisContext::new(target_boundaries).with_selectivity(1.0))
        }
    }
}

Three outcomes, and they map cleanly onto “how selective is this filter”: Success means the constraint could be pushed down and we can compute a selectivity; Infeasible means the predicate contradicts the data (the filter matches nothing); CannotPropagate means we learned nothing, so assume everything passes.

The propagation itself is where the AND of two comparisons gets merged back into one interval. For a comparison under a TRUE parent:

rust
// datafusion/physical-expr/src/intervals/cp_solver.rs:353-373
pub fn propagate_comparison(
    op: &Operator,
    parent: &Interval,
    left_child: &Interval,
    right_child: &Interval,
) -> Result<Option<(Interval, Interval)>> {
    if parent == &Interval::TRUE {
        match op {
            Operator::Eq => left_child.intersect(right_child).map(|result| {
                result.map(|intersection| (intersection.clone(), intersection))
            }),
            Operator::Gt => satisfy_greater(left_child, right_child, true),
            Operator::GtEq => satisfy_greater(left_child, right_child, false),
            Operator::Lt => satisfy_greater(right_child, left_child, true)
                .map(|t| t.map(reverse_tuple)),
            Operator::LtEq => satisfy_greater(right_child, left_child, false)
                .map(|t| t.map(reverse_tuple)),
            ...

a >= 10 under TRUE shrinks a’s interval from below; a <= 20 shrinks it from above. Because both are propagated against the same column node, the column’s interval ends up as [10, 20] — the BETWEEN range, reconstructed.

The whole flow:

text

   a BETWEEN 10 AND 20
            │
            ▼  expr_simplifier.rs:1660
   (a >= 10) AND (a <= 20)
            │
            ▼  cp_solver.rs:353  propagate_comparison
   column a interval: [0, 999]  ∩  [10, +inf]  ∩  (-inf, 20]
            │
            ▼
   column a interval: [10, 20]         ← the exact scalar interval
            │
            ▼  cardinality_ratio (interval_arithmetic.rs:1740)
   selectivity = |[10,20]| / |[0,999]|

Step 5: from interval to number

The selectivity is a ratio of cardinalities:

rust
// datafusion/expr-common/src/interval_arithmetic.rs:1740-1747
pub fn cardinality_ratio(initial_interval: &Interval, final_interval: &Interval) -> f64 {
    match (final_interval.cardinality(), initial_interval.cardinality()) {
        (Some(final_interval), Some(initial_interval)) => {
            (final_interval as f64) / (initial_interval as f64)
        }
        _ => 1.0,
    }
}

Interval::cardinality() only works for numeric, date, timestamp and decimal types — for a Utf8 column it returns None, and the ratio degrades to 1.0 (no filtering). This is the single biggest limitation of the whole scheme, and it is why DataFusion has a separate string-specific estimator (string_selectivity in datafusion/physical-plan/src/filter.rs:1292) that falls back to a configurable default.

Two more subtleties live in calculate_selectivity:

rust
// datafusion/physical-expr/src/analysis.rs:305-344
fn calculate_selectivity(
    target_boundaries: &[ExprBoundaries],
    initial_boundaries: &[ExprBoundaries],
) -> Result<f64> {
    // Since the intervals are assumed uniform and the values
    // are not correlated, we need to multiply the selectivities
    // of multiple columns to get the overall selectivity.
    let mut acc: f64 = 1.0;
    for (initial, target) in initial_boundaries.iter().zip(target_boundaries) {
        match (initial.interval.as_ref(), target.interval.as_ref()) {
            (Some(initial_interval), Some(target_interval)) => {
                if let Precision::Exact(distinct_count)
                | Precision::Inexact(distinct_count) = target.distinct_count
                    && let Some(s) = singleton_selectivity(
                        initial_interval, target_interval, distinct_count,
                    )
                {
                    acc *= s;
                    continue;
                }
                acc *= cardinality_ratio(initial_interval, target_interval);
            }
            ...

Two things to notice. First, the “uniform distribution” assumption is explicit in the comment: values are assumed uniformly spread over the interval and columns are assumed uncorrelated. city = 'Paris' AND state = 'CA' multiplies two independent selectivities, which is wrong whenever they are correlated — and they usually are.

Second, the singleton_selectivity shortcut. If the filter collapsed the column’s interval to a single point, the ratio-based estimate would still be 1/999 even when the column holds only 10 distinct values. The shortcut replaces the ratio with 1 / distinct_count:

rust
// datafusion/physical-expr/src/analysis.rs:274-300
fn singleton_selectivity(
    initial_interval: &Interval,
    target_interval: &Interval,
    distinct_count: usize,
) -> Option<f64> {
    if distinct_count == 0
        || target_interval.lower().is_null()
        || target_interval.lower() != target_interval.upper()
    {
        return None;
    }
    // Only treat this as a newly-applied equality filter when the initial
    // interval was not already that same singleton.
    let initial_is_same_singleton = !initial_interval.lower().is_null()
        && initial_interval.lower() == initial_interval.upper()
        && initial_interval.lower() == target_interval.lower();
    if initial_is_same_singleton {
        return None;
    }
    Some(1.0 / distinct_count as f64)
}

The initial_is_same_singleton guard is a bug fix worth calling out: if the statistics already said the column was a single value, applying 1/NDV again would double-count the restriction and under-estimate the row count.

Step 6: where the number is consumed

FilterExec is where the estimate becomes a Statistics:

rust
// datafusion/physical-plan/src/filter.rs:389-406
if check_support(predicate, schema) {
    let input_analysis_ctx = AnalysisContext::try_from_statistics(
        schema,
        &input_stats.column_statistics,
    )?;
    let analysis_ctx = analyze(predicate, input_analysis_ctx, schema)?;
    let selectivity = analysis_ctx.selectivity.unwrap_or(1.0);
    let filtered_num_rows =
        input_num_rows.with_estimated_selectivity(selectivity);
    let cs = collect_new_statistics(
        schema,
        &input_stats.column_statistics,
        analysis_ctx.boundaries,
        selectivity,
        &null_rejecting_columns,
        filtered_num_rows,
    );
    (selectivity, filtered_num_rows, cs)

with_estimated_selectivity is where the estimate is explicitly downgraded to Inexact, which is the honest thing to do:

rust
// datafusion/common/src/stats.rs:200-206
pub fn with_estimated_selectivity(self, selectivity: f64) -> Self {
    if self == Precision::Exact(0) {
        return self;
    }
    self.map(|v| ((v as f64 * selectivity).ceil()) as usize)
        .to_inexact()
}

Notice ceil(): a filter never increases the row count, but rounding down would claim it might. Rounding up is the conservative direction.

The boundaries that come out of analyze are also fed into collect_new_statistics, which tightens the min/max of the surviving columns. So WHERE a BETWEEN 10 AND 20 does not just produce a row count; it produces a narrower ColumnStatistics for a, which then propagates upward to the join, the aggregate, and the next filter.

The memoization that makes this affordable

Interval analysis is not cheap: it builds a petgraph DAG per expression. Running it once per filter per plan would dominate planning time on wide plans. DataFusion’s answer is a memoized bottom-up walk:

rust
// datafusion/physical-plan/src/statistics.rs:39-68
type CacheKey = (usize, Option<usize>);

fn cache_key(plan: &dyn ExecutionPlan, partition: Option<usize>) -> CacheKey {
    (from_ref::<dyn ExecutionPlan>(plan).cast::<()>() as usize, partition)
}

/// Per-context memoization cache for statistics computation.
///
/// Keys are `(plan node pointer address, partition)`. Every entry retains the
/// `Arc` whose pointer supplied its key, preventing a dropped plan's address
/// from being reused by a different plan while the entry exists.

The key is a raw pointer, and each cache entry holds the Arc that supplied the pointer. That is a small, clever trick: it makes address reuse impossible while the entry is alive, so the cache can be keyed on identity without inventing a plan-wide unique id. The cost is that the cache keeps dead plans alive until reset_cache() is called.

The walk itself consults a pluggable provider chain before the operator’s built-in statistics_from_inputs, which is the extension point used to feed catalog-level statistics into the optimizer.

Trade-offs and failure modes

The uniform distribution assumption is the big one. A column with a Zipfian or exponential distribution will produce a selectivity that is wrong by an order of magnitude. DataFusion has no histogram, no sketch, no sample. If your amount column has 99% of its mass in [0, 100] and the file’s min/max is [0, 1000000], then amount < 1000 gets a selectivity of 0.001 when the truth is closer to 0.5. The planner then thinks the join build side is tiny and picks the wrong one.

Independence across columns. calculate_selectivity multiplies per-column selectivities. Correlated predicates multiply twice, so the estimated cardinality can be off by orders of magnitude in the low direction.

Non-numeric columns get nothing. cardinality() returns None for Utf8, List, Struct. The interval machinery simply does not apply, and DataFusion falls back to string_selectivity or a flat default.

Exact versus Inexact is a correctness boundary, not a hint. A filter’s output row count is always Inexact unless it was provably zero. The optimizer is allowed to use Exact values to skip work entirely (removing a LIMIT because the input has exactly 5 rows), so a source that lies about Exact gets wrong answers, not just bad plans.

NOT BETWEEN is subtler than it looks. NOT (a BETWEEN -1 AND 1) over a ∈ [-1, 1] is infeasible — no row can satisfy it — and the solver correctly reports None for the interval. But NOT (a BETWEEN -1 AND 1) over a ∈ [-2, 2] does have rows, and the solver has to widen the interval rather than intersect it. The test at datafusion/physical-expr/src/analysis.rs:715-752 pins this behavior down.

Signed zero. NOT (a = +0.0) over a ∈ [-0.0, +0.0] must be infeasible, because +0.0 = +0.0 is TRUE in IEEE 754 even though PartialEq on ScalarValue is bit-wise. The solver has explicit handling for this, and the comment at datafusion/physical-expr/src/analysis.rs:782-790 explains exactly which short-circuit it defeats.

Configuration

There is no single “enable interval analysis” switch. The knobs that matter:

Option	Default	Effect
`datafusion.execution.collect_statistics`	`true`	Whether file/table statistics are gathered at all. Without them, every interval is unbounded and selectivity is 1.0.
`datafusion.optimizer.skip_failed_rules`	`false`	If an optimizer rule throws (e.g. an interval construction error), fail the query rather than silently continuing.
`datafusion.optimizer.max_passes`	`3`	How many times the logical optimizer re-runs the rule set; more passes mean more statistics-driven rewrites get a chance to fire.
`FILTER_EXEC_DEFAULT_SELECTIVITY` (`filter.rs:80`)	`20`	Not a config key but a `FilterExecBuilder::with_default_selectivity` knob: the fallback selectivity percentage for string predicates the interval machinery cannot handle.

The big lever is collect_statistics. Turn it off and every downstream estimate collapses to Absent: no pruning, no build-side selection, no interval analysis. It is also the knob that matters most for listing-heavy workloads, because gathering statistics means reading Parquet footers.

Bottom line

The trick that makes DataFusion’s planner work without histograms is that it does not try to be right about distributions. It only tries to be right about ranges, and it converts a range comparison into a number by dividing interval cardinalities. That is cheap, it is local, and it degrades gracefully — but the moment your data is skewed, or your columns are correlated, or your predicate is on a string, the estimate is fiction.

The honest summary: DataFusion’s statistics are a fast, structured guess, and the codebase is refreshingly explicit about it — every comment in analysis.rs that says “assumed uniform” or “not correlated” is a warning label left by whoever had to debug the resulting plan.

The vocabulary: Precision<T>

Step 1: BETWEEN is not special

Step 2: intervals over ScalarValue