Merge branch 'main' into fix-agg2

Author: mergify[bot]

src/query/storages/hive/hive/src/converters.rs

@@ -80,7 +80,7 @@ pub fn try_into_table_info(
     } else {
         None
     };
-    let schema = Arc::new(try_into_schema(fields, partition_keys.clone())?);
+    let schema = Arc::new(try_into_schema(fields)?);
 
     let location = if let Some(storage) = &hms_table.sd {
         storage
@@ -125,19 +125,12 @@ pub fn try_into_table_info(
     Ok(table_info)
 }
 
-fn try_into_schema(
-    hive_fields: Vec<hms::FieldSchema>,
-    partition_keys: Option<Vec<String>>,
-) -> Result<DataSchema> {
+fn try_into_schema(hive_fields: Vec<hms::FieldSchema>) -> Result<DataSchema> {
     let mut fields = Vec::new();
-    let partition_keys = partition_keys.unwrap_or_default();
     for field in hive_fields {
         let name = field.name.unwrap_or_default();
         let type_name = field.type_.unwrap_or_default();
-        let data_type = match partition_keys.contains(&name) {
-            true => try_from_filed_type_name(type_name)?,
-            false => NullableType::new_impl(try_from_filed_type_name(type_name)?),
-        };
+        let data_type = NullableType::new_impl(try_from_filed_type_name(type_name)?);
         let field = DataField::new(&name, data_type);
         fields.push(field);
     }

src/query/storages/hive/hive/src/hive_block_filter.rs

@@ -29,6 +29,7 @@ use common_storages_table_meta::meta::ColumnStatistics;
 use common_storages_table_meta::meta::StatisticsOfColumns;
 
 use crate::hive_parquet_block_reader::HiveParquetBlockReader;
+use crate::hive_table::HIVE_DEFAULT_PARTITION;
 
 #[derive(Clone)]
 pub struct HiveBlockFilter {
@@ -89,11 +90,18 @@ impl HiveBlockFilter {
 
             for (p_key, p_value) in part_columns {
                 if let Ok(idx) = self.data_schema.index_of(&p_key) {
-                    let v = DataValue::String(p_value.as_bytes().to_vec());
+                    let mut null_count = 0;
+                    let v = if p_value == HIVE_DEFAULT_PARTITION {
+                        null_count = row_group.num_rows();
+                        DataValue::Null
+                    } else {
+                        DataValue::String(p_value.as_bytes().to_vec())
+                    };
+
                     let col_stats = ColumnStatistics {
                         min: v.clone(),
                         max: v,
-                        null_count: 0,
+                        null_count: null_count as u64,
                         in_memory_size: 0,
                         distinct_of_values: None,
                     };

src/query/storages/hive/hive/src/hive_partition_filler.rs

@@ -14,17 +14,20 @@
 
 use std::sync::Arc;
 
+use common_arrow::arrow::bitmap::Bitmap;
 use common_datablocks::DataBlock;
 use common_datavalues::ColumnRef;
 use common_datavalues::ConstColumn;
 use common_datavalues::DataField;
 use common_datavalues::DataTypeImpl;
+use common_datavalues::NullableColumn;
 use common_datavalues::Series;
 use common_datavalues::SeriesFrom;
 use common_exception::ErrorCode;
 use common_exception::Result;
 
 use crate::hive_partition::HivePartInfo;
+use crate::hive_table::HIVE_DEFAULT_PARTITION;
 
 #[derive(Debug, Clone)]
 pub struct HivePartitionFiller {
@@ -39,8 +42,17 @@ macro_rules! generate_primitive_column {
 }
 
 fn generate_string_column(num_rows: usize, value: String) -> Result<ColumnRef> {
+    let validity = if value == HIVE_DEFAULT_PARTITION {
+        Some(Bitmap::from(vec![false]))
+    } else {
+        None
+    };
     let column = Series::from_data(vec![value]);
-    Ok(Arc::new(ConstColumn::new(column, num_rows)))
+
+    Ok(Arc::new(ConstColumn::new(
+        NullableColumn::wrap_inner(column, validity),
+        num_rows,
+    )))
 }
 
 impl HivePartitionFiller {
@@ -54,7 +66,11 @@ impl HivePartitionFiller {
         value: String,
         field: &DataField,
     ) -> Result<ColumnRef> {
-        match field.data_type().clone() {
+        let t = match field.data_type() {
+            DataTypeImpl::Nullable(v) => v.inner_type(),
+            _ => field.data_type(),
+        };
+        match t {
             DataTypeImpl::String(_) => generate_string_column(num_rows, value),
             DataTypeImpl::Int8(_) => generate_primitive_column!(i8, num_rows, value),
             DataTypeImpl::Int16(_) => generate_primitive_column!(i16, num_rows, value),

src/query/storages/hive/hive/src/hive_partition_pruner.rs

@@ -27,6 +27,8 @@ use common_storages_index::range_filter::RangeFilter;
 use common_storages_table_meta::meta::ColumnStatistics;
 use common_storages_table_meta::meta::StatisticsOfColumns;
 
+use crate::hive_table::HIVE_DEFAULT_PARTITION;
+
 pub struct HivePartitionPruner {
     pub ctx: Arc<dyn TableContext>,
     pub filters: Vec<Expression>,
@@ -54,9 +56,21 @@ impl HivePartitionPruner {
             for (index, singe_value) in partition.split('/').enumerate() {
                 let kv = singe_value.split('=').collect::<Vec<&str>>();
                 let field = self.partition_schema.fields()[index].clone();
+                let t = match field.data_type() {
+                    DataTypeImpl::Nullable(v) => v.inner_type(),
+                    _ => field.data_type(),
+                };
 
-                let v = match field.data_type() {
-                    DataTypeImpl::String(_) => DataValue::String(kv[1].as_bytes().to_vec()),
+                let mut null_count = 0;
+                let v = match t {
+                    DataTypeImpl::String(_) => {
+                        if kv[1] == HIVE_DEFAULT_PARTITION {
+                            null_count = 1;
+                            DataValue::Null
+                        } else {
+                            DataValue::String(kv[1].as_bytes().to_vec())
+                        }
+                    }
                     DataTypeImpl::Int8(_)
                     | DataTypeImpl::Int16(_)
                     | DataTypeImpl::Int32(_)
@@ -79,7 +93,7 @@ impl HivePartitionPruner {
                 let column_stats = ColumnStatistics {
                     min: v.clone(),
                     max: v,
-                    null_count: 0,
+                    null_count,
                     in_memory_size: 0,
                     distinct_of_values: None,
                 };

src/query/storages/hive/hive/src/hive_table.rs

@@ -62,6 +62,7 @@ use crate::HiveBlockFilter;
 use crate::HiveFileSplitter;
 
 pub const HIVE_TABLE_ENGIE: &str = "hive";
+pub const HIVE_DEFAULT_PARTITION: &str = "__HIVE_DEFAULT_PARTITION__";
 
 pub struct HiveTable {
     table_info: TableInfo,