// SPDX-License-Identifier: Apache-2.0

// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership

// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! This module provides data structures to represent statistics

use std::fmt::{self, Debug, Display};

use crate::{Result, ScalarValue};

use crate::error::_plan_err;

use crate::utils::aggregate::precision_add;

use arrow::datatypes::{DataType, Schema};

/// Represents a value with a degree of certainty. `Precision` is used to

/// propagate information the precision of statistical values.

#[derive(Clone, PartialEq, Eq, Default, Copy)]

pub enum Precision<T: Debug + Clone + PartialEq + Eq + PartialOrd> {

/// The exact value is known. Used for guaranteeing correctness.

///

/// Comes from definitive sources such as:

/// - Parquet file metadata (row counts, byte sizes)

/// - In-memory RecordBatch data (actual row counts, byte sizes, null counts)

/// - and more...

Exact(T),

/// The value is not known exactly, but is likely close to this value.

/// Used for cost-based optimizations.

///

/// Some operations that would result in `Inexact(T)` would be:

/// - Applying a filter (selectivity is unknown)

/// - Mixing exact and inexact values in arithmetic

/// - and more...

Inexact(T),

/// Nothing is known about the value. This is the default state.

///

/// Acts as an absorbing element in arithmetic -> any operation

/// involving `Absent` yields `Absent`. [`Precision::to_inexact`]

/// on `Absent` returns `Absent`, not `Inexact` — it represents

/// a fundamentally different state.

///

/// Common sources include:

/// - Data sources without statistics

/// - Parquet columns missing from file metadata

/// - Statistics that cannot be derived for an operation (e.g.,

/// `distinct_count` after a union, `total_byte_size` for joins)

#[default]

Absent,

}

impl<T: Debug + Clone + PartialEq + Eq + PartialOrd> Precision<T> {

/// If we have some value (exact or inexact), it returns that value.

/// Otherwise, it returns `None`.

pub fn get_value(&self) -> Option<&T> {

match self {

Precision::Exact(value) | Precision::Inexact(value) => Some(value),

Precision::Absent => None,

}

}

/// Transform the value in this [`Precision`] object, if one exists, using

/// the given function. Preserves the exactness state.

pub fn map<U, F>(self, f: F) -> Precision<U>

where

F: Fn(T) -> U,

U: Debug + Clone + PartialEq + Eq + PartialOrd,

{

match self {

Precision::Exact(val) => Precision::Exact(f(val)),

Precision::Inexact(val) => Precision::Inexact(f(val)),

_ => Precision::<U>::Absent,

}

}

/// Returns `Some(true)` if we have an exact value, `Some(false)` if we

/// have an inexact value, and `None` if there is no value.

pub fn is_exact(&self) -> Option<bool> {

match self {

Precision::Exact(_) => Some(true),

Precision::Inexact(_) => Some(false),

_ => None,

}

}

/// Returns the maximum of two (possibly inexact) values, conservatively

/// propagating exactness information. If one of the input values is

/// [`Precision::Absent`], the result is `Absent` too.

pub fn max(&self, other: &Precision<T>) -> Precision<T> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => {

Precision::Exact(if a >= b { a.clone() } else { b.clone() })

}

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => {

Precision::Inexact(if a >= b { a.clone() } else { b.clone() })

}

(_, _) => Precision::Absent,

}

}

/// Returns the minimum of two (possibly inexact) values, conservatively

/// propagating exactness information. If one of the input values is

/// [`Precision::Absent`], the result is `Absent` too.

pub fn min(&self, other: &Precision<T>) -> Precision<T> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => {

Precision::Exact(if a >= b { b.clone() } else { a.clone() })

}

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => {

Precision::Inexact(if a >= b { b.clone() } else { a.clone() })

}

(_, _) => Precision::Absent,

}

}

/// Demotes the precision state from exact to inexact (if present).

pub fn to_inexact(self) -> Self {

match self {

Precision::Exact(value) => Precision::Inexact(value),

_ => self,

}

}

}

