Fix bugs.

Author: RinChanNOWWW

src/query/storages/parquet/src/parquet_reader/meta.rs

@@ -65,6 +65,15 @@ impl ParquetReader {
         let mut stats_of_row_groups = HashMap::with_capacity(rgs.len());
 
         for index in indices {
+            if rgs
+                .iter()
+                .any(|rg| rg.columns()[*index].metadata().statistics.is_none())
+            {
+                return Err(ErrorCode::InvalidArgument(
+                    "Some columns of the row groups have no statistics",
+                ));
+            }
+
             let field = &schema.fields[*index];
             let column_stats = pread::statistics::deserialize(field, rgs)?;
             stats_of_row_groups.insert(*index, BatchStatistics::from(column_stats));
@@ -97,8 +106,8 @@ pub struct BatchStatistics {
 impl BatchStatistics {
     pub fn get(&self, index: usize) -> ColumnStatistics {
         ColumnStatistics {
-            min: self.min_values.get(0),
-            max: self.max_values.get(0),
+            min: self.min_values.get(index),
+            max: self.max_values.get(index),
             null_count: self.null_count.get_u64(index).unwrap(),
             in_memory_size: 0, // this field is not used.
             distinct_of_values: self.distinct_count.get_u64(index).ok(),
@@ -111,7 +120,7 @@ impl From<pread::statistics::Statistics> for BatchStatistics {
         let null_count = UInt64Column::from_arrow_array(&*stats.null_count);
         let distinct_count = UInt64Column::from_arrow_array(&*stats.distinct_count);
         let min_values = stats.min_value.clone().into_column();
-        let max_values = stats.min_value.clone().into_column();
+        let max_values = stats.max_value.clone().into_column();
         Self {
             null_count,
             distinct_count,

src/query/storages/parquet/src/table_function/read.rs

@@ -139,7 +139,7 @@ impl ParquetTable {
 
         // do parition at the begin of the whole pipeline.
         let push_downs = plan.push_downs.clone();
-        let schema = plan.schema();
+        let schema = plan.source_info.schema();
         pipeline.set_on_init(move || {
             let mut partitions = Vec::with_capacity(locations.len());
 

src/query/storages/pruner/src/range_pruner.rs

@@ -55,6 +55,9 @@ impl RangePruner for RangeFilter {
     }
 }
 
+/// Create a new [`RangePruner`] from expression and schema.
+///
+/// Note: the schema should be the schema of the table, not the schema of the input.
 pub fn new_range_pruner<'a>(
     ctx: &Arc<dyn TableContext>,
     filter_expr: Option<&'a [Expression]>,