Merge branch 'main' into issue_9057

Author: mergify[bot]

Makefile

@@ -62,6 +62,10 @@ build-native:
 unit-test:
 	ulimit -n 10000;ulimit -s 16384; RUST_LOG="ERROR" bash ./scripts/ci/ci-run-unit-tests.sh
 
+miri:
+	cargo miri setup
+	MIRIFLAGS="-Zmiri-disable-isolation" cargo miri test --no-default-features
+
 embedded-meta-test: build
 	rm -rf ./_meta_embedded*
 	bash ./scripts/ci/ci-run-tests-embedded-meta.sh

scripts/setup/rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
 channel = "nightly-2022-11-14"
-components = ["rustfmt", "clippy", "rust-src"]
+components = ["rustfmt", "clippy", "rust-src", "miri"]

src/common/jsonb/src/functions.rs

@@ -816,6 +816,7 @@ pub fn parse_json_path(path: &[u8]) -> Result<Vec<JsonPath>, Error> {
                 }
                 let s = std::str::from_utf8(&path[prev_idx..idx - 2])?;
                 let json_path = JsonPath::String(Cow::Borrowed(s));
+
                 json_paths.push(json_path);
             } else {
                 prev_idx = idx - 1;
@@ -836,6 +837,34 @@ pub fn parse_json_path(path: &[u8]) -> Result<Vec<JsonPath>, Error> {
                     return Err(Error::InvalidToken);
                 }
             }