impl Precision<usize> {

/// Calculates the sum of two (possibly inexact) [`usize`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

pub fn add(&self, other: &Precision<usize>) -> Precision<usize> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => a.checked_add(*b).map_or_else(

|| Precision::Inexact(a.saturating_add(*b)),

Precision::Exact,

),

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => {

Precision::Inexact(a.saturating_add(*b))

}

(_, _) => Precision::Absent,

}

}

/// Calculates the difference of two (possibly inexact) [`usize`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

pub fn sub(&self, other: &Precision<usize>) -> Precision<usize> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => a.checked_sub(*b).map_or_else(

|| Precision::Inexact(a.saturating_sub(*b)),

Precision::Exact,

),

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => {

Precision::Inexact(a.saturating_sub(*b))

}

(_, _) => Precision::Absent,

}

}

/// Calculates the multiplication of two (possibly inexact) [`usize`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

pub fn multiply(&self, other: &Precision<usize>) -> Precision<usize> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => a.checked_mul(*b).map_or_else(

|| Precision::Inexact(a.saturating_mul(*b)),

Precision::Exact,

),

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => {

Precision::Inexact(a.saturating_mul(*b))

}

(_, _) => Precision::Absent,

}

}

/// Return the estimate of applying a filter with estimated selectivity

/// `selectivity` to this Precision. A selectivity of `1.0` means that all

/// rows are selected. A selectivity of `0.5` means half the rows are

/// selected. An exact zero is preserved, since filtering an empty input

/// cannot produce rows; any other known value is demoted to inexact.

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()

}

}

impl Precision<ScalarValue> {

fn sum_data_type(data_type: &DataType) -> DataType {

match data_type {

DataType::Int8 | DataType::Int16 | DataType::Int32 => DataType::Int64,

DataType::UInt8 | DataType::UInt16 | DataType::UInt32 => DataType::UInt64,

_ => data_type.clone(),

}

}

fn cast_scalar_to_sum_type(value: &ScalarValue) -> Result<ScalarValue> {

let source_type = value.data_type();

let target_type = Self::sum_data_type(&source_type);

if source_type == target_type {

Ok(value.clone())

} else {

value.cast_to(&target_type)

}

}

/// Calculates the sum of two (possibly inexact) [`ScalarValue`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

///

/// Uses [`ScalarValue::add_checked`] so that integer overflow returns

/// an error (mapped to `Absent`) instead of silently wrapping.

///

/// For performance-sensitive paths prefer `precision_add` which

/// avoids the Arrow array round-trip.

pub fn add(&self, other: &Precision<ScalarValue>) -> Precision<ScalarValue> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => a

.add_checked(b)

.map(Precision::Exact)

.unwrap_or(Precision::Absent),

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => a

.add_checked(b)

.map(Precision::Inexact)

.unwrap_or(Precision::Absent),

(_, _) => Precision::Absent,

}

}

/// Casts integer values to the wider SQL `SUM` return type.

///

/// This narrows overflow risk when `sum_value` statistics are merged:

/// `Int8/Int16/Int32 -> Int64` and `UInt8/UInt16/UInt32 -> UInt64`.

pub fn cast_to_sum_type(&self) -> Precision<ScalarValue> {

match (self.is_exact(), self.get_value()) {

(Some(true), Some(value)) => Self::cast_scalar_to_sum_type(value)

.map(Precision::Exact)

.unwrap_or(Precision::Absent),

(Some(false), Some(value)) => Self::cast_scalar_to_sum_type(value)

.map(Precision::Inexact)

.unwrap_or(Precision::Absent),

(_, _) => Precision::Absent,

}

}

/// SUM-style addition with integer widening to match SQL `SUM` return

/// types for smaller integral inputs.

pub fn add_for_sum(&self, other: &Precision<ScalarValue>) -> Precision<ScalarValue> {

let mut lhs = self.cast_to_sum_type();

let rhs = other.cast_to_sum_type();

precision_add(&mut lhs, &rhs);

lhs

}

/// Calculates the difference of two (possibly inexact) [`ScalarValue`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

