refactor(query): Improve parse json performance (#18451)

perf(query): Improve parse json performance

Author: b41sh

Cargo.lock

@@ -366,7 +366,7 @@ jaq-interpret = "1.5.0"
 jaq-parse = "1.0.3"
 jaq-std = "1.6.0"
 jiff = { version = "0.2.10", features = ["serde", "tzdb-bundle-always"] }
-jsonb = "0.5.2"
+jsonb = "0.5.3"
 jwt-simple = { version = "0.12.10", default-features = false, features = ["pure-rust"] }
 lenient_semver = "0.4.2"
 levenshtein_automata = "0.2.1"

Cargo.toml

@@ -378,17 +378,15 @@ pub fn cast_scalar_to_variant(
         }
         ScalarRef::Geometry(bytes) => {
             let geom = Ewkb(bytes).to_json().expect("failed to decode wkb data");
-            jsonb::parse_value(geom.as_bytes())
-                .expect("failed to parse geojson to json value")
-                .write_to_vec(buf);
+            jsonb::parse_owned_jsonb_with_buf(geom.as_bytes(), buf)
+                .expect("failed to parse geojson to json value");
             return;
         }
         ScalarRef::Geography(bytes) => {
             // todo: Implement direct conversion, omitting intermediate processes
             let geom = Ewkb(bytes.0).to_json().expect("failed to decode wkb data");
-            jsonb::parse_value(geom.as_bytes())
-                .expect("failed to parse geojson to json value")
-                .write_to_vec(buf);
+            jsonb::parse_owned_jsonb_with_buf(geom.as_bytes(), buf)
+                .expect("failed to parse geojson to json value");
             return;
         }
         ScalarRef::Vector(scalar) => with_vector_number_type!(|NUM_TYPE| match scalar {

src/query/expression/src/types/variant.rs

@@ -14,7 +14,7 @@
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb;
 use jsonb::RawJsonb;
 
 use crate::types::AnyType;
@@ -101,7 +101,7 @@ fn transform_scalar(scalar: ScalarRef<'_>, decode: bool) -> Result<Scalar> {
                 let raw_jsonb = RawJsonb::new(data);
                 Scalar::Variant(raw_jsonb.to_string().into_bytes())
             } else {
-                let value = parse_value(data).map_err(|err| {
+                let value = parse_owned_jsonb(data).map_err(|err| {
                     ErrorCode::UDFDataError(format!("parse json value error: {err}"))
                 })?;
                 Scalar::Variant(value.to_vec())

src/query/expression/src/utils/variant_transform.rs

@@ -28,8 +28,7 @@ use databend_common_expression::FromData;
 use databend_common_expression::RowConverter;
 use databend_common_expression::SortField;
 use itertools::Itertools;
-use jsonb::parse_value;
-use jsonb::RawJsonb;
+use jsonb::parse_owned_jsonb;
 use rand::distributions::Alphanumeric;
 use rand::distributions::Standard;
 use rand::prelude::Distribution;
@@ -386,9 +385,8 @@ fn test_variant() {
     for value in values {
         if let Some(value) = value {
             validity.push(true);
-            let val = parse_value(value.as_bytes()).unwrap();
-            let buf = val.to_vec();
-            let raw_jsonb = RawJsonb::new(&buf);
+            let owned_jsonb = parse_owned_jsonb(value.as_bytes()).unwrap();
+            let raw_jsonb = owned_jsonb.as_raw();
             let compare_buf = raw_jsonb.convert_to_comparable();
             builder.put_slice(&compare_buf);
         } else {

src/query/expression/tests/it/row.rs

@@ -60,7 +60,7 @@ use databend_common_io::parse_bitmap;
 use databend_common_io::parse_bytes_to_ewkb;
 use databend_common_io::prelude::FormatSettings;
 use databend_common_io::Interval;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
 use num_traits::NumCast;
 
@@ -468,9 +468,8 @@ impl FastFieldDecoderValues {
     ) -> Result<()> {
         let mut buf = Vec::new();
         self.read_string_inner(reader, &mut buf, positions)?;
-        match parse_value(&buf) {
-            Ok(value) => {
-                value.write_to_vec(&mut column.data);
+        match parse_owned_jsonb_with_buf(&buf, &mut column.data) {
+            Ok(_) => {
                 column.commit_row();
             }
             Err(_) => {

src/query/formats/src/field_decoder/fast_values.rs

@@ -53,7 +53,7 @@ use databend_common_io::geography::geography_from_ewkt_bytes;
 use databend_common_io::parse_bitmap;
 use databend_common_io::parse_bytes_to_ewkb;
 use databend_common_io::Interval;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
 
 use crate::binary::decode_binary;
@@ -306,9 +306,8 @@ impl NestedValues {
     ) -> Result<()> {
         let mut buf = Vec::new();
         self.read_string_inner(reader, &mut buf)?;
-        match parse_value(&buf) {
-            Ok(value) => {
-                value.write_to_vec(&mut column.data);
+        match parse_owned_jsonb_with_buf(&buf, &mut column.data) {
+            Ok(_) => {
                 column.commit_row();
             }
             Err(e) => {

src/query/formats/src/field_decoder/nested.rs

@@ -51,7 +51,7 @@ use databend_common_io::parse_bytes_to_ewkb;
 use databend_common_io::Interval;
 use databend_common_meta_app::principal::CsvFileFormatParams;
 use databend_common_meta_app::principal::TsvFileFormatParams;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
 use num_traits::NumCast;
 
@@ -287,9 +287,8 @@ impl SeparatedTextDecoder {
     }
 
     fn read_variant(&self, column: &mut BinaryColumnBuilder, data: &[u8]) -> Result<()> {
-        match parse_value(data) {
-            Ok(value) => {
-                value.write_to_vec(&mut column.data);
+        match parse_owned_jsonb_with_buf(data, &mut column.data) {
+            Ok(_) => {
                 column.commit_row();
             }
             Err(e) => {

src/query/formats/src/field_decoder/separated_text.rs

@@ -75,7 +75,7 @@ use geozero::CoordDimensions;
 use geozero::GeozeroGeometry;
 use geozero::ToGeo;
 use geozero::ToWkb;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use jsonb::RawJsonb;
 use num_traits::AsPrimitive;
 use proj4rs::transform::transform;
@@ -134,14 +134,11 @@ pub fn register(registry: &mut FunctionRegistry) {
             }
 
             match ewkb_to_geo(&mut Ewkb(ewkb)).and_then(|(geo, _)| geo_to_json(geo)) {
-                Ok(json) => match parse_value(json.as_bytes()) {
-                    Ok(json_val) => {
-                        json_val.write_to_vec(&mut builder.data);
-                    }
-                    Err(e) => {
+                Ok(json) => {
+                    if let Err(e) = parse_owned_jsonb_with_buf(json.as_bytes(), &mut builder.data) {
                         ctx.set_error(builder.len(), e.to_string());
                     }
-                },
+                }
                 Err(e) => {
                     ctx.set_error(builder.len(), e.to_string());
                 }