feat: add aggregate-stats foundation types and helpers

- Add SupportedStatAggr, StatsAwareFallbackReason, StatsAwareFileDecision,
  StatsAwareFileDecisionSnapshot, FileStatsItem, RowGroupStatsItem, and
  SendableFileStatsStream to store-api as pure data types
- Add aggr_stats foundation module to common-query with requirement matching,
  partition filtering, and column-stats collection from parquet metadata
- Add RowGroupPruningStatistics to table::predicate for parquet row-group pruning
- Promote table predicate stats from cfg(test) to pub(crate)
- Drive parquet physical-to-logical scalar conversion with Arrow DataType
- Add bounds checks for schema-evolved parquet columns

This is PR1 of a multi-PR split: data types and foundation only.
No scanner runtime, optimizer rewrites, or sqlness.

Signed-off-by: discord9 <discord9@163.com>

Author: discord9

Cargo.lock

@@ -0,0 +1,382 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::cmp::Ordering;
+use std::collections::{HashMap, HashSet};
+
+use datafusion::parquet::file::statistics::Statistics as ParquetStats;
+use datafusion::scalar::ScalarValue;
+use datafusion_common::{DataFusionError, Result};
+use datatypes::arrow::datatypes::{DataType, TimeUnit};
+use datatypes::schema::SchemaRef as RegionSchemaRef;
+use datatypes::value::Value;
+use store_api::region_engine::{FileStatsItem, SupportedStatAggr};
+
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct FileColumnStats {
+    pub null_count: Option<u64>,
+    pub min_value: Option<Value>,
+    pub max_value: Option<Value>,
+}
+
+#[derive(Debug, Clone, Default, PartialEq)]
+pub struct StatsCandidateFile {
+    pub num_rows: Option<u64>,
+    pub column_stats: HashMap<String, FileColumnStats>,
+}
+
+impl StatsCandidateFile {
+    pub fn from_file_stats(
+        file_stats: &FileStatsItem,
+        region_partition_expr: Option<&str>,
+        requirements: &[SupportedStatAggr],
+        region_schema: &RegionSchemaRef,
+    ) -> Result<Option<Self>> {
+        let column_names = required_columns(requirements);
+        if !matches_partition_expr(
+            file_stats.file_partition_expr.as_deref(),
+            region_partition_expr,
+        ) {
+            return Ok(None);
+        }
+        let column_stats = collect_column_stats(file_stats, region_schema, &column_names)?;
+
+        let candidate = Self {
+            num_rows: file_stats.num_rows,
+            column_stats,
+        };
+        for requirement in requirements {
+            if candidate.stat_value(requirement)?.is_none() {
+                return Ok(None);
+            }
+        }
+        Ok(Some(candidate))
+    }
+
+    pub fn stat_value(&self, requirement: &SupportedStatAggr) -> Result<Option<Value>> {
+        match requirement {
+            SupportedStatAggr::CountRows => self.num_rows.map(count_value).transpose(),
+            SupportedStatAggr::CountNonNull { column_name } => {
+                let Some(column_stats) = self.column_stats.get(column_name) else {
+                    return Ok(None);
+                };
+                let Some(num_rows) = self.num_rows else {
+                    return Ok(None);
+                };
+                let Some(null_count) = column_stats.null_count else {
+                    return Ok(None);
+                };
+                let Some(non_null_count) = num_rows.checked_sub(null_count) else {
+                    return Err(DataFusionError::Internal(format!(
+                        "StatsScanExec found null_count > num_rows for column {}",
+                        column_name
+                    )));
+                };
+                count_value(non_null_count).map(Some)
+            }
+            SupportedStatAggr::MinValue { column_name } => Ok(self
+                .column_stats
+                .get(column_name)
+                .and_then(|stats| stats.min_value.clone())),
+            SupportedStatAggr::MaxValue { column_name } => Ok(self
+                .column_stats
+                .get(column_name)
+                .and_then(|stats| stats.max_value.clone())),
+        }
+    }
+}
+
+fn count_value(value: u64) -> Result<Value> {
+    let value = i64::try_from(value).map_err(|_| {
+        DataFusionError::Internal(format!(
+            "StatsScanExec count state exceeds Int64 range: {}",
+            value
+        ))
+    })?;
+    Ok(Value::Int64(value))
+}
+
+fn matches_partition_expr(
+    file_partition_expr: Option<&str>,
+    region_partition_expr: Option<&str>,
+) -> bool {
+    match (file_partition_expr, region_partition_expr) {
+        (Some(file_expr), Some(region_expr)) => file_expr == region_expr,
+        (None, None) => true,
+        _ => false,
+    }
+}
+
+fn required_columns(requirements: &[SupportedStatAggr]) -> HashSet<String> {
+    requirements
+        .iter()
+        .filter_map(|requirement| match requirement {
+            SupportedStatAggr::CountRows => None,
+            SupportedStatAggr::CountNonNull { column_name }
+            | SupportedStatAggr::MinValue { column_name }
+            | SupportedStatAggr::MaxValue { column_name } => Some(column_name.clone()),
+        })
+        .collect()
+}
+
+fn collect_column_stats(
+    file_stats: &FileStatsItem,
+    region_schema: &RegionSchemaRef,
+    column_names: &HashSet<String>,
+) -> Result<HashMap<String, FileColumnStats>> {
+    column_names
+        .iter()
+        .map(|column_name| {
+            Ok((
+                column_name.clone(),
+                collect_one_column_stats(file_stats, region_schema, column_name)?,
+            ))
+        })
+        .collect()
+}
+
+fn collect_one_column_stats(
+    file_stats: &FileStatsItem,
+    region_schema: &RegionSchemaRef,
+    column_name: &str,
+) -> Result<FileColumnStats> {
+    let Some(column_index) = region_schema.column_index_by_name(column_name) else {
+        return Ok(FileColumnStats::default());
+    };
+
+    let arrow_type = region_schema.arrow_schema().field(column_index).data_type();
+
+    Ok(FileColumnStats {
+        null_count: sum_null_counts(file_stats, column_index)?,
+        min_value: best_row_group_value(file_stats, column_index, Ordering::Less, arrow_type)?,
+        max_value: best_row_group_value(file_stats, column_index, Ordering::Greater, arrow_type)?,
+    })
+}
+
+fn sum_null_counts(file_stats: &FileStatsItem, column_index: usize) -> Result<Option<u64>> {
+    if file_stats.row_groups.is_empty() {
+        return Ok(None);
+    }
+
+    let mut total = 0_u64;
+    for row_group in &file_stats.row_groups {
+        if column_index >= row_group.metadata.num_columns() {
+            return Ok(None);
+        }
+        let Some(stats) = row_group.metadata.column(column_index).statistics() else {
+            return Ok(None);
+        };
+        let Some(value) = stats.null_count_opt() else {
+            return Ok(None);
+        };
+        total = total.checked_add(value).ok_or_else(|| {
+            DataFusionError::Internal("StatsScanExec null-count overflow".to_string())
+        })?;
+    }
+    Ok(Some(total))
+}
+
+fn best_row_group_value(
+    file_stats: &FileStatsItem,
+    column_index: usize,
+    target: Ordering,
+    arrow_type: &DataType,
+) -> Result<Option<Value>> {
+    let mut best = None;
+
+    for row_group in &file_stats.row_groups {
+        if column_index >= row_group.metadata.num_columns() {
+            return Ok(None);
+        }
+        let Some(stats) = row_group.metadata.column(column_index).statistics() else {
+            return Ok(None);
+        };
+        let Some(scalar) = parquet_bound_scalar(stats, target, arrow_type) else {
+            return Ok(None);
+        };
+        let value = Value::try_from(scalar).map_err(|error| {
+            DataFusionError::Internal(format!(
+                "StatsScanExec failed to convert row-group scalar: {}",
+                error
+            ))
+        })?;
+        let should_replace = best.as_ref().is_none_or(|current| {
+            value
+                .partial_cmp(current)
+                .is_some_and(|ordering| ordering == target)
+        });
+        if should_replace {
+            best = Some(value);
+        }
+    }
+
+    Ok(best)
+}
+
+fn parquet_bound_scalar(
+    stats: &ParquetStats,
+    target: Ordering,
+    arrow_type: &DataType,
+) -> Option<ScalarValue> {
+    let use_min = target == Ordering::Less;
+
+    match stats {
+        ParquetStats::Boolean(stats) => {
+            if !matches!(arrow_type, DataType::Boolean) {
+                return None;
+            }
+            Some(ScalarValue::Boolean(Some(if use_min {
+                *stats.min_opt()?
+            } else {
+                *stats.max_opt()?
+            })))
+        }
+        ParquetStats::Int32(stats) => {
+            let raw = if use_min {
+                *stats.min_opt()?
+            } else {
+                *stats.max_opt()?
+            };
+            match arrow_type {
+                DataType::Int32
+                | DataType::UInt32
+                | DataType::Int16
+                | DataType::UInt16
+                | DataType::Int8
+                | DataType::UInt8
+                | DataType::Time32(_) => Some(ScalarValue::Int32(Some(raw))),
+                DataType::Date32 => Some(ScalarValue::Date32(Some(raw))),
+                _ => None,
+            }
+        }
+        ParquetStats::Int64(stats) => {
+            let raw = if use_min {
+                *stats.min_opt()?
+            } else {
+                *stats.max_opt()?
+            };
+            match arrow_type {
+                DataType::Int64 | DataType::UInt64 => Some(ScalarValue::Int64(Some(raw))),
+                DataType::Timestamp(TimeUnit::Second, _) => {
+                    Some(ScalarValue::TimestampSecond(Some(raw), None))
+                }
+                DataType::Timestamp(TimeUnit::Millisecond, _) => {
+                    Some(ScalarValue::TimestampMillisecond(Some(raw), None))
+                }
+                DataType::Timestamp(TimeUnit::Microsecond, _) => {
+                    Some(ScalarValue::TimestampMicrosecond(Some(raw), None))
+                }
+                DataType::Timestamp(TimeUnit::Nanosecond, _) => {
+                    Some(ScalarValue::TimestampNanosecond(Some(raw), None))
+                }
+                DataType::Date64 => Some(ScalarValue::Date64(Some(raw))),
+                DataType::Duration(_) => Some(ScalarValue::Int64(Some(raw))),
+                _ => None,
+            }
+        }
+        ParquetStats::Int96(_) => None,
+        ParquetStats::Float(stats) => {
+            if !matches!(arrow_type, DataType::Float32) {
+                return None;
+            }
+            Some(ScalarValue::Float32(Some(if use_min {
+                *stats.min_opt()?
+            } else {
+                *stats.max_opt()?
+            })))
+        }
+        ParquetStats::Double(stats) => {
+            if !matches!(arrow_type, DataType::Float64) {
+                return None;
+            }
+            Some(ScalarValue::Float64(Some(if use_min {
+                *stats.min_opt()?
+            } else {
+                *stats.max_opt()?
+            })))
+        }
+        ParquetStats::ByteArray(stats) => {
+            let bytes = if use_min {
+                stats.min_bytes_opt()?
+            } else {
+                stats.max_bytes_opt()?
+            };
+            match arrow_type {
+                DataType::Utf8 | DataType::LargeUtf8 => String::from_utf8(bytes.to_owned())
+                    .ok()
+                    .map(|s| ScalarValue::Utf8(Some(s))),
+                _ => None,
+            }
+        }
+        ParquetStats::FixedLenByteArray(_) => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use datatypes::schema::Schema;
+
+    use super::*;
+
+    #[test]
+    fn stats_candidate_accepts_unpartitioned_file_for_count_rows() {
+        let file_stats = FileStatsItem {
+            file_id: "file-1".to_string(),
+            num_rows: Some(42),
+            file_partition_expr: None,
+            row_groups: vec![],
+        };
+        let region_schema: RegionSchemaRef = Arc::new(Schema::new(vec![]));
+
+        let candidate = StatsCandidateFile::from_file_stats(
+            &file_stats,
+            None,
+            &[SupportedStatAggr::CountRows],
+            &region_schema,
+        )
+        .expect("count rows stats should be readable");
+
+        assert!(candidate.is_some());
+        assert_eq!(
+            candidate
+                .unwrap()
+                .stat_value(&SupportedStatAggr::CountRows)
+                .expect("count rows should be convertible"),
+            Some(Value::Int64(42))
+        );
+    }
+
+    #[test]
+    fn stats_candidate_rejects_partition_mismatch() {
+        let file_stats = FileStatsItem {
+            file_id: "file-2".to_string(),
+            num_rows: Some(42),
+            file_partition_expr: Some("host = 'a'".to_string()),
+            row_groups: vec![],
+        };
+        let region_schema: RegionSchemaRef = Arc::new(Schema::new(vec![]));
+
+        let candidate = StatsCandidateFile::from_file_stats(
+            &file_stats,
+            Some("host = 'b'"),
+            &[SupportedStatAggr::CountRows],
+            &region_schema,
+        )
+        .expect("count rows stats should be readable");
+
+        assert!(candidate.is_none());
+    }
+}

src/common/query/src/aggr_stats.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod aggr_stats;
 pub mod columnar_value;
 pub mod error;
 pub mod logical_plan;