pub fn sub(&self, other: &Precision<ScalarValue>) -> Precision<ScalarValue> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => {

a.sub(b).map(Precision::Exact).unwrap_or(Precision::Absent)

}

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => a

.sub(b)

.map(Precision::Inexact)

.unwrap_or(Precision::Absent),

(_, _) => Precision::Absent,

}

}

/// Calculates the multiplication of two (possibly inexact) [`ScalarValue`] values,

/// conservatively propagating exactness information. If one of the input

/// values is [`Precision::Absent`], the result is `Absent` too.

pub fn multiply(&self, other: &Precision<ScalarValue>) -> Precision<ScalarValue> {

match (self, other) {

(Precision::Exact(a), Precision::Exact(b)) => a

.mul_checked(b)

.map(Precision::Exact)

.unwrap_or(Precision::Absent),

(Precision::Inexact(a), Precision::Exact(b))

| (Precision::Exact(a), Precision::Inexact(b))

| (Precision::Inexact(a), Precision::Inexact(b)) => a

.mul_checked(b)

.map(Precision::Inexact)

.unwrap_or(Precision::Absent),

(_, _) => Precision::Absent,

}

}

/// Casts the value to the given data type, propagating exactness information.

pub fn cast_to(&self, data_type: &DataType) -> Result<Precision<ScalarValue>> {

match self {

Precision::Exact(value) => value.cast_to(data_type).map(Precision::Exact),

Precision::Inexact(value) => value.cast_to(data_type).map(Precision::Inexact),

Precision::Absent => Ok(Precision::Absent),

}

}

}

impl<T: Debug + Clone + PartialEq + Eq + PartialOrd> From<Option<T>> for Precision<T> {

fn from(option: Option<T>) -> Self {

option.map_or(Precision::Absent, Precision::Exact)

}

}

impl<T: Debug + Clone + PartialEq + Eq + PartialOrd> Debug for Precision<T> {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

match self {

Precision::Exact(inner) => write!(f, "Exact({inner:?})"),

Precision::Inexact(inner) => write!(f, "Inexact({inner:?})"),

Precision::Absent => write!(f, "Absent"),

}

}

}

impl<T: Debug + Clone + PartialEq + Eq + PartialOrd> Display for Precision<T> {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

match self {

Precision::Exact(inner) => write!(f, "Exact({inner:?})"),

Precision::Inexact(inner) => write!(f, "Inexact({inner:?})"),

Precision::Absent => write!(f, "Absent"),

}

}

}

impl From<Precision<usize>> for Precision<ScalarValue> {

fn from(value: Precision<usize>) -> Self {

match value {

Precision::Exact(v) => Precision::Exact(ScalarValue::UInt64(Some(v as u64))),

Precision::Inexact(v) => {

Precision::Inexact(ScalarValue::UInt64(Some(v as u64)))

}

Precision::Absent => Precision::Absent,

}

}

}

/// Statistics for a relation

/// Fields are optional and can be inexact because the sources

/// sometimes provide approximate estimates for performance reasons

/// and the transformations output are not always predictable.

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct Statistics {

/// The number of rows estimated to be scanned.

pub num_rows: Precision<usize>,

/// The total bytes of the output data.

///

/// Note that this is not the same as the total bytes that may be scanned,

/// processed, etc.

/// E.g. we may read 1GB of data from a Parquet file but the Arrow data

/// the node produces may be 2GB; it's this 2GB that is tracked here.

pub total_byte_size: Precision<usize>,

/// Statistics on a column level.

///

/// It must contains a [`ColumnStatistics`] for each field in the schema of

/// the table to which the [`Statistics`] refer.

pub column_statistics: Vec<ColumnStatistics>,

}

/// Returns `true` when the statistics prove that the input contains no rows.

///

/// Inexact or absent row counts are not sufficient to treat an input as empty.

pub fn is_known_empty(statistics: &Statistics) -> bool {

statistics.num_rows == Precision::Exact(0)

}

/// Fallback to use when NDV overlap can not be estimated from column bounds.

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]

