fix: respect inexact flags in row group metadata

Author: CookiePieWw

datafusion/core/tests/parquet/file_statistics.rs

@@ -28,7 +28,7 @@ use datafusion::execution::context::SessionState;
 use datafusion::execution::session_state::SessionStateBuilder;
 use datafusion::prelude::SessionContext;
 use datafusion_common::stats::Precision;
-use datafusion_common::DFSchema;
+use datafusion_common::{DFSchema, ScalarValue};
 use datafusion_execution::cache::cache_manager::CacheManagerConfig;
 use datafusion_execution::cache::cache_unit::{
     DefaultFileStatisticsCache, DefaultListFilesCache,
@@ -100,6 +100,35 @@ async fn check_stats_precision_with_filter_pushdown() {
     );
 }
 
+#[tokio::test]
+async fn check_stats_inexact_for_truncated_values() {
+    let filename = "long_strings.parquet";
+    let table_path = ListingTableUrl::parse(filename).unwrap();
+
+    let opt =
+        ListingOptions::new(Arc::new(ParquetFormat::default())).with_collect_stat(true);
+    let table = get_listing_table(&table_path, None, &opt).await;
+
+    let (_, _, state) = get_cache_runtime_state();
+
+    let exec = table.scan(&state, None, &[], None).await.unwrap();
+    assert_eq!(
+        exec.partition_statistics(None).unwrap().num_rows,
+        Precision::Exact(2),
+        "Stats without filter should be exact"
+    );
+
+    let col_stats = &exec.partition_statistics(None).unwrap().column_statistics[0];
+    assert_eq!(
+        col_stats.min_value,
+        Precision::Inexact(ScalarValue::Utf8View(Some("A".repeat(4096))))
+    );
+    assert_eq!(
+        col_stats.max_value,
+        Precision::Inexact(ScalarValue::Utf8View(Some("Z".repeat(4095) + "[")))
+    );
+}
+
 #[tokio::test]
 async fn load_table_stats_with_session_level_cache() {
     let testdata = datafusion::test_util::parquet_test_data();

datafusion/datasource-parquet/src/file_format.rs

@@ -19,13 +19,13 @@
 
 use std::any::Any;
 use std::cell::RefCell;
-use std::fmt;
 use std::fmt::Debug;
 use std::ops::Range;
 use std::rc::Rc;
 use std::sync::Arc;
+use std::{fmt, vec};
 
-use arrow::array::RecordBatch;
+use arrow::array::{ArrayRef, BooleanArray, RecordBatch};
 use arrow::datatypes::{Fields, Schema, SchemaRef, TimeUnit};
 use datafusion_datasource::file_compression_type::FileCompressionType;
 use datafusion_datasource::file_sink_config::{FileSink, FileSinkConfig};
@@ -36,7 +36,8 @@ use datafusion_datasource::write::{
 use datafusion_datasource::file_format::{FileFormat, FileFormatFactory};
 use datafusion_datasource::write::demux::DemuxedStreamReceiver;
 
-use arrow::compute::sum;
+use arrow::compute::kernels::cmp::eq;
+use arrow::compute::{and, sum};
 use arrow::datatypes::{DataType, Field, FieldRef};
 use datafusion_common::config::{ConfigField, ConfigFileType, TableParquetOptions};
 #[cfg(feature = "parquet_encryption")]
@@ -46,7 +47,7 @@ use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::stats::Precision;
 use datafusion_common::{
     internal_datafusion_err, internal_err, not_impl_err, ColumnStatistics,
-    DataFusionError, GetExt, HashSet, Result, DEFAULT_PARQUET_EXTENSION,
+    DataFusionError, GetExt, HashSet, Result, ScalarValue, DEFAULT_PARQUET_EXTENSION,
 };
 use datafusion_common::{HashMap, Statistics};
 use datafusion_common_runtime::{JoinSet, SpawnedTask};
@@ -1170,7 +1171,8 @@ pub async fn fetch_statistics(
 /// # When only some columns have statistics:
 ///
 /// For columns with statistics:
-/// - Min/max values are properly extracted and represented as Precision::Exact
+/// - Min/max values are properly extracted and represented as [Precision::Exact] or [Precision::Inexact]
+///   depending on the `is_max_value_exact` and `is_min_value_exact` flags.
 /// - Null counts are calculated by summing across row groups
 ///
 /// For columns without statistics,
@@ -1216,6 +1218,8 @@ pub fn statistics_from_parquet_meta_calc(
         let (mut max_accs, mut min_accs) = create_max_min_accs(&table_schema);
         let mut null_counts_array =
             vec![Precision::Exact(0); table_schema.fields().len()];
+        let mut is_max_value_exact = vec![Some(true); table_schema.fields().len()];
+        let mut is_min_value_exact = vec![Some(true); table_schema.fields().len()];
 
         table_schema
             .fields()
@@ -1232,6 +1236,8 @@ pub fn statistics_from_parquet_meta_calc(
                             &mut min_accs,
                             &mut max_accs,
                             &mut null_counts_array,
+                            &mut is_min_value_exact,
+                            &mut is_max_value_exact,
                             idx,
                             num_rows,
                             &stats_converter,
@@ -1251,6 +1257,8 @@ pub fn statistics_from_parquet_meta_calc(
             null_counts_array,
             &mut max_accs,
             &mut min_accs,
+            &mut is_max_value_exact,
+            &mut is_min_value_exact,
         )
     } else {
         Statistics::unknown_column(&table_schema)
@@ -1264,21 +1272,39 @@ fn get_col_stats(
     null_counts: Vec<Precision<usize>>,
     max_values: &mut [Option<MaxAccumulator>],
     min_values: &mut [Option<MinAccumulator>],
+    is_max_value_exact: &mut [Option<bool>],
+    is_min_value_exact: &mut [Option<bool>],
 ) -> Vec<ColumnStatistics> {
     (0..schema.fields().len())
         .map(|i| {
-            let max_value = match max_values.get_mut(i).unwrap() {
-                Some(max_value) => max_value.evaluate().ok(),
-                None => None,
+            let max_value = match (
+                max_values.get_mut(i).unwrap(),
+                is_max_value_exact.get(i).unwrap(),
+            ) {
+                (Some(max_value), Some(true)) => {
+                    max_value.evaluate().ok().map(Precision::Exact)
+                }
+                (Some(max_value), Some(false)) | (Some(max_value), None) => {
+                    max_value.evaluate().ok().map(Precision::Inexact)
+                }
+                (None, _) => None,
             };
-            let min_value = match min_values.get_mut(i).unwrap() {
-                Some(min_value) => min_value.evaluate().ok(),
-                None => None,
+            let min_value = match (
+                min_values.get_mut(i).unwrap(),
+                is_min_value_exact.get(i).unwrap(),
+            ) {
+                (Some(min_value), Some(true)) => {
+                    min_value.evaluate().ok().map(Precision::Exact)
+                }
+                (Some(min_value), Some(false)) | (Some(min_value), None) => {
+                    min_value.evaluate().ok().map(Precision::Inexact)
+                }
+                (None, _) => None,
             };
             ColumnStatistics {
                 null_count: null_counts[i],
-                max_value: max_value.map(Precision::Exact).unwrap_or(Precision::Absent),
-                min_value: min_value.map(Precision::Exact).unwrap_or(Precision::Absent),
+                max_value: max_value.unwrap_or(Precision::Absent),
+                min_value: min_value.unwrap_or(Precision::Absent),
                 sum_value: Precision::Absent,
                 distinct_count: Precision::Absent,
             }
@@ -1290,6 +1316,8 @@ fn summarize_min_max_null_counts(
     min_accs: &mut [Option<MinAccumulator>],
     max_accs: &mut [Option<MaxAccumulator>],
     null_counts_array: &mut [Precision<usize>],
+    is_min_value_exact: &mut [Option<bool>],
+    is_max_value_exact: &mut [Option<bool>],
     arrow_schema_index: usize,
     num_rows: usize,
     stats_converter: &StatisticsConverter,
@@ -1298,13 +1326,29 @@ fn summarize_min_max_null_counts(
     let max_values = stats_converter.row_group_maxes(row_groups_metadata)?;
     let min_values = stats_converter.row_group_mins(row_groups_metadata)?;
     let null_counts = stats_converter.row_group_null_counts(row_groups_metadata)?;
+    let is_max_value_exact_stat =
+        stats_converter.row_group_is_max_value_exact(row_groups_metadata)?;
+    let is_min_value_exact_stat =
+        stats_converter.row_group_is_min_value_exact(row_groups_metadata)?;
 
     if let Some(max_acc) = &mut max_accs[arrow_schema_index] {
-        max_acc.update_batch(&[max_values])?;
+        max_acc.update_batch(&[Arc::clone(&max_values)])?;
+        let mut cur_max_acc = max_acc.clone();
+        is_max_value_exact[arrow_schema_index] = has_any_exact_match(
+            cur_max_acc.evaluate()?,
+            max_values,
+            is_max_value_exact_stat,
+        );
     }
 
     if let Some(min_acc) = &mut min_accs[arrow_schema_index] {
-        min_acc.update_batch(&[min_values])?;
+        min_acc.update_batch(&[Arc::clone(&min_values)])?;
+        let mut cur_min_acc = min_acc.clone();
+        is_min_value_exact[arrow_schema_index] = has_any_exact_match(
+            cur_min_acc.evaluate()?,
+            min_values,
+            is_min_value_exact_stat,
+        );
     }
 
     null_counts_array[arrow_schema_index] = Precision::Exact(match sum(&null_counts) {
@@ -1967,6 +2011,33 @@ fn create_max_min_accs(
     (max_values, min_values)
 }
 
+/// Checks if any occurrence of `value` in `array` corresponds to a `true`
+/// entry in the `exactness` array.
+///
+/// This is used to determine if a calculated statistic (e.g., min or max)
+/// is exact, by checking if at least one of its source values was exact.
+///
+/// # Example
+/// - `value`: `0`
+/// - `array`: `[0, 1, 0, 3, 0, 5]`
+/// - `exactness`: `[true, false, false, false, false, false]`
+///
+/// The value `0` appears at indices `[0, 2, 4]`. The corresponding exactness
+/// values are `[true, false, false]`. Since at least one is `true`, the
+/// function returns `Some(true)`.
+fn has_any_exact_match(
+    value: ScalarValue,
+    array: ArrayRef,
+    exactness: BooleanArray,
+) -> Option<bool> {
+    value.to_scalar().ok().and_then(|value| {
+        eq(&value, &array)
+            .ok()
+            .and_then(|mask| and(&mask, &exactness).ok())
+            .and_then(|result| Some(result.true_count() > 0))
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;

datafusion/functions-aggregate/src/min_max.rs

@@ -736,7 +736,7 @@ macro_rules! min_max {
 }
 
 /// An accumulator to compute the maximum value
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct MaxAccumulator {
     max: ScalarValue,
 }
@@ -1057,7 +1057,7 @@ impl AggregateUDFImpl for Min {
 }
 
 /// An accumulator to compute the minimum value
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct MinAccumulator {
     min: ScalarValue,
 }