Merge pull request #2783 from dantengsky/fix-2782

ISSUE-2782: rm unnecessary clones while merging col stats & some minor file name refactorying

Author: databend-bot

query/src/datasources/table/fuse/table_do_append.rs renamed to query/src/datasources/table/fuse/append.rs

@@ -13,16 +13,18 @@
 //  limitations under the License.
 //
 
+mod append;
 pub(crate) mod index;
 pub(crate) mod io;
 mod meta;
+mod read;
+mod read_plan;
 mod table;
-mod table_do_append;
-mod table_do_read;
-mod table_do_read_partitions;
-mod table_do_truncate;
+mod truncate;
 pub(crate) mod util;
 
+#[cfg(test)]
+mod read_plan_test;
 #[cfg(test)]
 mod table_test;
 #[cfg(test)]

query/src/datasources/table/fuse/mod.rs

@@ -13,16 +13,20 @@
 //  limitations under the License.
 //
 
+use std::collections::HashSet;
+
 use common_base::BlockingWait;
 use common_context::IOContext;
 use common_context::TableIOContext;
 use common_dal::read_obj;
 use common_exception::Result;
 use common_planners::Extras;
+use common_planners::Part;
 use common_planners::Partitions;
 use common_planners::Statistics;
 
 use super::index;
+use crate::datasources::table::fuse::BlockMeta;
 use crate::datasources::table::fuse::FuseTable;
 
 impl FuseTable {
@@ -43,10 +47,43 @@ impl FuseTable {
             }
             .wait_in(&io_ctx.get_runtime(), None)??;
 
-            let (statistics, parts) = Self::to_partitions(&block_metas, push_downs);
+            let (statistics, parts) = to_partitions(&block_metas, push_downs);
             Ok((statistics, parts))
         } else {
             Ok((Statistics::default(), vec![]))
         }
     }
 }
+
+pub(crate) fn to_partitions(
+    blocks_metas: &[BlockMeta],
+    push_downs: Option<Extras>,
+) -> (Statistics, Partitions) {
+    let proj_cols =
+        push_downs.and_then(|extras| extras.projection.map(HashSet::<usize>::from_iter));
+    blocks_metas.iter().fold(
+        (Statistics::default(), Partitions::default()),
+        |(mut stats, mut parts), block_meta| {
+            parts.push(Part {
+                name: block_meta.location.location.clone(),
+                version: 0,
+            });
+
+            stats.read_rows += block_meta.row_count as usize;
+
+            match &proj_cols {
+                Some(proj) => {
+                    stats.read_bytes += block_meta
+                        .col_stats
+                        .iter()
+                        .filter(|(cid, _)| proj.contains(&(**cid as usize)))
+                        .map(|(_, col_stats)| col_stats.in_memory_size)
+                        .sum::<u64>() as usize
+                }
+                None => stats.read_bytes += block_meta.block_size as usize,
+            }
+
+            (stats, parts)
+        },
+    )
+}

query/src/datasources/table/fuse/table_do_read.rs renamed to query/src/datasources/table/fuse/read.rs

@@ -0,0 +1,95 @@
+//  Copyright 2021 Datafuse Labs.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+use std::collections::HashMap;
+
+use common_datavalues::DataValue;
+use common_exception::Result;
+use common_planners::Extras;
+
+use crate::datasources::table::fuse::read_plan::to_partitions;
+use crate::datasources::table::fuse::BlockLocation;
+use crate::datasources::table::fuse::BlockMeta;
+use crate::datasources::table::fuse::ColStats;
+
+#[test]
+fn test_to_partitions() -> Result<()> {
+    // setup
+    let num_of_col = 10;
+    let num_of_block = 5;
+
+    let col_stats_gen = |col_size| ColStats {
+        min: DataValue::Int8(Some(1)),
+        max: DataValue::Int8(Some(2)),
+        null_count: 0,
+        in_memory_size: col_size as u64,
+    };
+
+    let cols_stats = (0..num_of_col)
+        .into_iter()
+        .map(|col_id| (col_id as u32, col_stats_gen(col_id)))
+        .collect::<HashMap<_, _>>();
+
+    let block_meta = BlockMeta {
+        row_count: 0,
+        block_size: cols_stats
+            .iter()
+            .map(|(_, col_stats)| col_stats.in_memory_size)
+            .sum(),
+        col_stats: cols_stats.clone(),
+        location: BlockLocation {
+            location: "".to_string(),
+            meta_size: 0,
+        },
+    };
+
+    let blocks_metas = (0..num_of_block)
+        .into_iter()
+        .map(|_| block_meta.clone())
+        .collect::<Vec<_>>();
+
+    // CASE I:  no projection
+    let (s, _) = to_partitions(&blocks_metas, None);
+    let expected_block_size: u64 = cols_stats
+        .iter()
+        .map(|(_, col_stats)| col_stats.in_memory_size)
+        .sum();
+    assert_eq!(expected_block_size * num_of_block, s.read_bytes as u64);
+
+    // CASE II: col pruning
+    // projection which keeps the odd ones
+    let proj = (0..num_of_col)
+        .into_iter()
+        .filter(|v| v & 1 != 0)
+        .collect::<Vec<usize>>();
+
+    // for each block, the block size we expects (after pruning)
+    let expected_block_size: u64 = cols_stats
+        .iter()
+        .filter(|(cid, _)| proj.contains(&(**cid as usize)))
+        .map(|(_, col_stats)| col_stats.in_memory_size)
+        .sum();
+
+    // kick off
+    let push_down = Some(Extras {
+        projection: Some(proj),
+        filters: vec![],
+        limit: None,
+        order_by: vec![],
+    });
+    let (stats, _) = to_partitions(&blocks_metas, push_down);
+    assert_eq!(expected_block_size * num_of_block, stats.read_bytes as u64);
+    Ok(())
+}