pub enum NdvFallback {

/// Use the larger input NDV. This is the conservative default for

/// related fragments such as files from the same table.

#[default]

Max,

/// Sum the input NDVs. This is a conservative upper bound for

/// independent inputs such as `UNION ALL`.

Sum,

}

impl NdvFallback {

fn merge(self, left: usize, right: usize) -> usize {

match self {

Self::Max => usize::max(left, right),

Self::Sum => left.saturating_add(right),

}

}

}

impl Default for Statistics {

/// Returns a new [`Statistics`] instance with all fields set to unknown

/// and no columns.

fn default() -> Self {

Self {

num_rows: Precision::Absent,

total_byte_size: Precision::Absent,

column_statistics: vec![],

}

}

}

impl Statistics {

/// Returns a [`Statistics`] instance for the given schema by assigning

/// unknown statistics to each column in the schema.

pub fn new_unknown(schema: &Schema) -> Self {

Self {

num_rows: Precision::Absent,

total_byte_size: Precision::Absent,

column_statistics: Statistics::unknown_column(schema),

}

}

/// Calculates `total_byte_size` based on the schema and `num_rows`.

/// If any of the columns has non-primitive width, or `num_rows` is unknown,

/// the previous `total_byte_size` is kept but downgraded to inexact rather

/// than discarded.

pub fn calculate_total_byte_size(&mut self, schema: &Schema) {

let mut row_size = Some(0);

for field in schema.fields() {

match field.data_type().primitive_width() {

Some(width) => {

row_size = row_size.map(|s| s + width);

}

None => {

row_size = None;

break;

}

}

}

match (row_size, &self.num_rows) {

(None, _) | (Some(_), Precision::Absent) => {

self.total_byte_size = self.total_byte_size.to_inexact();

}

(Some(size), _) => {

self.total_byte_size = self.num_rows.multiply(&Precision::Exact(size));

}

}

}

/// Returns an unbounded `ColumnStatistics` for each field in the schema.

pub fn unknown_column(schema: &Schema) -> Vec<ColumnStatistics> {

schema

.fields()

.iter()

.map(|_| ColumnStatistics::new_unknown())

.collect()

}

/// Set the number of rows

pub fn with_num_rows(mut self, num_rows: Precision<usize>) -> Self {

self.num_rows = num_rows;

self

}

/// Set the total size, in bytes

pub fn with_total_byte_size(mut self, total_byte_size: Precision<usize>) -> Self {

self.total_byte_size = total_byte_size;

self

}

/// Add a column to the column statistics

pub fn add_column_statistics(mut self, column_stats: ColumnStatistics) -> Self {

self.column_statistics.push(column_stats);

self

}

/// If the exactness of a [`Statistics`] instance is lost, this function relaxes

/// the exactness of all information by converting them [`Precision::Inexact`].

pub fn to_inexact(mut self) -> Self {

self.num_rows = self.num_rows.to_inexact();

self.total_byte_size = self.total_byte_size.to_inexact();

self.column_statistics = self

.column_statistics

.into_iter()

.map(|s| s.to_inexact())

.collect();

self

}

/// Project the statistics to the given column indices.

///

/// For example, if we had statistics for columns `{"a", "b", "c"}`,

/// projecting to `vec![2, 1]` would return statistics for columns `{"c",

/// "b"}`.

pub fn project(self, projection: Option<&impl AsRef<[usize]>>) -> Self {

let projection = projection.map(AsRef::as_ref);

self.project_impl(projection)

}

fn project_impl(mut self, projection: Option<&[usize]>) -> Self {

let Some(projection) = projection.map(AsRef::as_ref) else {

return self;

};

#[expect(clippy::large_enum_variant)]

enum Slot {

/// The column is taken and put into the specified statistics location

Taken(usize),

/// The original columns is present

Present(ColumnStatistics),

}

// Convert to Vec<Slot> so we can avoid copying the statistics

let mut columns: Vec<_> = std::mem::replace(

&mut self.column_statistics,

Vec::with_capacity(projection.len()),

)

.into_iter()

.map(Slot::Present)

.collect();

for idx in projection.iter() {

let next_idx = self.column_statistics.len();

let slot = std::mem::replace(

columns.get_mut(*idx).expect("projection out of bounds"),

Slot::Taken(next_idx),

);

match slot {

// The column was there, so just move it

Slot::Present(col) => self.column_statistics.push(col),

// The column was taken, so copy from the previous location

Slot::Taken(prev_idx) => self

.column_statistics

.push(self.column_statistics[prev_idx].clone()),

}

}

self

}

/// Calculates the statistics after applying `fetch` and `skip` operations.

///

/// Here, `self` denotes per-partition statistics. Use the `n_partitions`

/// parameter to compute global statistics in a multi-partition setting.

pub fn with_fetch(

mut self,

fetch: Option<usize>,

skip: usize,

n_partitions: usize,

) -> Result<Self> {

if fetch.is_none() && skip == 0 {

return Ok(self);

}

let fetch_val = fetch.unwrap_or(usize::MAX);

// Get the ratio of rows after / rows before on a per-partition basis

let num_rows_before = self.num_rows;

self.num_rows = match self {

Statistics {

num_rows: Precision::Exact(nr),

..

}

| Statistics {

num_rows: Precision::Inexact(nr),

..

} => {

// Here, the inexact case gives us an estimate of the number of rows.

if nr <= skip {

// All input data will be skipped. Preserve the exactness of

// the input estimate: if the input was inexact, the

// resulting zero is also inexact.

check_num_rows(Some(0), self.num_rows.is_exact().unwrap())

} else if nr <= fetch_val && skip == 0 {

// If the input does not reach the `fetch` globally, and `skip`

// is zero (meaning the input and output are identical), return

// input stats as is.

// TODO: Can input stats still be used, but adjusted, when `skip`

// is non-zero?

return Ok(self);

} else if nr - skip <= fetch_val {

// After `skip` input rows are skipped, the remaining rows are

// less than or equal to the `fetch` values, so `num_rows` must

// equal the remaining rows.

check_num_rows(

(nr - skip).checked_mul(n_partitions),

// We know that we have an estimate for the number of rows:

self.num_rows.is_exact().unwrap(),

)

} else {

// At this point we know that we were given a `fetch` value

// as the `None` case would go into the branch above. Since

// the input has more rows than `fetch + skip`, the number

// of rows will be the `fetch`, other statistics will have to be downgraded to inexact.

check_num_rows(

fetch_val.checked_mul(n_partitions),

// We know that we have an estimate for the number of rows:

self.num_rows.is_exact().unwrap(),

)

}

}

Statistics {

num_rows: Precision::Absent,

..

} => check_num_rows(fetch.and_then(|v| v.checked_mul(n_partitions)), false),

};

let ratio: Option<f64> = match (num_rows_before, self.num_rows) {

(

Precision::Exact(nr_before) | Precision::Inexact(nr_before),

Precision::Exact(nr_after) | Precision::Inexact(nr_after),

) => {

if nr_before == 0 {

Some(0.0)

} else {

Some(nr_after as f64 / nr_before as f64)

}

}

_ => None,

};

self.column_statistics = self

.column_statistics

.into_iter()

.map(|cs| {

let mut cs = cs.to_inexact();

// Scale byte_size by the row ratio

cs.byte_size = match (cs.byte_size, ratio) {

(Precision::Exact(n) | Precision::Inexact(n), Some(ratio)) => {

Precision::Inexact((n as f64 * ratio) as usize)

}

_ => Precision::Absent,

};

// NDV can never exceed the number of rows

if let Some(&rows) = self.num_rows.get_value() {

cs.distinct_count = cs.distinct_count.min(&Precision::Inexact(rows));

}

cs

})

.collect();

// Compute total_byte_size as sum of column byte_size values if all are present,

// otherwise fall back to scaling the original total_byte_size

let sum_scan_bytes: Option<usize> = self

.column_statistics

.iter()

.map(|cs| cs.byte_size.get_value().copied())

.try_fold(0usize, |acc, val| val.map(|v| acc + v));

self.total_byte_size = match sum_scan_bytes {

Some(sum) => Precision::Inexact(sum),

None => {

// Fall back to scaling original total_byte_size if not all columns have byte_size

match (&self.total_byte_size, ratio) {

(Precision::Exact(n) | Precision::Inexact(n), Some(ratio)) => {

Precision::Inexact((*n as f64 * ratio) as usize)

}

_ => Precision::Absent,

}

}

};

Ok(self)

}

/// Summarize zero or more statistics into a single `Statistics` instance.

///

/// The method assumes that all statistics are for the same schema.

/// If not, maybe you can call `SchemaMapper::map_column_statistics` to make them consistent.

///

/// This method uses [`NdvFallback::Max`] when `distinct_count` overlap

/// can not be estimated from column bounds.

///

/// Returns an error if the statistics do not match the specified schemas.

///

/// # Example

/// ```

/// # use datafusion_common::{ColumnStatistics, ScalarValue, Statistics};

/// # use arrow::datatypes::{Field, Schema, DataType};

/// # use datafusion_common::stats::Precision;

/// let stats1 = Statistics::default()

/// .with_num_rows(Precision::Exact(10))

/// .add_column_statistics(

/// ColumnStatistics::new_unknown()

/// .with_min_value(Precision::Exact(ScalarValue::from(1)))

/// .with_max_value(Precision::Exact(ScalarValue::from(100)))

/// .with_sum_value(Precision::Exact(ScalarValue::from(500))),

/// );

///

/// let stats2 = Statistics::default()

/// .with_num_rows(Precision::Exact(20))

/// .add_column_statistics(

/// ColumnStatistics::new_unknown()

/// .with_min_value(Precision::Exact(ScalarValue::from(5)))

/// .with_max_value(Precision::Exact(ScalarValue::from(200)))

/// .with_sum_value(Precision::Exact(ScalarValue::from(1000))),

/// );

///

/// let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);

/// let merged = Statistics::try_merge_iter(

/// &[stats1, stats2],

/// &schema,

/// ).unwrap();

///

/// assert_eq!(merged.num_rows, Precision::Exact(30));

/// assert_eq!(merged.column_statistics[0].min_value,

/// Precision::Exact(ScalarValue::from(1)));

/// assert_eq!(merged.column_statistics[0].max_value,

/// Precision::Exact(ScalarValue::from(200)));

/// assert_eq!(merged.column_statistics[0].sum_value,

/// Precision::Exact(ScalarValue::Int64(Some(1500))));

/// ```

pub fn try_merge_iter<'a, I>(items: I, schema: &Schema) -> Result<Statistics>

where

I: IntoIterator<Item = &'a Statistics>,

{

Self::try_merge_iter_with_ndv_fallback(items, schema, NdvFallback::Max)

}

/// Same as [`Statistics::try_merge_iter`], but lets callers choose the

/// fallback used when `distinct_count` overlap can not be estimated.

pub fn try_merge_iter_with_ndv_fallback<'a, I>(

items: I,

schema: &Schema,

ndv_fallback: NdvFallback,

) -> Result<Statistics>

