chore(query): pick improve parse json performance to release/v1.2.636-rc8.5 (#18468)

* perf(query): Improve parse json performance

* ignore wasmtime audit

Author: b41sh

.cargo/audit.toml

@@ -85,4 +85,6 @@ ignore = [
     "RUSTSEC-2025-0021",
     # openssl: Use-After-Free in `Md::fetch` and `Cipher::fetch`
     "RUSTSEC-2025-0022",
+    # Host panic with `fd_renumber` WASIp1 function
+    "RUSTSEC-2025-0046",
 ]

Cargo.lock

@@ -245,7 +245,7 @@ geozero = { version = "0.13.0", features = ["default", "with-wkb", "with-geos",
 hashbrown = { version = "0.14.3", default-features = false }
 http = "1"
 itertools = "0.10.5"
-jsonb = "0.5.2"
+jsonb = "0.5.3"
 jwt-simple = { version = "0.12.12", default-features = false, features = ["pure-rust"] }
 match-template = "0.0.1"
 mysql_async = { version = "0.34", default-features = false, features = ["native-tls-tls"] }
@@ -414,7 +414,7 @@ async-recursion = { git = "https://github.com/zhang2014/async-recursion.git", re
 backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "72265be" }
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "57795da" }
-jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "edd714b" }
+jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "6df281f" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
 # patched opendal which categories the XML dersierialization errors as recoverable
 opendal = { git = "https://github.com/datafuse-extras/opendal-for-release-v1.2.636", tag = "v0.49.0-xml-deser-retry" }

Cargo.toml

@@ -299,17 +299,15 @@ pub fn cast_scalar_to_variant(scalar: ScalarRef, tz: TzLUT, buf: &mut Vec<u8>) {
         }
         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;
         }
     };

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;
@@ -97,7 +97,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

@@ -31,8 +31,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;
@@ -401,9 +400,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

@@ -57,7 +57,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::prelude::FormatSettings;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
 use num::cast::AsPrimitive;
 use num_traits::NumCast;
@@ -469,9 +469,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

@@ -50,7 +50,7 @@ use databend_common_io::cursor_ext::ReadNumberExt;
 use databend_common_io::geography::geography_from_ewkt_bytes;
 use databend_common_io::parse_bitmap;
 use databend_common_io::parse_bytes_to_ewkb;
-use jsonb::parse_value;
+use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
 
 use crate::binary::decode_binary;
@@ -311,9 +311,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

@@ -50,7 +50,7 @@ use databend_common_io::parse_bitmap;
 use databend_common_io::parse_bytes_to_ewkb;
 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;
 
@@ -301,9 +301,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

@@ -31,6 +31,7 @@ use databend_common_expression::types::NumberType;
 use databend_common_expression::types::StringType;
 use databend_common_expression::types::UInt8Type;
 use databend_common_expression::types::ValueType;
+use databend_common_expression::types::VariantType;
 use databend_common_expression::Column;
 use databend_common_expression::EvalContext;
 use databend_common_expression::Function;
@@ -76,6 +77,15 @@ pub fn register(registry: &mut FunctionRegistry) {
         error_to_null(eval_binary_to_string),
     );
 
+    registry.register_passthrough_nullable_1_arg::<VariantType, BinaryType, _, _>(
+        "to_jsonb_binary",
+        |_, _| FunctionDomain::Full,
+        |val, _| match val {
+            ValueRef::Scalar(val) => Value::Scalar(val.to_vec()),
+            ValueRef::Column(col) => Value::Column(col),
+        },
+    );
+
     registry.register_passthrough_nullable_1_arg::<StringType, BinaryType, _, _>(
         "to_binary",
         |_, _| FunctionDomain::Full,