Merge pull request #9341 from BohuTANG/dev-refactor-fuse

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

Author: BohuTANG

Cargo.toml

@@ -45,7 +45,7 @@ members = [
     "src/query/sql",
     "src/query/storages/cache",
     "src/query/storages/factory",
-    "src/query/storages/fuse/fuse",
+    "src/query/storages/fuse",
     "src/query/storages/hive/hive",
     "src/query/storages/hive/hive-meta-store",
     "src/query/storages/index",

src/query/service/Cargo.toml

@@ -58,7 +58,7 @@ common-sql = { path = "../sql" }
 common-storage = { path = "../../common/storage" }
 common-storages-cache = { path = "../storages/cache" }
 common-storages-factory = { path = "../storages/factory" }
-common-storages-fuse = { path = "../storages/fuse/fuse" }
+common-storages-fuse = { path = "../storages/fuse" }
 common-storages-hive = { path = "../storages/hive/hive", optional = true }
 common-storages-index = { path = "../storages/index" }
 common-storages-information-schema = { path = "../storages/information-schema" }

src/query/storages/factory/Cargo.toml

@@ -16,7 +16,7 @@ common-catalog = { path = "../../catalog" }
 common-config = { path = "../../config" }
 common-exception = { path = "../../../common/exception" }
 common-meta-app = { path = "../../../meta/app" }
-common-storages-fuse = { path = "../fuse/fuse" }
+common-storages-fuse = { path = "../fuse" }
 common-storages-index = { path = "../index" }
 common-storages-memory = { path = "../memory" }
 common-storages-null = { path = "../null" }

src/query/storages/fuse/Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "common-storages-fuse"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[lib]
+doctest = false
+test = false
+
+[dependencies]
+common-arrow = { path = "../../../common/arrow" }
+common-base = { path = "../../../common/base" }
+common-cache = { path = "../../../common/cache" }
+common-catalog = { path = "../../catalog" }
+common-config = { path = "../../config" }
+common-datablocks = { path = "../../datablocks" }
+common-datavalues = { path = "../../datavalues" }
+common-exception = { path = "../../../common/exception" }
+common-functions = { path = "../../functions" }
+common-meta-app = { path = "../../../meta/app" }
+common-meta-types = { path = "../../../meta/types" }
+common-pipeline-core = { path = "../../pipeline/core" }
+common-pipeline-sources = { path = "../../pipeline/sources" }
+common-pipeline-transforms = { path = "../../pipeline/transforms" }
+common-sharing = { path = "../../sharing" }
+common-sql = { path = "../../sql" }
+common-storage = { path = "../../../common/storage" }
+common-storages-cache = { path = "../cache" }
+common-storages-index = { path = "../index" }
+common-storages-pruner = { path = "../pruner" }
+common-storages-table-meta = { path = "../table-meta" }
+
+async-trait = { version = "0.1.57", package = "async-trait-fn" }
+backoff = { version = "0.4.0", features = ["futures", "tokio"] }
+backon = "0.2"
+chrono = { workspace = true }
+futures = "0.3.24"
+futures-util = "0.3.24"
+itertools = "0.10.5"
+metrics = "0.20.1"
+opendal = "0.22"
+serde = { workspace = true }
+serde_json = { workspace = true }
+tracing = "0.1.36"
+typetag = "0.2.3"
+uuid = { version = "1.1.2", features = ["serde", "v4"] }

src/query/storages/fuse/fuse/Cargo.toml

@@ -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(