query/src/datasources/table/fuse/table_do_read_partitions.rs renamed to query/src/datasources/table/fuse/read_plan.rs

@@ -14,8 +14,6 @@
 //
 
 use std::any::Any;
-use std::collections::HashSet;
-use std::iter::FromIterator;
 use std::sync::Arc;
 
 use common_context::DataContext;
@@ -26,7 +24,6 @@ use common_exception::Result;
 use common_meta_types::TableInfo;
 use common_planners::Extras;
 use common_planners::InsertIntoPlan;
-use common_planners::Part;
 use common_planners::Partitions;
 use common_planners::ReadDataSourcePlan;
 use common_planners::Statistics;
@@ -35,7 +32,6 @@ use common_streams::SendableDataBlockStream;
 
 use super::util;
 use crate::catalogs::Table;
-use crate::datasources::table::fuse::BlockMeta;
 use crate::datasources::table::fuse::TableSnapshot;
 
 pub struct FuseTable {
@@ -122,37 +118,4 @@ impl FuseTable {
             Ok(None)
         }
     }
-
-    pub(crate) fn to_partitions(
-        blocks_metas: &[BlockMeta],
-        push_downs: Option<Extras>,
-    ) -> (Statistics, Partitions) {
-        let proj_cols =
-            push_downs.and_then(|extras| extras.projection.map(HashSet::<usize>::from_iter));
-        blocks_metas.iter().fold(
-            (Statistics::default(), Partitions::default()),
-            |(mut stats, mut parts), block_meta| {
-                parts.push(Part {
-                    name: block_meta.location.location.clone(),
-                    version: 0,
-                });
-
-                stats.read_rows += block_meta.row_count as usize;
-
-                match &proj_cols {
-                    Some(proj) => {
-                        stats.read_bytes += block_meta
-                            .col_stats
-                            .iter()
-                            .filter(|(cid, _)| proj.contains(&(**cid as usize)))
-                            .map(|(_, col_stats)| col_stats.in_memory_size)
-                            .sum::<u64>() as usize
-                    }
-                    None => stats.read_bytes += block_meta.block_size as usize,
-                }
-
-                (stats, parts)
-            },
-        )
-    }
 }

query/src/datasources/table/fuse/read_plan_test.rs

@@ -13,24 +13,17 @@
 //  limitations under the License.
 //
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_base::tokio;
-use common_datavalues::DataValue;
 use common_exception::Result;
-use common_planners::Extras;
 use common_planners::ReadDataSourcePlan;
 use common_planners::TruncateTablePlan;
 use futures::TryStreamExt;
 
 use crate::catalogs::Catalog;
 use crate::catalogs::ToReadDataSourcePlan;
 use crate::datasources::table::fuse::table_test_fixture::TestFixture;
