refactor(fuse): remove unused codes after fuse-result purged

Author: BohuTANG

src/query/storages/fuse/fuse/src/io/read/block_reader_parquet.rs

@@ -205,71 +205,6 @@ impl BlockReader {
         Ok((num_rows, columns_array_iter))
     }
 
-    async fn read_columns(&self, part: PartInfoPtr) -> Result<(usize, Vec<ArrayIter<'static>>)> {
-        let part = FusePartInfo::from_part(&part)?;
-
-        // TODO: add prefetch column data.
-        let num_rows = part.nums_rows;
-        let num_cols = self.projection.len();
-        let mut column_chunk_futs = Vec::with_capacity(num_cols);
-
-        let columns = self.projection.project_column_leaves(&self.column_leaves)?;
-        let indices = Self::build_projection_indices(&columns);
-        for (index, _) in indices {
-            let column_meta = &part.columns_meta[&index];
-            let column_reader = self.operator.object(&part.location);
-            let fut = async move {
-                let (idx, column_chunk) =
-                    Self::read_column(column_reader, index, column_meta.offset, column_meta.len)
-                        .await?;
-                Ok::<_, ErrorCode>((idx, column_chunk))
-            }
-            .instrument(debug_span!("read_col_chunk"));
-            column_chunk_futs.push(fut);
-        }
-
-        let num_cols = column_chunk_futs.len();
-        let chunks = futures::stream::iter(column_chunk_futs)
-            .buffered(std::cmp::min(10, num_cols))
-            .try_collect::<Vec<_>>()
-            .await?;
-
-        let mut chunk_map: HashMap<usize, Vec<u8>> = chunks.into_iter().collect();
-        let mut cnt_map = Self::build_projection_count_map(&columns);
-        let mut columns_array_iter = Vec::with_capacity(num_cols);
-        for column in &columns {
-            let field = column.field.clone();
-            let indices = &column.leaf_ids;
-            let mut column_metas = Vec::with_capacity(indices.len());
-            let mut column_chunks = Vec::with_capacity(indices.len());
-            let mut column_descriptors = Vec::with_capacity(indices.len());
-            for index in indices {
-                let column_meta = &part.columns_meta[index];
-                let cnt = cnt_map.get_mut(index).unwrap();
-                *cnt -= 1;
-                let column_chunk = if cnt > &mut 0 {
-                    chunk_map.get(index).unwrap().clone()
-                } else {
-                    chunk_map.remove(index).unwrap()
-                };
-                let column_descriptor = &self.parquet_schema_descriptor.columns()[*index];
-                column_metas.push(column_meta);
-                column_chunks.push(column_chunk);
-                column_descriptors.push(column_descriptor);
-            }
-            columns_array_iter.push(Self::to_array_iter(
-                column_metas,
-                column_chunks,
-                num_rows,
-                column_descriptors,
-                field,
-                &part.compression,
-            )?);
-        }
-
-        Ok((num_rows, columns_array_iter))
-    }
-
     pub fn build_block(&self, chunks: Vec<(usize, Box<dyn Array>)>) -> Result<DataBlock> {
         let mut results = Vec::with_capacity(chunks.len());
         let mut chunk_map: HashMap<usize, Box<dyn Array>> = chunks.into_iter().collect();
@@ -402,13 +337,6 @@ impl BlockReader {
         Ok((index, chunk))
     }
 
-    #[tracing::instrument(level = "debug", skip_all)]
-    pub async fn read(&self, part: PartInfoPtr) -> Result<DataBlock> {
-        let (num_rows, columns_array_iter) = self.read_columns(part).await?;
-        let mut deserializer = RowGroupDeserializer::new(columns_array_iter, num_rows, None);
-        self.try_next_block(&mut deserializer)
-    }
-
     fn try_next_block(&self, deserializer: &mut RowGroupDeserializer) -> Result<DataBlock> {
         match deserializer.next() {
             None => Err(ErrorCode::Internal(