feat(hive): add timestamp type for hive

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

@@ -29,6 +29,7 @@ use common_datavalues::DataSchema;
 use common_datavalues::DataTypeImpl;
 use common_datavalues::DateType;
 use common_datavalues::NullableType;
+use common_datavalues::TimestampType;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_hive_meta_store as hms;
@@ -171,7 +172,7 @@ fn try_from_filed_type_name(type_name: impl AsRef<str>) -> Result<DataTypeImpl>
             "DOUBLE PRECISION" => Ok(DataTypeImpl::Float64(Float64Type::default())),
 
             // timestamp
-            // "TIMESTAMP" => Ok(DataTypeImpl::Timestamp(TimestampType::create(3))),
+            "TIMESTAMP" => Ok(DataTypeImpl::Timestamp(TimestampType::create(3))),
             "DATE" => Ok(DataTypeImpl::Date(DateType::default())),
 
             _ => Err(ErrorCode::IllegalDataType(format!(

src/query/storages/hive/src/hive_parquet_block_reader.rs

@@ -20,12 +20,9 @@ use common_arrow::arrow::io::parquet::read::column_iter_to_arrays;
 use common_arrow::arrow::io::parquet::read::read_metadata_async;
 use common_arrow::arrow::io::parquet::read::ArrayIter;
 use common_arrow::arrow::io::parquet::read::RowGroupDeserializer;
-use common_arrow::arrow::io::parquet::write::to_parquet_schema;
 use common_arrow::parquet::metadata::ColumnChunkMetaData;
-use common_arrow::parquet::metadata::ColumnDescriptor;
 use common_arrow::parquet::metadata::FileMetaData;
 use common_arrow::parquet::metadata::RowGroupMetaData;
-use common_arrow::parquet::metadata::SchemaDescriptor;
 use common_arrow::parquet::read::BasicDecompressor;
 use common_arrow::parquet::read::PageReader;
 use common_base::base::tokio::sync::Semaphore;
@@ -50,7 +47,6 @@ pub struct HiveParquetBlockReader {
     projection: Vec<usize>,
     arrow_schema: Arc<Schema>,
     projected_schema: DataSchemaRef,
-    parquet_schema_descriptor: SchemaDescriptor,
     hive_partition_filler: Option<HivePartitionFiller>,
 }
 
@@ -64,13 +60,11 @@ impl HiveParquetBlockReader {
         let projected_schema = DataSchemaRef::new(schema.project(&projection));
 
         let arrow_schema = schema.to_arrow();
-        let parquet_schema_descriptor = to_parquet_schema(&arrow_schema)?;
 
         Ok(Arc::new(HiveParquetBlockReader {
             operator,
             projection,
             projected_schema,
-            parquet_schema_descriptor,
             arrow_schema: Arc::new(arrow_schema),
             hive_partition_filler,
         }))
@@ -80,21 +74,20 @@ impl HiveParquetBlockReader {
         column_meta: &ColumnChunkMetaData,
         chunk: Vec<u8>,
         rows: usize,
-        column_descriptor: &ColumnDescriptor,
         field: Field,
     ) -> Result<ArrayIter<'static>> {
+        let primitive_type = column_meta.descriptor().descriptor.primitive_type.clone();
         let pages = PageReader::new(
             std::io::Cursor::new(chunk),
             column_meta,
             Arc::new(|_, _| true),
             vec![],
         );
 
-        let primitive_type = &column_descriptor.descriptor.primitive_type;
         let decompressor = BasicDecompressor::new(pages, vec![]);
         Ok(column_iter_to_arrays(
             vec![decompressor],
-            vec![primitive_type],
+            vec![&primitive_type],
             field,
             Some(rows),
         )?)
@@ -220,14 +213,12 @@ impl HiveParquetBlockReader {
         for (index, column_chunk) in chunks.into_iter().enumerate() {
             let idx = self.projection[index];
             let field = self.arrow_schema.fields[idx].clone();
-            let column_descriptor = &self.parquet_schema_descriptor.columns()[idx];
             let column_meta = Self::get_parquet_column_metadata(row_group, &field.name)?;
 
             columns_array_iter.push(Self::to_deserialize(
                 column_meta,
                 column_chunk,
                 row_group.num_rows(),
-                column_descriptor,
                 field,
             )?);
         }