where

I: IntoIterator<Item = &'a Statistics>,

{

let mut items = items.into_iter();

let Some(first) = items.next() else {

return Ok(Statistics::new_unknown(schema));

};

let Some(second) = items.next() else {

return Ok(first.clone());

};

let num_cols = first.column_statistics.len();

let mut num_rows = first.num_rows;

let mut total_byte_size = first.total_byte_size;

let mut column_statistics = first.column_statistics.clone();

for col_stats in &mut column_statistics {

cast_sum_value_to_sum_type_in_place(&mut col_stats.sum_value);

}

// Merge the remaining items in a single pass.

for (i, stat) in std::iter::once(second).chain(items).enumerate() {

if stat.column_statistics.len() != num_cols {

return _plan_err!(

"Cannot merge statistics with different number of columns: {} vs {} (item {})",

num_cols,

stat.column_statistics.len(),

i + 1

);

}

num_rows = num_rows.add(&stat.num_rows);

total_byte_size = total_byte_size.add(&stat.total_byte_size);

// Uses precision_add for sum (reuses the lhs accumulator for

// direct numeric addition), while preserving the NDV update

// ordering required by estimate_ndv_with_overlap.

for (col_stats, item_cs) in

column_statistics.iter_mut().zip(&stat.column_statistics)

{

col_stats.null_count = col_stats.null_count.add(&item_cs.null_count);

// NDV must be computed before min/max update (needs pre-merge ranges)

col_stats.distinct_count = match (

col_stats.distinct_count.get_value(),

item_cs.distinct_count.get_value(),

) {

(Some(&l), Some(&r)) => Precision::Inexact(

estimate_ndv_with_overlap(col_stats, item_cs, l, r)

.unwrap_or_else(|| ndv_fallback.merge(l, r)),

),

_ => Precision::Absent,

};

precision_min(&mut col_stats.min_value, &item_cs.min_value);

precision_max(&mut col_stats.max_value, &item_cs.max_value);

precision_add_for_sum_in_place(

&mut col_stats.sum_value,

&item_cs.sum_value,

);

col_stats.byte_size = col_stats.byte_size.add(&item_cs.byte_size);

}

}

Ok(Statistics {

num_rows,

total_byte_size,

column_statistics,

})

}

}

