Merge branch 'main' into view-schema

Author: mergify[bot]

Cargo.lock

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

Makefile

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

scripts/setup/rust-toolchain.toml

@@ -34,12 +34,13 @@ simd = ["arrow/simd"]
 # Workspace dependencies
 
 # Crates.io dependencies
-arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "95e117d", default-features = false, features = [
+arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "1da33ac", default-features = false, features = [
     "io_parquet",
     "io_parquet_compression",
 ] }
+
 arrow-format = { version = "0.8.0", features = ["flight-data", "flight-service", "ipc"] }
 futures = "0.3.24"
-parquet2 = { version = "0.16.3", default_features = false }
+parquet2 = { version = "0.17.0", default_features = false }
 
 [dev-dependencies]

src/common/arrow/Cargo.toml

@@ -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/src/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/common/jsonb/tests/it/functions.rs

@@ -40,6 +40,7 @@ pub fn serialize_data_blocks_with_compression(
         write_statistics: false,
         compression,
         version: Version::V2,
+        data_pagesize_limit: None,
     };
     let batches = blocks
         .into_iter()
@@ -87,16 +88,12 @@ pub fn serialize_data_blocks(
     serialize_data_blocks_with_compression(blocks, schema, buf, CompressionOptions::Lz4Raw)
 }
 
-fn col_encoding(_data_type: &ArrowDataType) -> Encoding {
-    // Although encoding does work, parquet2 has not implemented decoding of DeltaLengthByteArray yet, we fallback to Plain
-    // From parquet2: Decoding "DeltaLengthByteArray"-encoded required V2 pages is not yet implemented for Binary.
-    //
-    // match data_type {
-    //    ArrowDataType::Binary
-    //    | ArrowDataType::LargeBinary
-    //    | ArrowDataType::Utf8
-    //    | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
-    //    _ => Encoding::Plain,
-    //}
-    Encoding::Plain
+fn col_encoding(data_type: &ArrowDataType) -> Encoding {
+    match data_type {
+        ArrowDataType::Binary
+        | ArrowDataType::LargeBinary
+        | ArrowDataType::Utf8
+        | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
+        _ => Encoding::Plain,
+    }
 }

src/query/datablocks/src/serialize.rs

@@ -149,10 +149,6 @@ pub trait Column: Send + Sync {
         DFTryFrom::try_from(value)
     }
 
-    fn to_values(&self) -> Vec<DataValue> {
-        (0..self.len()).map(|i| self.get(i)).collect()
-    }
-
     /// Visit each row value of Column
     fn for_each<F>(&self, f: F)
     where

src/query/datavalues/src/columns/column.rs

@@ -120,9 +120,7 @@ impl AggregateFunction for AggregateApproxCountDistinctFunction {
     fn accumulate_row(&self, place: StateAddr, columns: &[ColumnRef], _row: usize) -> Result<()> {
         let state = place.get::<AggregateApproxCountDistinctState>();
         for column in columns {
-            column.to_values().iter().for_each(|value| {
-                state.hll.push(value);
-            });
+            (0..column.len()).for_each(|i| state.hll.push(&column.get(i)));
         }
         Ok(())
     }

src/query/functions/src/aggregates/aggregate_approx_count_distinct.rs

@@ -73,7 +73,7 @@ impl Sink for SubqueryReceiveSink {
     fn consume(&mut self, data_block: DataBlock) -> Result<()> {
         for column_index in 0..data_block.num_columns() {
             let column = data_block.column(column_index);
-            let mut values = column.to_values();
+            let mut values = (0..column.len()).map(|i| column.get(i)).collect();
             self.input_columns[column_index].append(&mut values)
         }