Function analyze

datafusion/physical-expr/src/analysis.rs:164–231

Attempts to refine column boundaries and compute a selectivity value. The function accepts boundaries of the input columns in the `context` parameter. It then tries to tighten these boundaries based on the provided `expr`. The resulting selectivity value is calculated by comparing the initial and final boundaries. The computation assumes that the data within the column is uniformly distributed an

(
    expr: &Arc<dyn PhysicalExpr>,
    context: AnalysisContext,
    schema: &Schema,
)

///
/// * `AnalysisContext` constructed by pruned boundaries and a selectivity value.
pub fn analyze(
    expr: &Arc<dyn PhysicalExpr>,
    context: AnalysisContext,
    schema: &Schema,
) -> Result<AnalysisContext> {
    let initial_boundaries = &context.boundaries;

    if initial_boundaries
        .iter()
        .all(|bound| bound.interval.is_none())
    {
        assert_or_internal_err!(
            !initial_boundaries
                .iter()
                .any(|bound| bound.distinct_count != Precision::Exact(0)),
            "ExprBoundaries has a non-zero distinct count although it represents an empty table"
        );
        assert_or_internal_err!(
            context.selectivity.unwrap_or(0.0) == 0.0,
            "AnalysisContext has a non-zero selectivity although it represents an empty table"
        );
        Ok(context)
    } else if initial_boundaries
        .iter()
        .any(|bound| bound.interval.is_none())
    {
        internal_err!(
            "AnalysisContext is an inconsistent state. Some columns represent empty table while others don't"
        )
    } else {
        let mut target_boundaries = context.boundaries;
        let mut graph = ExprIntervalGraph::try_new(Arc::clone(expr), schema)?;
        let columns = collect_columns(expr)
            .into_iter()
            .map(|c| Arc::new(c) as _)
            .collect::<Vec<_>>();

        let mut target_indices_and_boundaries = vec![];
        let target_expr_and_indices = graph.gather_node_indices(columns.as_slice());

        for (expr, index) in &target_expr_and_indices {
            if let Some(column) = expr.downcast_ref::<Column>()
                && let Some(bound) =
                    target_boundaries.iter().find(|b| b.column == *column)
            {
                // Now, it's safe to unwrap
                target_indices_and_boundaries
                    .push((*index, bound.interval.as_ref().unwrap().clone()));
            }
        }

        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
                target_boundaries