-use crate::datasources::table::fuse::BlockLocation;
-use crate::datasources::table::fuse::BlockMeta;
-use crate::datasources::table::fuse::ColStats;
-use crate::datasources::table::fuse::FuseTable;
 
 #[tokio::test]
 async fn test_fuse_table_simple_case() -> Result<()> {
@@ -202,73 +195,3 @@ async fn test_fuse_table_truncate() -> Result<()> {
 
     Ok(())
 }
-
-#[test]
-fn test_fuse_table_to_partitions_stats_with_col_pruning() -> Result<()> {
-    // setup
-    let num_of_col = 10;
-    let num_of_block = 5;
-
-    let col_stats_gen = |col_size| ColStats {
-        min: DataValue::Int8(Some(1)),
-        max: DataValue::Int8(Some(2)),
-        null_count: 0,
-        in_memory_size: col_size as u64,
-    };
-
-    let cols_stats = (0..num_of_col)
-        .into_iter()
-        .map(|col_id| (col_id as u32, col_stats_gen(col_id)))
-        .collect::<HashMap<_, _>>();
-
-    let block_meta = BlockMeta {
-        row_count: 0,
-        block_size: cols_stats
-            .iter()
-            .map(|(_, col_stats)| col_stats.in_memory_size)
-            .sum(),
-        col_stats: cols_stats.clone(),
-        location: BlockLocation {
-            location: "".to_string(),
-            meta_size: 0,
-        },
-    };
-
-    let blocks_metas = (0..num_of_block)
-        .into_iter()
-        .map(|_| block_meta.clone())
-        .collect::<Vec<_>>();
-
-    // CASE I:  no projection
-    let (s, _) = FuseTable::to_partitions(&blocks_metas, None);
-    let expected_block_size: u64 = cols_stats
-        .iter()
-        .map(|(_, col_stats)| col_stats.in_memory_size)
-        .sum();
-    assert_eq!(expected_block_size * num_of_block, s.read_bytes as u64);
-
-    // CASE II: col pruning
-    // projection which keeps the odd ones
-    let proj = (0..num_of_col)
-        .into_iter()
-        .filter(|v| v & 1 != 0)
-        .collect::<Vec<usize>>();
-
-    // for each block, the block size we expects (after pruning)
-    let expected_block_size: u64 = cols_stats
-        .iter()
-        .filter(|(cid, _)| proj.contains(&(**cid as usize)))
-        .map(|(_, col_stats)| col_stats.in_memory_size)
-        .sum();
-
-    // kick off
-    let push_down = Some(Extras {
-        projection: Some(proj),
-        filters: vec![],
-        limit: None,
-        order_by: vec![],
-    });
-    let (stats, _) = FuseTable::to_partitions(&blocks_metas, push_down);
-    assert_eq!(expected_block_size * num_of_block, stats.read_bytes as u64);
-    Ok(())
-}

query/src/datasources/table/fuse/table.rs

@@ -13,6 +13,7 @@
 //  limitations under the License.
 //
 
+use std::borrow::Borrow;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 
@@ -137,15 +138,15 @@ pub(super) fn block_stats(data_block: &DataBlock) -> Result<BlockStats> {
         .collect()
 }
 
-pub fn column_stats_reduce_with_schema(
-    stats: &[HashMap<ColumnId, ColStats>],
+pub fn column_stats_reduce_with_schema<T: Borrow<HashMap<ColumnId, ColStats>>>(
+    stats: &[T],
     schema: &DataSchema,
 ) -> Result<HashMap<ColumnId, ColStats>> {
     let len = stats.len();
 
     // transpose Vec<HashMap<_,(_,_)>> to HashMap<_, (_, Vec<_>)>
     let col_stat_list = stats.iter().fold(HashMap::new(), |acc, item| {
-        item.iter().fold(
+        item.borrow().iter().fold(
             acc,
             |mut acc: HashMap<ColumnId, Vec<&ColStats>>, (col_id, stats)| {
                 let entry = acc.entry(*col_id);
@@ -185,7 +186,7 @@ pub fn column_stats_reduce_with_schema(
 
             // TODO
             // for some data types, we shall balance the accuracy and the length
-            // e.g. for a string col, which max value is "xxxxxxxxxxxx....", we record the max as something like "y"
+            // e.g. for a string col, which max value is "abcdef....", we record the max as something like "b"
             let min =
                 common_datavalues::DataValue::try_into_data_array(min_stats.as_slice(), data_type)?
                     .min()?;
@@ -210,10 +211,7 @@ pub fn merge_stats(schema: &DataSchema, l: &Stats, r: &Stats) -> Result<Stats> {
         block_count: l.block_count + r.block_count,
         uncompressed_byte_size: l.uncompressed_byte_size + r.uncompressed_byte_size,
         compressed_byte_size: l.compressed_byte_size + r.compressed_byte_size,
-        col_stats: util::column_stats_reduce_with_schema(
-            &[l.col_stats.clone(), r.col_stats.clone()],
-            schema,
-        )?,
+        col_stats: util::column_stats_reduce_with_schema(&[&l.col_stats, &r.col_stats], schema)?,
     };
     Ok(s)
 }