Fix bugs.

- Parquet schema mismatched. Databend don't support dictionary types, so
don't parse schema from meta key value data.
- Output schema should be prune by prewhere info.

Author: RinChanNOWWW

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

@@ -18,6 +18,7 @@ use common_arrow::arrow::io::parquet::read as pread;
 use common_arrow::parquet::metadata::ColumnChunkMetaData;
 use common_arrow::parquet::metadata::FileMetaData;
 use common_arrow::parquet::metadata::RowGroupMetaData;
+use common_datavalues::DataField;
 use common_datavalues::DataSchema;
 use common_exception::ErrorCode;
 use common_exception::Result;
@@ -39,12 +40,19 @@ impl ParquetReader {
 
     #[inline]
     pub fn infer_schema(meta: &FileMetaData) -> Result<DataSchema> {
-        let mut arrow_schema = pread::infer_schema(meta)?;
-        // Need to change all the field name to lowercase.
-        for field in &mut arrow_schema.fields {
-            field.name = field.name.to_lowercase();
-        }
-        Ok(DataSchema::from(arrow_schema))
+        // Do not use `pread::infer_schema(meta)` becuase it will use metadata `ARROW:schema`.
+        // There maybe dictionary types in the schema, which is not supported by Databend.
+        // So we need to convert the primitive schema directly.
+        let field = pread::schema::parquet_to_arrow_schema(meta.schema().fields())
+            .into_iter()
+            .map(|mut f| {
+                // Need to change all the field name to lowercase.
+                f.name = f.name.to_lowercase();
+                DataField::from(&f)
+            })
+            .collect::<Vec<_>>();
+
+        Ok(DataSchema::new(field))
     }
 
     pub fn get_column_metas<'a>(

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

@@ -81,11 +81,7 @@ impl ParquetTable {
         )
     }
 
-    fn adjust_io_request(
-        &self,
-        ctx: &Arc<dyn TableContext>,
-        projection: &Projection,
-    ) -> Result<usize> {
+    fn adjust_io_request(&self, ctx: &Arc<dyn TableContext>, num_columns: usize) -> Result<usize> {
         let conf = GlobalConfig::instance();
         let mut max_memory_usage = ctx.get_settings().get_max_memory_usage()? as usize;
         if conf.query.table_cache_enabled {
@@ -97,7 +93,7 @@ impl ParquetTable {
         let block_file_size = 300 * 1024 * 1024_usize;
         let table_column_len = self.table_info.schema().fields().len();
         let per_column_bytes = block_file_size / table_column_len;
-        let scan_column_bytes = per_column_bytes * projection.len();
+        let scan_column_bytes = per_column_bytes * num_columns;
         let estimate_io_requests = max_memory_usage / scan_column_bytes;
 
         let setting_io_requests = std::cmp::max(
@@ -115,9 +111,20 @@ impl ParquetTable {
         plan: &DataSourcePlan,
         pipeline: &mut Pipeline,
     ) -> Result<()> {
-        let projection = PushDownInfo::projection_of_push_downs(&plan.schema(), &plan.push_downs);
-        let output_schema = Arc::new(projection.project_schema(&plan.source_info.schema()));
-        let max_io_requests = self.adjust_io_request(&ctx, &projection)?;
+        let columns_to_read =
+            PushDownInfo::projection_of_push_downs(&plan.schema(), &plan.push_downs);
+        let max_io_requests = self.adjust_io_request(&ctx, columns_to_read.len())?;
+
+        // If there is a `PrewhereInfo`, the final output should be `PrehwereInfo.output_columns`.
+        // `PrewhereInfo.output_columns` should be a subset of `PushDownInfo.projection`.
+        let output_projection = match PushDownInfo::prewhere_of_push_downs(&plan.push_downs) {
+            None => {
+                PushDownInfo::projection_of_push_downs(&self.table_info.schema(), &plan.push_downs)
+            }
+            Some(v) => v.output_columns.clone(),
+        };
+        let output_schema = Arc::new(output_projection.project_schema(&plan.source_info.schema()));
+
         let prewhere_reader = self.build_prewhere_reader(plan)?;
         let prewhere_filter =
             self.build_prewhere_filter_executor(ctx.clone(), plan, prewhere_reader.schema())?;