fix: Schema mismatch issue that may occur due to case insensitivity when loading StageTable (#17949)

* fix: Schema mismatch issue that may occur due to case insensitivity when loading StageTable

* chore: codefmt

* chore: `RecordBatchTransformer` skips StableTable and has Tuple type

* chore: add var `parquet_fast_read_bytes` on `case_sensitive_parquet.test`

* chore: `RecordBatchTransformer` skip `StageTable`

Author: KKould

src/query/storages/delta/src/table.rs

@@ -46,6 +46,7 @@ use databend_common_storages_parquet::ParquetFilePart;
 use databend_common_storages_parquet::ParquetPart;
 use databend_common_storages_parquet::ParquetPruner;
 use databend_common_storages_parquet::ParquetReaderBuilder;
+use databend_common_storages_parquet::ParquetSourceType;
 use databend_storages_common_pruner::partition_prunner::FetchPartitionScalars;
 use databend_storages_common_pruner::partition_prunner::PartitionPruner;
 use databend_storages_common_table_meta::table::OPT_KEY_ENGINE_META;
@@ -254,7 +255,8 @@ impl DeltaTable {
                 .with_pruner(Some(pruner))
                 .with_partition_columns(self.meta.partition_columns.clone());
 
-        let parquet_reader = Arc::new(builder.build_full_reader(false)?);
+        let parquet_reader =
+            Arc::new(builder.build_full_reader(ParquetSourceType::DeltaLake, false)?);
 
         let output_schema = Arc::new(DataSchema::from(plan.schema()));
         pipeline.add_source(

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

@@ -53,6 +53,7 @@ use databend_common_storage::init_operator;
 use databend_common_storage::DataOperator;
 use databend_common_storages_parquet::ParquetPruner;
 use databend_common_storages_parquet::ParquetReaderBuilder;
+use databend_common_storages_parquet::ParquetSourceType;
 use databend_storages_common_pruner::partition_prunner::PartitionPruner;
 use databend_storages_common_table_meta::meta::SnapshotId;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
@@ -196,7 +197,7 @@ impl HiveTable {
                 .with_pruner(Some(pruner))
                 .with_partition_columns(partition_keys);
 
-        let parquet_reader = Arc::new(builder.build_full_reader(false)?);
+        let parquet_reader = Arc::new(builder.build_full_reader(ParquetSourceType::Hive, false)?);
 
         let output_schema = Arc::new(DataSchema::from(plan.schema()));
         pipeline.add_source(

src/query/storages/iceberg/src/table.rs

@@ -359,12 +359,11 @@ impl IcebergTable {
                 builder = builder.with_topk(topk.as_ref());
             }
 
-            let row_group_reader = Arc::new(builder.build_row_group_reader(need_row_number)?);
+            let row_group_reader = Arc::new(
+                builder.build_row_group_reader(ParquetSourceType::Iceberg, need_row_number)?,
+            );
 
             let topk = Arc::new(topk);
-            let projection =
-                PushDownInfo::projection_of_push_downs(&table_schema, plan.push_downs.as_ref());
-            let output_schema = Arc::new(projection.project_schema(&table_schema));
 
             pipeline.add_source(
                 |output| {
@@ -378,7 +377,6 @@ impl IcebergTable {
                         internal_columns.clone(),
                         plan.push_downs.clone(),
                         table_schema.clone(),
-                        output_schema.clone(),
                         op.clone(),
                     )
                 },

src/query/storages/parquet/src/parquet_reader/reader/builder.rs

@@ -44,7 +44,9 @@ use crate::parquet_reader::NoPretchPolicyBuilder;
 use crate::parquet_reader::ParquetWholeFileReader;
 use crate::parquet_reader::PredicateAndTopkPolicyBuilder;
 use crate::parquet_reader::TopkOnlyPolicyBuilder;
+use crate::transformer::RecordBatchTransformer;
 use crate::ParquetPruner;
+use crate::ParquetSourceType;
 
 pub struct ParquetReaderBuilder<'a> {
     ctx: Arc<dyn TableContext>,
@@ -214,6 +216,7 @@ impl<'a> ParquetReaderBuilder<'a> {
 
     pub fn build_full_reader(
         &mut self,
+        source_type: ParquetSourceType,
         need_file_row_number: bool,
     ) -> Result<ParquetWholeFileReader> {
         let batch_size = self.ctx.get_settings().get_parquet_max_block_size()? as usize;
@@ -231,6 +234,8 @@ impl<'a> ParquetReaderBuilder<'a> {
             .unwrap();
 
         let (_, _, output_schema, _) = self.built_output.as_ref().unwrap();
+        let transformer = (!matches!(source_type, ParquetSourceType::StageTable))
+            .then(|| RecordBatchTransformer::build(output_schema.clone()));
         Ok(ParquetWholeFileReader {
             op_registry: self.op_registry.clone(),
             expect_file_schema: self
@@ -242,12 +247,17 @@ impl<'a> ParquetReaderBuilder<'a> {
             projection,
             field_paths,
             pruner: self.pruner.clone(),
+            transformer,
             need_page_index: self.options.prune_pages(),
             batch_size,
         })
     }
 
-    pub fn build_row_group_reader(&mut self, need_file_row_number: bool) -> Result<RowGroupReader> {
+    pub fn build_row_group_reader(
+        &mut self,
+        source_type: ParquetSourceType,
+        need_file_row_number: bool,
+    ) -> Result<RowGroupReader> {
         let batch_size = self.ctx.get_settings().get_max_block_size()? as usize;
 
         if !need_file_row_number {
@@ -256,6 +266,14 @@ impl<'a> ParquetReaderBuilder<'a> {
         }
         self.build_output()?;
 
+        let transformer = (!matches!(source_type, ParquetSourceType::StageTable))
+            .then(|| {
+                self.built_output.as_ref().map(|(_, _, output_schema, _)| {
+                    RecordBatchTransformer::build(output_schema.clone())
+                })
+            })
+            .flatten();
+
         let mut policy_builders = default_policy_builders();
         let default_policy = match (self.built_predicate.as_ref(), self.built_topk.as_ref()) {
             (Some(_), Some(_)) => {
@@ -292,6 +310,7 @@ impl<'a> ParquetReaderBuilder<'a> {
             schema_desc: self.schema_desc.clone(),
             policy_builders,
             default_policy,
+            transformer,
         })
     }
 

src/query/storages/parquet/src/parquet_reader/reader/full_reader.rs

@@ -48,6 +48,7 @@ use crate::parquet_reader::predicate::ParquetPredicate;
 use crate::parquet_reader::utils::transform_record_batch;
 use crate::parquet_reader::utils::transform_record_batch_by_field_paths;
 use crate::parquet_reader::utils::FieldPaths;
+use crate::transformer::RecordBatchTransformer;
 use crate::ParquetPruner;
 
 /// The reader to read a whole parquet file.
@@ -71,6 +72,8 @@ pub struct ParquetWholeFileReader {
 
     pub(super) pruner: Option<ParquetPruner>,
 
+    pub(super) transformer: Option<RecordBatchTransformer>,
+
     // Options
     pub(super) need_page_index: bool,
     pub(super) batch_size: usize,
@@ -226,20 +229,27 @@ impl ParquetWholeFileReader {
             }
         }
         let reader = builder.build()?;
+        let mut transformer = self.transformer.clone();
         // Write `if` outside iteration to reduce branches.
         if let Some(field_paths) = self.field_paths.as_ref() {
             reader
                 .into_iter()
                 .map(|batch| {
-                    let batch = batch?;
+                    let mut batch = batch?;
+                    if let Some(transformer) = &mut transformer {
+                        batch = transformer.process_record_batch(batch)?
+                    }
                     transform_record_batch_by_field_paths(&batch, field_paths)
                 })
                 .collect()
         } else {
             reader
                 .into_iter()
                 .map(|batch| {
-                    let batch = batch?;
+                    let mut batch = batch?;
+                    if let Some(transformer) = &mut transformer {
+                        batch = transformer.process_record_batch(batch)?
+                    }
                     Ok(
                         DataBlock::from_record_batch(&self.output_schema.as_ref().into(), &batch)?
                             .0,

src/query/storages/parquet/src/parquet_reader/reader/row_group_reader.rs

@@ -42,6 +42,7 @@ pub struct RowGroupReader {
     pub(super) policy_builders: PolicyBuilders,
 
     pub(super) schema_desc: SchemaDescPtr,
+    pub(super) transformer: Option<RecordBatchTransformer>,
     // Options
     pub(super) batch_size: usize,
 }
@@ -62,7 +63,6 @@ impl RowGroupReader {
         read_settings: &ReadSettings,
         part: &ParquetRowGroupPart,
         topk_sorter: &mut Option<TopKSorter>,
-        transformer: RecordBatchTransformer,
     ) -> Result<Option<ReadPolicyImpl>> {
         if let Some((sorter, min_max)) = topk_sorter.as_ref().zip(part.sort_min_max.as_ref()) {
             if sorter.never_match(min_max) {
@@ -105,7 +105,7 @@ impl RowGroupReader {
                 row_group,
                 selection,
                 topk_sorter,
-                Some(transformer),
+                self.transformer.clone(),
                 self.batch_size,
             )
             .await

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

@@ -16,7 +16,6 @@ use std::sync::Arc;
 
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::InternalColumnType;
-use databend_common_catalog::plan::PushDownInfo;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
@@ -52,9 +51,6 @@ impl ParquetTable {
         let need_row_number = internal_columns.contains(&InternalColumnType::FileRowNumber);
 
         let table_schema: TableSchemaRef = self.table_info.schema();
-        let projection =
-            PushDownInfo::projection_of_push_downs(&table_schema, plan.push_downs.as_ref());
-        let output_schema = Arc::new(projection.project_schema(&table_schema));
         // If there is a `ParquetFilePart`, we should create pruner for it.
         // Although `ParquetFilePart`s are always staying at the end of `parts` when `do_read_partitions`,
         // but parts are reshuffled when `redistribute_source_fragment`, so let us check all of them.
@@ -100,9 +96,14 @@ impl ParquetTable {
             builder = builder.with_pruner(pruner).with_topk(topk.as_ref());
         }
 
-        let row_group_reader = Arc::new(builder.build_row_group_reader(need_row_number)?);
+        let row_group_reader = Arc::new(
+            builder.build_row_group_reader(ParquetSourceType::StageTable, need_row_number)?,
+        );
         let full_file_reader = if has_file_part {
-            Some(Arc::new(builder.build_full_reader(need_row_number)?))
+            Some(Arc::new(builder.build_full_reader(
+                ParquetSourceType::StageTable,
+                need_row_number,
+            )?))
         } else {
             None
         };
@@ -120,7 +121,6 @@ impl ParquetTable {
                     internal_columns.clone(),
                     plan.push_downs.clone(),
                     table_schema.clone(),
-                    output_schema.clone(),
                     op.clone(),
                 )
             },

src/query/storages/parquet/src/source.rs

@@ -57,7 +57,6 @@ use crate::parquet_reader::ParquetWholeFileReader;
 use crate::parquet_reader::RowGroupReader;
 use crate::partition::ParquetRowGroupPart;
 use crate::read_settings::ReadSettings;
-use crate::transformer::RecordBatchTransformer;
 use crate::ParquetFilePart;
 use crate::ParquetPart;
 use crate::ParquetReaderBuilder;
@@ -74,7 +73,8 @@ pub enum ParquetSourceType {
     StageTable,
     ResultCache,
     Iceberg,
-    // DeltaLake,
+    DeltaLake,
+    Hive,
 }
 
 pub struct ParquetSource {
@@ -107,7 +107,6 @@ pub struct ParquetSource {
     push_downs: Option<PushDownInfo>,
     topk: Arc<Option<TopK>>,
     op_registry: Arc<dyn OperatorRegistry>,
-    transformer: RecordBatchTransformer,
 }
 
 impl ParquetSource {
@@ -122,7 +121,6 @@ impl ParquetSource {
         internal_columns: Vec<InternalColumnType>,
         push_downs: Option<PushDownInfo>,
         table_schema: TableSchemaRef,
-        output_schema: TableSchemaRef,
         op_registry: Arc<dyn OperatorRegistry>,
     ) -> Result<ProcessorPtr> {
         let scan_progress = ctx.get_scan_progress();
@@ -133,7 +131,6 @@ impl ParquetSource {
             .as_ref()
             .as_ref()
             .map(|t| TopKSorter::new(t.limit, t.asc));
-        let transformer = RecordBatchTransformer::build(output_schema);
 
         Ok(ProcessorPtr::create(Box::new(Self {
             source_type,
@@ -153,7 +150,6 @@ impl ParquetSource {
             push_downs,
             topk,
             op_registry,
-            transformer,
         })))
     }
 }
@@ -284,7 +280,6 @@ impl Processor for ParquetSource {
                                     ),
                                     part,
                                     &mut self.topk_sorter,
-                                    self.transformer.clone(),
                                 )
                                 .await?
                             {
@@ -355,7 +350,6 @@ impl ParquetSource {
                 part.file.as_str(),
             )?;
         }
-        self.transformer.match_by_field_name(from_stage_table);
         // The schema of the table in iceberg may be inconsistent with the schema in parquet
         let reader = if self.row_group_reader.schema_desc().root_schema()
             != meta.file_metadata().schema_descr().root_schema()
@@ -381,7 +375,9 @@ impl ParquetSource {
                 builder = builder.with_topk(self.topk.as_ref().as_ref());
             }
 
-            Cow::Owned(Arc::new(builder.build_row_group_reader(need_row_number)?))
+            Cow::Owned(Arc::new(
+                builder.build_row_group_reader(self.source_type, need_row_number)?,
+            ))
         } else {
             Cow::Borrowed(&self.row_group_reader)
         };
@@ -408,7 +404,6 @@ impl ParquetSource {
                     &ReadSettings::from_ctx(&self.ctx)?.with_enable_cache(!from_stage_table),
                     &part,
                     &mut self.topk_sorter,
-                    self.transformer.clone(),
                 )
                 .await?;