/// Estimates the combined number of distinct values (NDV) when merging two

/// column statistics, using range overlap to avoid double-counting shared values.

///

/// Assumes values are distributed uniformly within each input's

/// `[min, max]` range (the standard assumption when only summary

/// statistics are available). Under uniformity the fraction of an input's

/// distinct values that land in a sub-range equals the fraction of

/// the range that sub-range covers.

///

/// The combined value space is split into three disjoint regions:

///

/// ```text

/// |-- only A --|-- overlap --|-- only B --|

/// ```

///

/// * **Only in A/B** - values outside the other input's range

/// contribute `(1 - overlap_a) * NDV_a` and `(1 - overlap_b) * NDV_b`.

/// * **Overlap** - both inputs may produce values here. We take

/// `max(overlap_a * NDV_a, overlap_b * NDV_b)` rather than the

/// sum because values in the same sub-range are likely shared

/// (the smaller set is assumed to be a subset of the larger).

///

/// The formula ranges between `[max(NDV_a, NDV_b), NDV_a + NDV_b]`,

/// from full overlap to no overlap.

///

/// ```text

/// NDV = max(overlap_a * NDV_a, overlap_b * NDV_b) [intersection]

/// + (1 - overlap_a) * NDV_a [only in A]

/// + (1 - overlap_b) * NDV_b [only in B]