+        } else if c == b'"' {
+            prev_idx = idx;
+            loop {
+                let c = read_char(path, &mut idx)?;
+                if c == b'\\' {
+                    idx += 1;
+                    let c = read_char(path, &mut idx)?;
+                    if c == b'"' {
+                        idx += 1;
+                    }
+                } else if c != b'"' {
+                    idx += 1;
+                } else {
+                    // Try to read to check if has extra strings, string value can only have one.
+                    let c = read_char(path, &mut idx);
+                    match c {
+                        Ok(_) => return Err(Error::InvalidToken),
+                        Err(_) => break,
+                    }
+                }
+            }
+            let s = std::str::from_utf8(&path[prev_idx..idx - 1])?;
+            let json_path = JsonPath::String(Cow::Borrowed(s));
+            if json_paths.is_empty() {
+                json_paths.push(json_path);
+            } else {
+                return Err(Error::InvalidToken);
+            }
         } else {
             if c == b':' || c == b'.' {
                 if idx == 1 {

src/common/jsonb/tests/it/functions.rs

@@ -32,6 +32,7 @@ use common_jsonb::object_keys;
 use common_jsonb::parse_json_path;
 use common_jsonb::parse_value;
 use common_jsonb::to_string;
+use common_jsonb::Error;
 use common_jsonb::JsonPath;
 use common_jsonb::Number;
 use common_jsonb::Object;
@@ -361,6 +362,8 @@ fn test_parse_json_path() {
             JsonPath::String(Cow::from("k2")),
             JsonPath::String(Cow::from("k3")),
         ]),
+        ("\"k1\"", vec![JsonPath::String(Cow::from("k1"))]),
+        ("\"k1k2\"", vec![JsonPath::String(Cow::from("k1k2"))]),
         (r#"k1["k2"][1]"#, vec![
             JsonPath::String(Cow::from("k1")),
             JsonPath::String(Cow::from("k2")),
@@ -372,6 +375,18 @@ fn test_parse_json_path() {
         let path = parse_json_path(s.as_bytes()).unwrap();
         assert_eq!(&path[..], &expect[..]);
     }
+
+    let wrong_sources = vec![
+        (r#"\"\"\\k1\"\""#, Error::InvalidToken),
+        (r#"\\k1\\'"#, Error::InvalidToken),
+    ];
+    for (s, expect) in wrong_sources {
+        let path = parse_json_path(s.as_bytes());
+        match path {
+            Ok(_) => println!(),
+            Err(_) => assert_eq!(Error::InvalidToken, expect),
+        }
+    }
 }
 
 #[test]

src/query/service/tests/it/storages/fuse/operations/mutation/segments_compact_mutator.rs

@@ -663,9 +663,10 @@ impl CompactSegmentTestFixture {
             let mut stats_acc = StatisticsAccumulator::default();
             for block in blocks {
                 let block = block?;
-                let col_stats = gen_columns_statistics(&block)?;
+                let col_stats = gen_columns_statistics(&block, None)?;
 
-                let mut block_statistics = BlockStatistics::from(&block, "".to_owned(), None)?;
+                let mut block_statistics =
+                    BlockStatistics::from(&block, "".to_owned(), None, None)?;
                 let block_meta = block_writer.write(block, col_stats, None).await?;
                 block_statistics.block_file_location = block_meta.location.0.clone();
 

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -46,13 +46,50 @@ use crate::storages::fuse::table_test_fixture::TestFixture;
 
 #[test]
 fn test_ft_stats_block_stats() -> common_exception::Result<()> {
-    let schema = DataSchemaRefExt::create(vec![DataField::new("a", i32::to_data_type())]);
-    let block = DataBlock::create(schema, vec![Series::from_data(vec![1, 2, 3])]);
-    let r = gen_columns_statistics(&block)?;
-    assert_eq!(1, r.len());
+    let schema = DataSchemaRefExt::create(vec![
+        DataField::new("a", i32::to_data_type()),
+        DataField::new("b", Vu8::to_data_type()),
+    ]);
+    let block = DataBlock::create(schema, vec![
+        Series::from_data(vec![1, 2, 3]),
+        Series::from_data(vec!["aa", "aa", "bb"]),
+    ]);
+    let r = gen_columns_statistics(&block, None)?;
+    assert_eq!(2, r.len());
+    let col_stats = r.get(&0).unwrap();
+    assert_eq!(col_stats.min, DataValue::Int64(1));
+    assert_eq!(col_stats.max, DataValue::Int64(3));
+    assert_eq!(col_stats.distinct_of_values, Some(3));
+    let col_stats = r.get(&1).unwrap();
+    assert_eq!(col_stats.min, DataValue::String(b"aa".to_vec()));
+    assert_eq!(col_stats.max, DataValue::String(b"bb".to_vec()));
+    assert_eq!(col_stats.distinct_of_values, Some(2));
+    Ok(())
+}
+
+#[test]
+fn test_ft_stats_block_stats_with_column_distinct_count() -> common_exception::Result<()> {
+    let schema = DataSchemaRefExt::create(vec![
+        DataField::new("a", i32::to_data_type()),
+        DataField::new("b", Vu8::to_data_type()),
+    ]);
+    let block = DataBlock::create(schema, vec![
+        Series::from_data(vec![1, 2, 3]),
+        Series::from_data(vec!["aa", "aa", "bb"]),
+    ]);
+    let mut column_distinct_count = HashMap::new();
+    column_distinct_count.insert(0, 3);
+    column_distinct_count.insert(1, 2);
+    let r = gen_columns_statistics(&block, Some(column_distinct_count))?;
+    assert_eq!(2, r.len());
     let col_stats = r.get(&0).unwrap();
     assert_eq!(col_stats.min, DataValue::Int64(1));
     assert_eq!(col_stats.max, DataValue::Int64(3));
+    assert_eq!(col_stats.distinct_of_values, Some(3));
+    let col_stats = r.get(&1).unwrap();
+    assert_eq!(col_stats.min, DataValue::String(b"aa".to_vec()));
+    assert_eq!(col_stats.max, DataValue::String(b"bb".to_vec()));
+    assert_eq!(col_stats.distinct_of_values, Some(2));
     Ok(())
 }
 
@@ -69,7 +106,7 @@ fn test_ft_tuple_stats_block_stats() -> common_exception::Result<()> {
     let column = StructColumn::from_data(inner_columns, tuple_data_type).arc();
 
     let block = DataBlock::create(schema, vec![column]);
-    let r = gen_columns_statistics(&block)?;
+    let r = gen_columns_statistics(&block, None)?;
     assert_eq!(2, r.len());
     let col0_stats = r.get(&0).unwrap();
     assert_eq!(col0_stats.min, DataValue::Int64(1));
@@ -89,7 +126,7 @@ fn test_ft_stats_col_stats_reduce() -> common_exception::Result<()> {
     let blocks = TestFixture::gen_sample_blocks_ex(num_of_blocks, rows_per_block, val_start_with);
     let col_stats = blocks
         .iter()
-        .map(|b| gen_columns_statistics(&b.clone().unwrap()))
+        .map(|b| gen_columns_statistics(&b.clone().unwrap(), None))
         .collect::<common_exception::Result<Vec<_>>>()?;
     let r = reducers::reduce_block_statistics(&col_stats, None);
     assert!(r.is_ok());
@@ -158,8 +195,9 @@ async fn test_accumulator() -> common_exception::Result<()> {
 
     for item in blocks {
         let block = item?;
-        let col_stats = gen_columns_statistics(&block)?;
-        let block_statistics = BlockStatistics::from(&block, "does_not_matter".to_owned(), None)?;
+        let col_stats = gen_columns_statistics(&block, None)?;
+        let block_statistics =
+            BlockStatistics::from(&block, "does_not_matter".to_owned(), None, None)?;
         let block_writer = BlockWriter::new(&operator, &loc_generator);
         let block_meta = block_writer.write(block, col_stats, None).await?;
         stats_acc.add_with_block_meta(block_meta, block_statistics)?;
@@ -344,7 +382,7 @@ fn test_ft_stats_block_stats_string_columns_trimming_using_eval() -> common_exce
         let max_expr = max_col.get(0);
 
         // generate the statistics of column
-        let stats_of_columns = gen_columns_statistics(&block).unwrap();
+        let stats_of_columns = gen_columns_statistics(&block, None).unwrap();
 
         // check if the max value (untrimmed) is in degenerated condition:
         // - the length of string value is larger or equal than STRING_PREFIX_LEN

src/query/service/tests/it/storages/statistics/column_statistics.rs

@@ -89,15 +89,15 @@ fn test_column_traverse() -> Result<()> {
     let cols = traverse::traverse_columns_dfs(sample_block.columns())?;
 
     assert_eq!(5, cols.len());
-    (0..5).for_each(|i| assert_eq!(cols[i], sample_cols[i], "checking col {}", i));
+    (0..5).for_each(|i| assert_eq!(cols[i].1, sample_cols[i], "checking col {}", i));
 
     Ok(())
 }
 
 #[test]
 fn test_column_statistic() -> Result<()> {
     let (sample_block, sample_cols) = gen_sample_block();
-    let col_stats = gen_columns_statistics(&sample_block)?;
+    let col_stats = gen_columns_statistics(&sample_block, None)?;
 
     assert_eq!(5, col_stats.len());
 

src/query/storages/fuse/fuse-result/src/result_table_sink.rs

@@ -176,7 +176,7 @@ impl Processor for ResultTableSink {
         match std::mem::replace(&mut self.state, State::None) {
             State::NeedSerialize(block) => {
                 let location = self.locations.gen_block_location();
-                let block_statistics = BlockStatistics::from(&block, location.clone(), None)?;
+                let block_statistics = BlockStatistics::from(&block, location.clone(), None, None)?;
 
                 let mut data = Vec::with_capacity(100 * 1024 * 1024);
                 let schema = block.schema().clone();

src/query/storages/fuse/fuse-result/src/writer.rs

@@ -101,7 +101,7 @@ impl ResultTableWriter {
     pub async fn append_block(&mut self, block: DataBlock) -> Result<PartInfoPtr> {
         let location = self.locations.gen_block_location();
         let mut data = Vec::with_capacity(100 * 1024 * 1024);
-        let block_statistics = BlockStatistics::from(&block, location.clone(), None)?;
+        let block_statistics = BlockStatistics::from(&block, location.clone(), None, None)?;
         let schema = block.schema().clone();
         let (size, meta_data) = serialize_data_blocks(vec![block], &schema, &mut data)?;
 

src/query/storages/fuse/fuse/src/operations/fuse_sink.rs

@@ -13,6 +13,7 @@
 //  limitations under the License.
 
 use std::any::Any;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -51,6 +52,33 @@ struct BloomIndexState {
     location: Location,
 }
 
+impl BloomIndexState {
+    pub fn try_create(
+        block: &DataBlock,
+        location: Location,
+    ) -> Result<(Self, HashMap<usize, usize>)> {
+        // write index
+        let bloom_index = BlockFilter::try_create(&[block])?;
+        let index_block = bloom_index.filter_block;
+        let mut data = Vec::with_capacity(100 * 1024);
+        let index_block_schema = &bloom_index.filter_schema;
+        let (size, _) = serialize_data_blocks_with_compression(
+            vec![index_block],
+            index_block_schema,
+            &mut data,
+            CompressionOptions::Uncompressed,
+        )?;
+        Ok((
+            Self {
+                data,
+                size,
+                location,
+            },
+            bloom_index.column_distinct_count,
+        ))
+    }
+}
+
 enum State {
     None,
     NeedSerialize(DataBlock),
@@ -168,28 +196,16 @@ impl Processor for FuseTableSink {
 
                 let (block_location, block_id) = self.meta_locations.gen_block_location();
 
-                let bloom_index_state = {
-                    // write index
-                    let bloom_index = BlockFilter::try_create(&[&block])?;
-                    let index_block = bloom_index.filter_block;
-                    let location = self.meta_locations.block_bloom_index_location(&block_id);
-                    let mut data = Vec::with_capacity(100 * 1024);
-                    let index_block_schema = &bloom_index.filter_schema;
-                    let (size, _) = serialize_data_blocks_with_compression(
-                        vec![index_block],
-                        index_block_schema,
-                        &mut data,
-                        CompressionOptions::Uncompressed,
-                    )?;
-                    BloomIndexState {
-                        data,
-                        size,
-                        location,
-                    }
-                };
+                let location = self.meta_locations.block_bloom_index_location(&block_id);
+                let (bloom_index_state, column_distinct_count) =
+                    BloomIndexState::try_create(&block, location)?;
 
-                let block_statistics =
-                    BlockStatistics::from(&block, block_location.0, cluster_stats)?;
+                let block_statistics = BlockStatistics::from(
+                    &block,
+                    block_location.0,
+                    cluster_stats,
+                    Some(column_distinct_count),
+                )?;
                 // we need a configuration of block size threshold here
                 let mut data = Vec::with_capacity(100 * 1024 * 1024);
                 let schema = block.schema().clone();