feat(storage): adapt string stats length for common prefixes (#20199)

fix(storage): adapt string stats prefix length

Author: dantengsky

src/query/storages/fuse/src/io/write/stream/column_statistics_builder.rs

@@ -47,10 +47,7 @@ use databend_common_expression::with_number_type;
 use databend_storages_common_table_meta::meta::ColumnStatistics;
 use enum_dispatch::enum_dispatch;
 
-use crate::statistics::STATS_STRING_PREFIX_LEN;
-use crate::statistics::Trim;
-use crate::statistics::trim_string_max_with_len;
-use crate::statistics::trim_string_min_with_len;
+use crate::statistics::trim_column_min_max;
 
 pub type CommonBuilder<T> = GenericColumnStatisticsBuilder<T, CommonAdapter>;
 pub type DecimalBuilder<T> = GenericColumnStatisticsBuilder<T, DecimalAdapter>;
@@ -105,7 +102,7 @@ where
 
 pub fn create_column_stats_builder(
     data_type: &DataType,
-    string_col_len: usize,
+    string_col_len: Option<usize>,
 ) -> ColumnStatisticsBuilder {
     let inner_type = data_type.remove_nullable();
     macro_rules! match_number_type_create {
@@ -226,7 +223,7 @@ where
     null_count: usize,
     in_memory_size: usize,
     data_type: DataType,
-    string_col_len: usize,
+    string_col_len: Option<usize>,
 
     _phantom: PhantomData<(T, A)>,
 }
@@ -245,12 +242,12 @@ where
             null_count: 0,
             in_memory_size: 0,
             data_type,
-            string_col_len: STATS_STRING_PREFIX_LEN,
+            string_col_len: None,
             _phantom: PhantomData,
         }
     }
 
