fix: fix native internal column bug

Author: lichuang

src/query/storages/fuse/src/operations/read/native_data_source_deserializer.rs

@@ -282,16 +282,29 @@ impl NativeDeserializeDataTransform {
         Ok(())
     }
 
+    fn fill_block_meta_index(data_block: DataBlock, fuse_part: &FusePartInfo) -> Result<DataBlock> {
+        // Fill `BlockMetaInfoPtr` if query internal columns
+        let meta: Option<BlockMetaInfoPtr> =
+            Some(Box::new(fuse_part.block_meta_index().unwrap().to_owned()));
+        data_block.add_meta(meta)
+    }
+
     /// All columns are default values, not need to read.
     fn finish_process_with_default_values(&mut self) -> Result<()> {
         let _ = self.chunks.pop_front();
         let part = self.parts.pop_front().unwrap();
-        let part = FusePartInfo::from_part(&part)?;
+        let fuse_part = FusePartInfo::from_part(&part)?;
 
-        let num_rows = part.nums_rows;
+        let num_rows = fuse_part.nums_rows;
         let data_block = self.block_reader.build_default_values_block(num_rows)?;
         let data_block = data_block.resort(&self.src_schema, &self.output_schema)?;
 
+        let data_block = if !self.block_reader.query_internal_columns() {
+            data_block
+        } else {
+            Self::fill_block_meta_index(data_block, fuse_part)?
+        };
+
         self.add_block(data_block)?;
 
         self.inited = false;
@@ -304,10 +317,16 @@ impl NativeDeserializeDataTransform {
     fn finish_process_with_empty_block(&mut self) -> Result<()> {
         let _ = self.chunks.pop_front();
         let part = self.parts.pop_front().unwrap();
-        let part = FusePartInfo::from_part(&part)?;
+        let fuse_part = FusePartInfo::from_part(&part)?;
 
-        let num_rows = part.nums_rows;
+        let num_rows = fuse_part.nums_rows;
         let data_block = DataBlock::new(vec![], num_rows);
+        let data_block = if !self.block_reader.query_internal_columns() {
+            data_block
+        } else {
+            Self::fill_block_meta_index(data_block, fuse_part)?
+        };
+
         self.add_block(data_block)?;
         Ok(())
     }

tests/sqllogictests/suites/base/00_dummy/05_0031_internal_column

@@ -5,7 +5,7 @@ statement ok
 DROP TABLE IF EXISTS `05_0031_t`
 
 statement ok
-CREATE TABLE `05_0031_t`(a int)
+CREATE TABLE `05_0031_t`(a int, b int default 2)
 
 statement error 1110
 ALTER TABLE `05_0031_t` ADD COLUMN _row_id float
@@ -14,17 +14,17 @@ statement error 1110
 CREATE TABLE `05_0031_t_1`(_row_id int)
 
 statement ok
-INSERT INTO TABLE `05_0031_t` values(1),(2)
+INSERT INTO TABLE `05_0031_t` (a) values(1),(2)
 
 statement ok
-INSERT INTO TABLE `05_0031_t` values(3)
+INSERT INTO TABLE `05_0031_t` (a) values(3)
 
 query II
-SELECT a,_row_id FROM `05_0031_t` order by _row_id
+SELECT b,_row_id,a FROM `05_0031_t` order by _row_id
 ----
-1 0
-2 1
-3 4194304
+2 0 1
+2 1 2
+2 4194304 3
 
 query II
 SELECT a,_row_id FROM `05_0031_t` where _row_id = 0 
@@ -35,7 +35,7 @@ statement ok
 DROP TABLE IF EXISTS `05_0031_t_1`
 
 statement ok
-CREATE TABLE `05_0031_t_1`(b int)
+CREATE TABLE `05_0031_t_1`(c int)
 
 statement ok
 INSERT INTO TABLE `05_0031_t_1` values(5),(6)