/// ```

///

/// Returns `None` when min/max are absent or distance is unsupported

/// (e.g. strings), in which case the caller should fall back to a simpler

/// estimate.

pub fn estimate_ndv_with_overlap(

left: &ColumnStatistics,

right: &ColumnStatistics,

ndv_left: usize,

ndv_right: usize,

) -> Option<usize> {

let left_min = left.min_value.get_value()?;

let left_max = left.max_value.get_value()?;

let right_min = right.min_value.get_value()?;

let right_max = right.max_value.get_value()?;

let range_left = left_max.distance_u64(left_min)?;

let range_right = right_max.distance_u64(right_min)?;

// Constant columns (range == 0) can't use the proportional overlap

// formula below, so check interval overlap directly instead.

if range_left == 0 || range_right == 0 {

let overlaps = left_min <= right_max && right_min <= left_max;

return Some(if overlaps {

usize::max(ndv_left, ndv_right)

} else {

ndv_left + ndv_right

});

}

let overlap_min = if left_min >= right_min {

left_min

} else {

right_min

};

let overlap_max = if left_max <= right_max {

left_max

} else {

right_max

};

// Disjoint ranges: no overlap, NDVs are additive

if overlap_min > overlap_max {

return Some(ndv_left + ndv_right);

}

let overlap_range = overlap_max.distance_u64(overlap_min)? as f64;

let overlap_left = overlap_range / range_left as f64;

let overlap_right = overlap_range / range_right as f64;

let intersection = f64::max(

overlap_left * ndv_left as f64,

overlap_right * ndv_right as f64,

);

let only_left = (1.0 - overlap_left) * ndv_left as f64;

let only_right = (1.0 - overlap_right) * ndv_right as f64;

Some((intersection + only_left + only_right).round() as usize)

}