-    fn create_with_string_len(data_type: DataType, string_col_len: usize) -> Self {
+    fn create_with_string_len(data_type: DataType, string_col_len: Option<usize>) -> Self {
         Self {
             min: None,
             max: None,
@@ -351,47 +348,25 @@ where
     }
 
     fn finalize(self) -> Result<ColumnStatistics> {
-        let string_col_len = self.string_col_len;
-        let min = if let Some(v) = self.min {
+        let raw_min = if let Some(v) = self.min {
             let v = A::value_to_scalar(v);
-            let scalar = T::upcast_scalar_with_type(v, &self.data_type);
-            match scalar {
-                Scalar::String(s) => trim_string_min_with_len(s, string_col_len)
-                    .map(Scalar::String)
-                    .unwrap_or(Scalar::Null),
-                other => other.trim_min().unwrap_or(Scalar::Null),
-            }
+            T::upcast_scalar_with_type(v, &self.data_type)
         } else {
             Scalar::Null
         };
-        let max = if let Some(v) = self.max {
+        let raw_max = if let Some(v) = self.max {
             let v = A::value_to_scalar(v);
-            let scalar = T::upcast_scalar_with_type(v, &self.data_type);
-            match scalar {
-                Scalar::String(s) => {
-                    if let Some(v) = trim_string_max_with_len(s, string_col_len) {
-                        Scalar::String(v)
-                    } else {
-                        return Err(ErrorCode::Internal(
-                            "Unable to trim string: first chars are all replacement_point"
-                                .to_string(),
-                        ));
-                    }
-                }
-                other => {
-                    if let Some(v) = other.trim_max() {
-                        v
-                    } else {
-                        return Err(ErrorCode::Internal(
-                            "Unable to trim string: first 16 chars are all replacement_point"
-                                .to_string(),
-                        ));
-                    }
-                }
-            }
+            T::upcast_scalar_with_type(v, &self.data_type)
         } else {
             Scalar::Null
         };
+        let (min, max) =
+            trim_column_min_max(raw_min, raw_max, self.string_col_len).ok_or_else(|| {
+                ErrorCode::Internal(
+                    "Unable to trim string: retained prefix is at the end of the Unicode range"
+                        .to_string(),
+                )
+            })?;
 
         Ok(ColumnStatistics::new(
             min,

src/query/storages/fuse/src/io/write/stream/column_statistics_state.rs

@@ -45,10 +45,7 @@ impl ColumnStatisticsState {
         let col_stats = stats_columns
             .iter()
             .map(|(col_id, data_type)| {
-                let string_len = col_stats_truncate_lens
-                    .get(col_id)
-                    .copied()
-                    .unwrap_or(crate::statistics::STATS_STRING_PREFIX_LEN);
+                let string_len = col_stats_truncate_lens.get(col_id).copied();
                 (*col_id, create_column_stats_builder(data_type, string_len))
             })
             .collect();
@@ -146,6 +143,7 @@ mod tests {
     use databend_storages_common_index::RangeIndex;
 
     use super::*;
+    use crate::statistics::END_OF_UNICODE_RANGE;
     use crate::statistics::gen_columns_statistics;
 
     #[test]
@@ -198,4 +196,33 @@ mod tests {
         assert_eq!(stats_0, stats_1);
         Ok(())
     }
+
+    #[test]
+    fn test_column_stats_state_adaptive_string_prefix() -> Result<()> {
+        let field = TableField::new("a", TableDataType::String);
+        let schema = Arc::new(TableSchema::new(vec![field]));
+        let prefix = "abcdefghijklmnop";
+        let min = format!("{prefix}a-min-suffix");
+        let max = format!("{prefix}z-max-suffix");
+        let block = DataBlock::new_from_columns(vec![StringType::from_data(vec![
+            min.as_str(),
+            max.as_str(),
+        ])]);
+        let stats_columns = vec![(0, DataType::String)];
+
+        let mut state = ColumnStatisticsState::new(&stats_columns, &[], &BTreeMap::new());
+        state.add_block(&schema, &block)?;
+        let stats = state.finalize(HashMap::new())?;
+        let col_stats = stats.get(&0).unwrap();
+
+        assert_eq!(
+            col_stats.min(),
+            &databend_common_expression::Scalar::String(format!("{prefix}a"))
+        );
+        assert_eq!(
+            col_stats.max(),
+            &databend_common_expression::Scalar::String(format!("{prefix}{END_OF_UNICODE_RANGE}"))
+        );
+        Ok(())
+    }
 }

src/query/storages/fuse/src/statistics/column_statistic.rs

@@ -68,11 +68,19 @@ pub fn gen_columns_statistics(
                 let (distinct_of_values, unset_bits) =
                     if s == Scalar::Null { (0, rows) } else { (1, 0) };
 
+                let Some((min, max)) = trim_column_min_max(
+                    s.clone(),
+                    s.clone(),
+                    col_stats_truncate_lens.get(&column_id).copied(),
+                ) else {
+                    continue;
+                };
+
                 // when we read it back from parquet, it is a Column instead of Scalar
                 let in_memory_size = s.as_ref().estimated_scalar_repeat_size(rows, &data_type);
                 let col_stats = ColumnStatistics::new(
-                    s.clone(),
-                    s.clone(),
+                    min,
+                    max,
                     unset_bits as u64,
                     in_memory_size as u64,
                     Some(distinct_of_values),
@@ -82,56 +90,27 @@ pub fn gen_columns_statistics(
             }
             Value::Column(col) => {
                 // later, during the evaluation of expressions, name of field does not matter
-                let mut min = Scalar::Null;
-                let mut max = Scalar::Null;
-
-                if col.len() > 0 {
+                let (min, max) = if col.len() > 0 {
                     let (mins, _) = eval_aggr("min", vec![], &[col.clone().into()], rows, vec![])?;
                     let (maxs, _) = eval_aggr("max", vec![], &[col.clone().into()], rows, vec![])?;
 
-                    let truncate_len = col_stats_truncate_lens
-                        .get(&column_id)
-                        .copied()
-                        .unwrap_or(STATS_STRING_PREFIX_LEN);
-
-                    if mins.len() > 0 {
-                        min = if let Some(v) = mins.index(0) {
-                            let owned = v.to_owned();
-                            let trimmed = match owned {
-                                Scalar::String(s) => {
-                                    trim_string_min_with_len(s, truncate_len).map(Scalar::String)
-                                }
-                                other => other.trim_min(),
-                            };
-                            if let Some(v) = trimmed {
-                                v
-                            } else {
-                                continue;
-                            }
-                        } else {
-                            continue;
-                        }
-                    }
-
-                    if maxs.len() > 0 {
-                        max = if let Some(v) = maxs.index(0) {
-                            let owned = v.to_owned();
-                            let trimmed = match owned {
-                                Scalar::String(s) => {
-                                    trim_string_max_with_len(s, truncate_len).map(Scalar::String)
-                                }
-                                other => other.trim_max(),
-                            };
-                            if let Some(v) = trimmed {
-                                v
-                            } else {
-                                continue;
-                            }
-                        } else {
-                            continue;
-                        }
-                    }
-                }
+                    let Some(raw_min) = mins.index(0).map(|v| v.to_owned()) else {
+                        continue;
+                    };
+                    let Some(raw_max) = maxs.index(0).map(|v| v.to_owned()) else {
+                        continue;
+                    };
+                    let Some(min_max) = trim_column_min_max(
+                        raw_min,
+                        raw_max,
+                        col_stats_truncate_lens.get(&column_id).copied(),
+                    ) else {
+                        continue;
+                    };
+                    min_max
+                } else {
+                    (Scalar::Null, Scalar::Null)
+                };
 
                 let (is_all_null, bitmap) = col.validity();
                 let unset_bits = match (is_all_null, bitmap) {
@@ -190,6 +169,43 @@ pub trait Trim: Sized {
 
 pub const END_OF_UNICODE_RANGE: char = '\u{10FFFF}';
 pub const STATS_STRING_PREFIX_LEN: usize = 16;
+const MAX_AUTO_STATS_STRING_PREFIX_LEN: usize = 32;
+const MAX_AUTO_STATS_STRING_COMMON_PREFIX_LEN: usize = MAX_AUTO_STATS_STRING_PREFIX_LEN - 1;
+
+fn auto_stats_string_prefix_len(min: &str, max: &str) -> usize {
+    let common_prefix_len = min
+        .chars()
+        .zip(max.chars())
+        .take(MAX_AUTO_STATS_STRING_COMMON_PREFIX_LEN)
+        .take_while(|(min_char, max_char)| min_char == max_char)
+        .count();
+
+    if common_prefix_len < STATS_STRING_PREFIX_LEN {
+        STATS_STRING_PREFIX_LEN
+    } else {
+        common_prefix_len
+            .saturating_add(1)
+            .min(MAX_AUTO_STATS_STRING_PREFIX_LEN)
+    }
+}
+
+pub(crate) fn trim_column_min_max(
+    min: Scalar,
+    max: Scalar,
+    string_truncate_len: Option<usize>,
+) -> Option<(Scalar, Scalar)> {
+    match (min, max) {
+        (Scalar::String(min), Scalar::String(max)) => {
+            let truncate_len =
+                string_truncate_len.unwrap_or_else(|| auto_stats_string_prefix_len(&min, &max));
+            Some((
+                Scalar::String(trim_string_min_with_len(min, truncate_len)?),
+                Scalar::String(trim_string_max_with_len(max, truncate_len)?),
+            ))
+        }
+        (min, max) => Some((min.trim_min()?, max.trim_max()?)),
+    }
+}
 
 impl Trim for Scalar {
     fn trim_min(self) -> Option<Self> {
@@ -317,10 +333,48 @@ pub fn trim_string_max_with_len(s: String, len: usize) -> Option<String> {
 mod tests {
     use databend_common_expression::Scalar;
 
+    use super::MAX_AUTO_STATS_STRING_COMMON_PREFIX_LEN;
+    use super::MAX_AUTO_STATS_STRING_PREFIX_LEN;
+    use super::auto_stats_string_prefix_len;
     use crate::statistics::END_OF_UNICODE_RANGE;
     use crate::statistics::STATS_STRING_PREFIX_LEN;
     use crate::statistics::Trim;
 
+    #[test]
+    fn test_auto_stats_string_prefix_len() {
+        assert_eq!(
+            auto_stats_string_prefix_len("abcdefghijklmnoa", "abcdefghijklmnoz"),
+            STATS_STRING_PREFIX_LEN
+        );
+
+        let prefix = "abcdefghijklmnop";
+        assert_eq!(
+            auto_stats_string_prefix_len(
+                &format!("{prefix}a-min-suffix"),
+                &format!("{prefix}z-max-suffix"),
+            ),
+            STATS_STRING_PREFIX_LEN + 1
+        );
+
+        let long_prefix = "a".repeat(MAX_AUTO_STATS_STRING_COMMON_PREFIX_LEN);
+        assert_eq!(
+            auto_stats_string_prefix_len(
+                &format!("{long_prefix}x-min"),
+                &format!("{long_prefix}y-max"),
+            ),
+            MAX_AUTO_STATS_STRING_PREFIX_LEN
+        );
+
+        let capped_prefix = "好".repeat(MAX_AUTO_STATS_STRING_PREFIX_LEN);
+        assert_eq!(
+            auto_stats_string_prefix_len(
+                &format!("{capped_prefix}甲"),
+                &format!("{capped_prefix}乙"),
+            ),
+            MAX_AUTO_STATS_STRING_PREFIX_LEN
+        );
+    }
+
     #[test]
     fn test_trim_max() {
         {

src/query/storages/fuse/src/statistics/mod.rs

@@ -41,6 +41,7 @@ pub use column_statistic::Trim;
 pub use column_statistic::calc_column_distinct_of_values;
 pub use column_statistic::gen_columns_statistics;
 pub use column_statistic::scalar_min_max;
+pub(crate) use column_statistic::trim_column_min_max;
 pub use column_statistic::trim_string_max_with_len;
 pub use column_statistic::trim_string_min_with_len;
 pub use reducers::merge_statistics;