/// Returns the minimum precision while not allocating a new value,

/// mirrors the semantics of `PartialOrd`.

#[inline]

fn precision_min<T>(lhs: &mut Precision<T>, rhs: &Precision<T>)

where

T: Debug + Clone + PartialEq + Eq + PartialOrd,

{

*lhs = match (std::mem::take(lhs), rhs) {

(Precision::Exact(left), Precision::Exact(right)) => {

if left <= *right {

Precision::Exact(left)

} else {

Precision::Exact(right.clone())

}

}

(Precision::Exact(left), Precision::Inexact(right))

| (Precision::Inexact(left), Precision::Exact(right))

| (Precision::Inexact(left), Precision::Inexact(right)) => {

if left <= *right {

Precision::Inexact(left)

} else {

Precision::Inexact(right.clone())

}

}

(_, _) => Precision::Absent,

};

}

/// Returns the maximum precision while not allocating a new value,

/// mirrors the semantics of `PartialOrd`.

#[inline]

fn precision_max<T>(lhs: &mut Precision<T>, rhs: &Precision<T>)

where

T: Debug + Clone + PartialEq + Eq + PartialOrd,

{

*lhs = match (std::mem::take(lhs), rhs) {

(Precision::Exact(left), Precision::Exact(right)) => {

if left >= *right {

Precision::Exact(left)

} else {

Precision::Exact(right.clone())

}

}

(Precision::Exact(left), Precision::Inexact(right))

| (Precision::Inexact(left), Precision::Exact(right))

| (Precision::Inexact(left), Precision::Inexact(right)) => {

if left >= *right {

Precision::Inexact(left)

} else {

Precision::Inexact(right.clone())

}

}

(_, _) => Precision::Absent,

};

}

#[inline]

fn cast_sum_value_to_sum_type_in_place(value: &mut Precision<ScalarValue>) {

let (is_exact, inner) = match std::mem::take(value) {

Precision::Exact(v) => (true, v),

Precision::Inexact(v) => (false, v),

Precision::Absent => return,

};

let source_type = inner.data_type();

let target_type = Precision::<ScalarValue>::sum_data_type(&source_type);

let wrap_precision_fn: fn(ScalarValue) -> Precision<ScalarValue> = if is_exact {

Precision::Exact

} else {

Precision::Inexact

};

*value = if source_type == target_type {

wrap_precision_fn(inner)

} else {

inner

.cast_to(&target_type)

.map(wrap_precision_fn)

.unwrap_or(Precision::Absent)

};

}

#[inline]

fn precision_add_for_sum_in_place(

lhs: &mut Precision<ScalarValue>,

rhs: &Precision<ScalarValue>,

) {

let (value, wrap_fn): (&ScalarValue, fn(ScalarValue) -> Precision<ScalarValue>) =

match rhs {

Precision::Exact(v) => (v, Precision::Exact),

Precision::Inexact(v) => (v, Precision::Inexact),

Precision::Absent => {

*lhs = Precision::Absent;

return;

}

};

let source_type = value.data_type();

let target_type = Precision::<ScalarValue>::sum_data_type(&source_type);

if source_type == target_type {

precision_add(lhs, rhs);

} else {

let rhs = value

.cast_to(&target_type)

.map(wrap_fn)

.unwrap_or(Precision::Absent);

precision_add(lhs, &rhs);

}

}

/// Creates an estimate of the number of rows in the output using the given

/// optional value and exactness flag.

fn check_num_rows(value: Option<usize>, is_exact: bool) -> Precision<usize> {

if let Some(value) = value {

if is_exact {

Precision::Exact(value)

} else {

// If the input stats are inexact, so are the output stats.