1use std::borrow::Cow;
16use std::sync::Arc;
17
18use arrow::array::{ArrayRef, BinaryViewArray, new_null_array};
19use arrow::compute;
20use arrow_schema::Field;
21use datafusion_common::arrow::array::{
22 Array, AsArray, BinaryViewBuilder, BooleanBuilder, Float64Builder, Int64Builder,
23 StringViewBuilder,
24};
25use datafusion_common::arrow::datatypes::DataType;
26use datafusion_common::{DataFusionError, Result, ScalarValue, exec_datafusion_err, exec_err};
27use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, Signature, Volatility};
28use datatypes::extension::json::is_json2_extension_type;
29use datatypes::vectors::json::array::JsonArray;
30use derive_more::Display;
31
32use crate::function::{Function, extract_args};
33use crate::helper;
34
35fn get_json_by_path(json: &[u8], path: &str) -> Option<Vec<u8>> {
36 let json_path = jsonb::jsonpath::parse_json_path(path.as_bytes());
37 match json_path {
38 Ok(json_path) => {
39 let mut sub_jsonb = Vec::new();
40 let mut sub_offsets = Vec::new();
41 match jsonb::get_by_path(json, json_path, &mut sub_jsonb, &mut sub_offsets) {
42 Ok(_) => Some(sub_jsonb),
43 Err(_) => None,
44 }
45 }
46 _ => None,
47 }
48}
49
50trait JsonGetResultBuilder {
51 fn append_value(&mut self, value: &[u8]) -> Result<()>;
52
53 fn append_null(&mut self);
54
55 fn build(&mut self) -> ArrayRef;
56}
57
58fn result_builder(
59 len: usize,
60 with_type: &DataType,
61 is_json2: bool,
62) -> Result<Box<dyn JsonGetResultBuilder>> {
63 let builder = match with_type {
64 DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Box::new(StringResultBuilder {
65 inner: StringViewBuilder::with_capacity(len),
66 is_json2,
67 })
68 as Box<dyn JsonGetResultBuilder>,
69 DataType::Int64 => Box::new(IntResultBuilder(Int64Builder::with_capacity(len))),
70 DataType::Float64 => Box::new(FloatResultBuilder(Float64Builder::with_capacity(len))),
71 DataType::Boolean => Box::new(BoolResultBuilder(BooleanBuilder::with_capacity(len))),
72 t => {
73 return exec_err!("json_get with unknown type {t}");
74 }
75 };
76 Ok(builder)
77}
78
79struct StringResultBuilder {
81 inner: StringViewBuilder,
82 is_json2: bool,
83}
84
85impl JsonGetResultBuilder for StringResultBuilder {
86 fn append_value(&mut self, value: &[u8]) -> Result<()> {
87 let value = if self.is_json2 && (jsonb::is_array(value) || jsonb::is_object(value)) {
90 Some(jsonb::to_string(value))
91 } else {
92 jsonb::to_str(value).ok()
93 };
94 self.inner.append_option(value);
95 Ok(())
96 }
97
98 fn append_null(&mut self) {
99 self.inner.append_null();
100 }
101
102 fn build(&mut self) -> ArrayRef {
103 Arc::new(self.inner.finish())
104 }
105}
106
107#[derive(Default, Display, Debug)]
108#[display("{}", Self::NAME.to_ascii_uppercase())]
109pub struct JsonGetString(JsonGetWithType);
110
111impl JsonGetString {
112 pub const NAME: &'static str = "json_get_string";
113}
114
115impl Function for JsonGetString {
116 fn name(&self) -> &str {
117 Self::NAME
118 }
119
120 fn return_type(&self, _: &[DataType]) -> Result<DataType> {
121 Ok(DataType::Utf8View)
122 }
123
124 fn signature(&self) -> &Signature {
125 &self.0.signature
126 }
127
128 fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result<ColumnarValue> {
129 args.args
130 .push(ColumnarValue::Scalar(ScalarValue::Utf8View(None)));
131 self.0.invoke_with_args(args)
132 }
133}
134
135struct IntResultBuilder(Int64Builder);
136
137impl JsonGetResultBuilder for IntResultBuilder {
138 fn append_value(&mut self, value: &[u8]) -> Result<()> {
139 self.0.append_option(jsonb::to_i64(value).ok());
140 Ok(())
141 }
142
143 fn append_null(&mut self) {
144 self.0.append_null();
145 }
146
147 fn build(&mut self) -> ArrayRef {
148 Arc::new(self.0.finish())
149 }
150}
151
152#[derive(Default, Display, Debug)]
153#[display("{}", Self::NAME.to_ascii_uppercase())]
154pub struct JsonGetInt(JsonGetWithType);
155
156impl JsonGetInt {
157 pub const NAME: &'static str = "json_get_int";
158}
159
160impl Function for JsonGetInt {
161 fn name(&self) -> &str {
162 Self::NAME
163 }
164
165 fn return_type(&self, _: &[DataType]) -> Result<DataType> {
166 Ok(DataType::Int64)
167 }
168
169 fn signature(&self) -> &Signature {
170 &self.0.signature
171 }
172
173 fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result<ColumnarValue> {
174 args.args
175 .push(ColumnarValue::Scalar(ScalarValue::Int64(None)));
176 self.0.invoke_with_args(args)
177 }
178}
179
180struct FloatResultBuilder(Float64Builder);
181
182impl JsonGetResultBuilder for FloatResultBuilder {
183 fn append_value(&mut self, value: &[u8]) -> Result<()> {
184 self.0.append_option(jsonb::to_f64(value).ok());
185 Ok(())
186 }
187
188 fn append_null(&mut self) {
189 self.0.append_null();
190 }
191
192 fn build(&mut self) -> ArrayRef {
193 Arc::new(self.0.finish())
194 }
195}
196
197#[derive(Default, Display, Debug)]
198#[display("{}", Self::NAME.to_ascii_uppercase())]
199pub struct JsonGetFloat(JsonGetWithType);
200
201impl JsonGetFloat {
202 pub const NAME: &'static str = "json_get_float";
203}
204
205impl Function for JsonGetFloat {
206 fn name(&self) -> &str {
207 Self::NAME
208 }
209
210 fn return_type(&self, _: &[DataType]) -> Result<DataType> {
211 Ok(DataType::Float64)
212 }
213
214 fn signature(&self) -> &Signature {
215 &self.0.signature
216 }
217
218 fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result<ColumnarValue> {
219 args.args
220 .push(ColumnarValue::Scalar(ScalarValue::Float64(None)));
221 self.0.invoke_with_args(args)
222 }
223}
224
225struct BoolResultBuilder(BooleanBuilder);
226
227impl JsonGetResultBuilder for BoolResultBuilder {
228 fn append_value(&mut self, value: &[u8]) -> Result<()> {
229 self.0.append_option(jsonb::to_bool(value).ok());
230 Ok(())
231 }
232
233 fn append_null(&mut self) {
234 self.0.append_null();
235 }
236
237 fn build(&mut self) -> ArrayRef {
238 Arc::new(self.0.finish())
239 }
240}
241
242#[derive(Default, Display, Debug)]
243#[display("{}", Self::NAME.to_ascii_uppercase())]
244pub struct JsonGetBool(JsonGetWithType);
245
246impl JsonGetBool {
247 pub const NAME: &'static str = "json_get_bool";
248}
249
250impl Function for JsonGetBool {
251 fn name(&self) -> &str {
252 Self::NAME
253 }
254
255 fn return_type(&self, _: &[DataType]) -> Result<DataType> {
256 Ok(DataType::Boolean)
257 }
258
259 fn signature(&self) -> &Signature {
260 &self.0.signature
261 }
262
263 fn invoke_with_args(&self, mut args: ScalarFunctionArgs) -> Result<ColumnarValue> {
264 args.args
265 .push(ColumnarValue::Scalar(ScalarValue::Boolean(None)));
266 self.0.invoke_with_args(args)
267 }
268}
269
270fn jsonb_get(
271 jsons: &BinaryViewArray,
272 path: &str,
273 builder: &mut dyn JsonGetResultBuilder,
274) -> Result<()> {
275 let size = jsons.len();
276 for i in 0..size {
277 let json = jsons.is_valid(i).then(|| jsons.value(i));
278 let result = match json {
279 Some(json) => get_json_by_path(json, path),
280 _ => None,
281 };
282 if let Some(v) = result {
283 builder.append_value(&v)?;
284 } else {
285 builder.append_null();
286 }
287 }
288 Ok(())
289}
290
291fn json_struct_get(array: &ArrayRef, path: &str, with_type: &DataType) -> Result<ArrayRef> {
292 let segments = path
293 .trim_start_matches("$")
294 .split('.')
295 .filter(|segment| !segment.is_empty())
296 .collect::<Vec<_>>();
297
298 let mut curr = array.clone();
299
300 for (idx, segment) in segments.iter().enumerate() {
301 if curr.data_type().is_binary() {
302 let target = nested_projection_type(&segments[idx..], with_type);
303 curr = JsonArray::from(&curr)
304 .project_to(&target)
305 .map_err(|e| exec_datafusion_err!("{e}"))?;
306 }
307
308 let Some(json) = curr.as_struct_opt() else {
309 return exec_err!("unknown JSON array datatype: {}", curr.data_type());
310 };
311 let Some(sub_json) = json.column_by_name(segment) else {
312 return Ok(new_null_array(with_type, array.len()));
313 };
314 curr = sub_json.clone();
315 }
316
317 if curr.data_type() == with_type {
318 Ok(curr)
319 } else {
320 JsonArray::from(&curr)
321 .project_to(with_type)
322 .map_err(|e| exec_datafusion_err!("{e}"))
323 }
324}
325
326fn nested_projection_type(path: &[&str], leaf_type: &DataType) -> DataType {
331 path.iter()
332 .rev()
333 .fold(leaf_type.clone(), |data_type, name| {
334 DataType::Struct(vec![Arc::new(Field::new(*name, data_type, true))].into())
335 })
336}
337
338#[derive(Debug, Display)]
342#[display("{}", Self::NAME.to_ascii_uppercase())]
343pub struct JsonGetWithType {
344 signature: Signature,
345}
346
347impl JsonGetWithType {
348 pub const NAME: &'static str = "json_get";
349}
350
351impl Default for JsonGetWithType {
352 fn default() -> Self {
353 Self {
354 signature: Signature::variadic_any(Volatility::Immutable),
355 }
356 }
357}
358
359impl Function for JsonGetWithType {
360 fn name(&self) -> &str {
361 Self::NAME
362 }
363
364 fn return_type(&self, _input_types: &[DataType]) -> datafusion_common::Result<DataType> {
365 Err(DataFusionError::Internal(
366 "This method isn't meant to be called".to_string(),
367 ))
368 }
369
370 fn return_field_from_args(
371 &self,
372 args: datafusion_expr::ReturnFieldArgs<'_>,
373 ) -> datafusion_common::Result<Arc<Field>> {
374 match args.scalar_arguments.get(2) {
375 Some(Some(v)) => {
376 let mut data_type = v.data_type();
377 if matches!(data_type, DataType::Utf8 | DataType::LargeUtf8) {
378 data_type = DataType::Utf8View;
379 }
380
381 Ok(Arc::new(Field::new(self.name(), data_type, true)))
382 }
383 _ => Ok(Arc::new(Field::new(self.name(), DataType::Utf8View, true))),
384 }
385 }
386
387 fn signature(&self) -> &Signature {
388 &self.signature
389 }
390
391 fn invoke_with_args(
392 &self,
393 args: ScalarFunctionArgs,
394 ) -> datafusion_common::Result<ColumnarValue> {
395 let args_len = args.args.len();
396 if args_len != 2 && args_len != 3 {
397 return exec_err!("json_get expects 2 or 3 arguments, got {args_len}");
398 }
399
400 let arg0 = args.args[0].to_array(args.number_rows)?;
401 let len = arg0.len();
402
403 let path = if let ColumnarValue::Scalar(path) = &args.args[1]
404 && let Some(Some(path)) = path.try_as_str()
405 {
406 path
407 } else {
408 return exec_err!(
409 r#"json_get expects a string literal "path" argument, got {}"#,
410 args.args[1]
411 );
412 };
413
414 let with_type = args
415 .args
416 .get(2)
417 .map(|x| x.data_type())
418 .unwrap_or(DataType::Utf8View);
419
420 let result = match arg0.data_type() {
421 DataType::Binary | DataType::LargeBinary | DataType::BinaryView => {
422 let arg0 = compute::cast(&arg0, &DataType::BinaryView)?;
423 let is_json2 = args.arg_fields.first().is_some_and(is_json2_extension_type);
424
425 if is_json2 && path.trim_start_matches('$').split('.').all(str::is_empty) {
426 JsonArray::from(&arg0)
427 .project_to(&with_type)
428 .map_err(|e| exec_datafusion_err!("{e:?}"))?
429 } else {
430 let jsons = arg0.as_binary_view();
431 let path = if is_json2 && !path.starts_with('$') {
432 Cow::Owned(format!("$.{path}"))
433 } else {
434 Cow::Borrowed(path)
435 };
436 let mut builder = result_builder(len, &with_type, is_json2)?;
437 jsonb_get(jsons, &path, builder.as_mut())?;
438 builder.build()
439 }
440 }
441 DataType::Struct(_) => json_struct_get(&arg0, path, &with_type)?,
442 _ => {
443 return exec_err!("JSON_GET not supported argument type {}", arg0.data_type());
444 }
445 };
446
447 Ok(ColumnarValue::Array(result))
448 }
449}
450
451#[derive(Display, Debug)]
453#[display("{}", Self::NAME.to_ascii_uppercase())]
454pub(super) struct JsonGetObject {
455 signature: Signature,
456}
457
458impl JsonGetObject {
459 const NAME: &'static str = "json_get_object";
460}
461
462impl Default for JsonGetObject {
463 fn default() -> Self {
464 Self {
465 signature: helper::one_of_sigs2(
466 vec![
467 DataType::Binary,
468 DataType::LargeBinary,
469 DataType::BinaryView,
470 ],
471 vec![DataType::UInt8, DataType::LargeUtf8, DataType::Utf8View],
472 ),
473 }
474 }
475}
476
477impl Function for JsonGetObject {
478 fn name(&self) -> &str {
479 Self::NAME
480 }
481
482 fn return_type(&self, _: &[DataType]) -> datafusion_common::Result<DataType> {
483 Ok(DataType::BinaryView)
484 }
485
486 fn signature(&self) -> &Signature {
487 &self.signature
488 }
489
490 fn invoke_with_args(
491 &self,
492 args: ScalarFunctionArgs,
493 ) -> datafusion_common::Result<ColumnarValue> {
494 let [arg0, arg1] = extract_args(self.name(), &args)?;
495 let arg0 = compute::cast(&arg0, &DataType::BinaryView)?;
496 let jsons = arg0.as_binary_view();
497 let arg1 = compute::cast(&arg1, &DataType::Utf8View)?;
498 let paths = arg1.as_string_view();
499
500 let len = jsons.len();
501 let mut builder = BinaryViewBuilder::with_capacity(len);
502
503 for i in 0..len {
504 let json = jsons.is_valid(i).then(|| jsons.value(i));
505 let path = paths.is_valid(i).then(|| paths.value(i));
506 let result = if let (Some(json), Some(path)) = (json, path) {
507 let result = jsonb::jsonpath::parse_json_path(path.as_bytes()).and_then(|path| {
508 let mut data = Vec::new();
509 let mut offset = Vec::new();
510 jsonb::get_by_path(json, path, &mut data, &mut offset)
511 .map(|()| jsonb::is_object(&data).then_some(data))
512 });
513 result.map_err(|e| DataFusionError::Execution(e.to_string()))?
514 } else {
515 None
516 };
517 builder.append_option(result);
518 }
519
520 Ok(ColumnarValue::Array(Arc::new(builder.finish())))
521 }
522}
523
524#[cfg(test)]
525mod tests {
526 use std::sync::Arc;
527
528 use arrow::array::{BooleanArray, Int64Array, StructArray};
529 use arrow_schema::{Field, Fields};
530 use datafusion_common::ScalarValue;
531 use datafusion_common::arrow::array::{BinaryArray, BinaryViewArray, StringArray};
532 use datafusion_common::arrow::datatypes::{Float64Type, Int64Type};
533 use datatypes::extension::json::Json2ExtensionType;
534 use datatypes::types::parse_string_to_jsonb;
535 use serde_json::json;
536
537 use super::*;
538
539 fn test_json_struct() -> ArrayRef {
555 let payload_fields = Fields::from(vec![
556 Field::new("code", DataType::Int64, true),
557 Field::new("success", DataType::Boolean, true),
558 Field::new("result", DataType::Binary, true),
559 ]);
560 Arc::new(StructArray::new(
561 vec![
562 Field::new("kind", DataType::Utf8, true),
563 Field::new("payload", DataType::Struct(payload_fields.clone()), true),
564 ]
565 .into(),
566 vec![
567 Arc::new(StringArray::from_iter([Some("foo")])) as ArrayRef,
568 Arc::new(StructArray::new(
569 payload_fields,
570 vec![
571 Arc::new(Int64Array::from_iter([Some(404)])) as ArrayRef,
572 Arc::new(BooleanArray::from_iter([Some(false)])),
573 Arc::new(BinaryArray::from_iter([Some(
574 json!({
575 "error": "not found",
576 "time_cost": 1.234
577 })
578 .to_string()
579 .as_bytes(),
580 )])),
581 ],
582 None,
583 )),
584 ],
585 None,
586 ))
587 }
588
589 fn test_json_field(json: &ArrayRef, is_json2: bool) -> Arc<Field> {
590 let field = Field::new("json", json.data_type().clone(), true);
591 Arc::new(if is_json2 {
592 field.with_extension_type(Json2ExtensionType::default())
593 } else {
594 field
595 })
596 }
597
598 #[test]
599 fn test_json_get_int() {
600 let json_get_int = JsonGetInt::default();
601
602 assert_eq!("json_get_int", json_get_int.name());
603 assert_eq!(
604 DataType::Int64,
605 json_get_int
606 .return_type(&[DataType::Binary, DataType::Utf8])
607 .unwrap()
608 );
609
610 let json_strings = [
611 r#"{"a": {"b": 2}, "b": 2, "c": 3}"#,
612 r#"{"a": 4, "b": {"c": 6}, "c": 6}"#,
613 r#"{"a": 7, "b": 8, "c": {"a": 7}}"#,
614 ];
615 let json_struct = test_json_struct();
616
617 let path_expects = vec![
618 ("$.a.b", Some(2)),
619 ("$.a", Some(4)),
620 ("$.c", None),
621 ("$.kind", None),
622 ("$.payload.code", Some(404)),
623 ("$.payload.success", Some(0)),
624 ("$.payload.result.time_cost", Some(1)),
625 ("$.payload.not-exists", None),
626 ("$.not-exists", None),
627 ("$", None),
628 ];
629
630 let mut jsons = json_strings
631 .iter()
632 .map(|s| {
633 let value = jsonb::parse_value(s.as_bytes()).unwrap();
634 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
635 })
636 .collect::<Vec<_>>();
637 let json_struct_arrays =
638 std::iter::repeat_n(json_struct, path_expects.len() - jsons.len()).collect::<Vec<_>>();
639 jsons.extend(json_struct_arrays);
640
641 for i in 0..jsons.len() {
642 let json = &jsons[i];
643 let (path, expect) = path_expects[i];
644
645 let args = ScalarFunctionArgs {
646 args: vec![
647 ColumnarValue::Array(json.clone()),
648 ColumnarValue::Scalar(path.into()),
649 ],
650 arg_fields: vec![],
651 number_rows: 1,
652 return_field: Arc::new(Field::new("x", DataType::Int64, false)),
653 config_options: Arc::new(Default::default()),
654 };
655 let result = json_get_int
656 .invoke_with_args(args)
657 .and_then(|x| x.to_array(1))
658 .unwrap();
659
660 let result = result.as_primitive::<Int64Type>();
661 assert_eq!(1, result.len());
662 let actual = result.is_valid(0).then(|| result.value(0));
663 assert_eq!(actual, expect);
664 }
665 }
666
667 #[test]
668 fn test_json_get_float() {
669 let json_get_float = JsonGetFloat::default();
670
671 assert_eq!("json_get_float", json_get_float.name());
672 assert_eq!(
673 DataType::Float64,
674 json_get_float
675 .return_type(&[DataType::Binary, DataType::Utf8])
676 .unwrap()
677 );
678
679 let json_strings = [
680 r#"{"a": {"b": 2.1}, "b": 2.2, "c": 3.3}"#,
681 r#"{"a": 4.4, "b": {"c": 6.6}, "c": 6.6}"#,
682 r#"{"a": 7.7, "b": 8.8, "c": {"a": 7.7}}"#,
683 ];
684 let json_struct = test_json_struct();
685
686 let path_expects = vec![
687 ("$.a.b", Some(2.1)),
688 ("$.a", Some(4.4)),
689 ("$.c", None),
690 ("$.kind", None),
691 ("$.payload.code", Some(404.0)),
692 ("$.payload.success", Some(0.0)),
693 ("$.payload.result.time_cost", Some(1.234)),
694 ("$.payload.not-exists", None),
695 ("$.not-exists", None),
696 ("$", None),
697 ];
698
699 let mut jsons = json_strings
700 .iter()
701 .map(|s| {
702 let value = jsonb::parse_value(s.as_bytes()).unwrap();
703 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
704 })
705 .collect::<Vec<_>>();
706 let json_struct_arrays =
707 std::iter::repeat_n(json_struct, path_expects.len() - jsons.len()).collect::<Vec<_>>();
708 jsons.extend(json_struct_arrays);
709
710 for i in 0..jsons.len() {
711 let json = &jsons[i];
712 let (path, expect) = path_expects[i];
713
714 let args = ScalarFunctionArgs {
715 args: vec![
716 ColumnarValue::Array(json.clone()),
717 ColumnarValue::Scalar(path.into()),
718 ],
719 arg_fields: vec![],
720 number_rows: 1,
721 return_field: Arc::new(Field::new("x", DataType::Float64, false)),
722 config_options: Arc::new(Default::default()),
723 };
724 let result = json_get_float
725 .invoke_with_args(args)
726 .and_then(|x| x.to_array(1))
727 .unwrap();
728
729 let result = result.as_primitive::<Float64Type>();
730 assert_eq!(1, result.len());
731 let actual = result.is_valid(0).then(|| result.value(0));
732 assert_eq!(actual, expect);
733 }
734 }
735
736 #[test]
737 fn test_json_get_bool() {
738 let json_get_bool = JsonGetBool::default();
739
740 assert_eq!("json_get_bool", json_get_bool.name());
741 assert_eq!(
742 DataType::Boolean,
743 json_get_bool
744 .return_type(&[DataType::Binary, DataType::Utf8])
745 .unwrap()
746 );
747
748 let json_strings = [
749 r#"{"a": {"b": true}, "b": false, "c": true}"#,
750 r#"{"a": false, "b": {"c": true}, "c": false}"#,
751 r#"{"a": true, "b": false, "c": {"a": true}}"#,
752 ];
753 let json_struct = test_json_struct();
754
755 let path_expects = vec![
756 ("$.a.b", Some(true)),
757 ("$.a", Some(false)),
758 ("$.c", None),
759 ("$.kind", None),
760 ("$.payload.code", Some(true)),
761 ("$.payload.success", Some(false)),
762 ("$.payload.result.time_cost", Some(true)),
763 ("$.payload.not-exists", None),
764 ("$.not-exists", None),
765 ("$", None),
766 ];
767
768 let mut jsons = json_strings
769 .iter()
770 .map(|s| {
771 let value = jsonb::parse_value(s.as_bytes()).unwrap();
772 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
773 })
774 .collect::<Vec<_>>();
775 let json_struct_arrays =
776 std::iter::repeat_n(json_struct, path_expects.len() - jsons.len()).collect::<Vec<_>>();
777 jsons.extend(json_struct_arrays);
778
779 for i in 0..jsons.len() {
780 let json = &jsons[i];
781 let (path, expect) = path_expects[i];
782
783 let args = ScalarFunctionArgs {
784 args: vec![
785 ColumnarValue::Array(json.clone()),
786 ColumnarValue::Scalar(path.into()),
787 ],
788 arg_fields: vec![],
789 number_rows: 1,
790 return_field: Arc::new(Field::new("x", DataType::Boolean, false)),
791 config_options: Arc::new(Default::default()),
792 };
793 let result = json_get_bool
794 .invoke_with_args(args)
795 .and_then(|x| x.to_array(1))
796 .unwrap();
797
798 let result = result.as_boolean();
799 assert_eq!(1, result.len());
800 let actual = result.is_valid(0).then(|| result.value(0));
801 assert_eq!(actual, expect);
802 }
803 }
804
805 #[test]
806 fn test_json_get_string() {
807 let json_get_string = JsonGetString::default();
808
809 assert_eq!("json_get_string", json_get_string.name());
810 assert_eq!(
811 DataType::Utf8View,
812 json_get_string
813 .return_type(&[DataType::Binary, DataType::Utf8])
814 .unwrap()
815 );
816
817 let json_strings = [
818 r#"{"a": {"b": "a"}, "b": "b", "c": "c"}"#,
819 r#"{"a": "d", "b": {"c": "e"}, "c": "f"}"#,
820 r#"{"a": "g", "b": "h", "c": {"a": "g"}}"#,
821 ];
822 let json_struct = test_json_struct();
823
824 let paths = vec![
825 "$.a.b",
826 "$.a",
827 "",
828 "$.kind",
829 "$.payload.code",
830 "$.payload.result.time_cost",
831 "$.payload",
832 "$.payload.success",
833 "$.payload.result",
834 "$.payload.result.error",
835 "$.payload.result.not-exists",
836 "$.payload.not-exists",
837 "$.not-exists",
838 "$",
839 ];
840 let expects = [
841 Some("a"),
842 Some("d"),
843 None,
844 Some("foo"),
845 Some("404"),
846 Some("1.234"),
847 Some(
848 r#"{"code":404,"result":{"error":"not found","time_cost":1.234},"success":false}"#,
849 ),
850 Some("false"),
851 Some(r#"{"error":"not found","time_cost":1.234}"#),
852 Some("not found"),
853 None,
854 None,
855 None,
856 Some(
857 r#"{"kind":"foo","payload":{"code":404,"result":{"error":"not found","time_cost":1.234},"success":false}}"#,
858 ),
859 ];
860
861 let mut jsons = json_strings
862 .iter()
863 .map(|s| {
864 let value = jsonb::parse_value(s.as_bytes()).unwrap();
865 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
866 })
867 .collect::<Vec<_>>();
868 let json_struct_arrays =
869 std::iter::repeat_n(json_struct, expects.len() - jsons.len()).collect::<Vec<_>>();
870 jsons.extend(json_struct_arrays);
871
872 for i in 0..jsons.len() {
873 let json = &jsons[i];
874 let path = paths[i];
875 let expect = expects[i];
876
877 let args = ScalarFunctionArgs {
878 args: vec![
879 ColumnarValue::Array(json.clone()),
880 ColumnarValue::Scalar(path.into()),
881 ],
882 arg_fields: vec![
883 test_json_field(json, i >= json_strings.len()),
884 Arc::new(Field::new("path", DataType::Utf8, false)),
885 ],
886 number_rows: 1,
887 return_field: Arc::new(Field::new("x", DataType::Utf8View, false)),
888 config_options: Arc::new(Default::default()),
889 };
890 let result = json_get_string
891 .invoke_with_args(args)
892 .and_then(|x| x.to_array(1))
893 .unwrap();
894
895 let result = result.as_string_view();
896 assert_eq!(1, result.len());
897 let actual = result.is_valid(0).then(|| result.value(0));
898 assert_eq!(actual, expect);
899 }
900 }
901
902 #[test]
903 fn test_json_get_object() -> Result<()> {
904 let udf = JsonGetObject::default();
905 assert_eq!("json_get_object", udf.name());
906 assert_eq!(
907 DataType::BinaryView,
908 udf.return_type(&[DataType::BinaryView, DataType::Utf8View])?
909 );
910
911 let json_value = parse_string_to_jsonb(r#"{"a": {"b": {"c": {"d": 1}}}}"#).unwrap();
912 let paths = vec!["$", "$.a", "$.a.b", "$.a.b.c", "$.a.b.c.d", "$.e", "$.a.e"];
913 let number_rows = paths.len();
914
915 let args = ScalarFunctionArgs {
916 args: vec![
917 ColumnarValue::Scalar(ScalarValue::Binary(Some(json_value))),
918 ColumnarValue::Array(Arc::new(StringArray::from_iter_values(paths))),
919 ],
920 arg_fields: vec![],
921 number_rows,
922 return_field: Arc::new(Field::new("x", DataType::Binary, false)),
923 config_options: Arc::new(Default::default()),
924 };
925 let result = udf
926 .invoke_with_args(args)
927 .and_then(|x| x.to_array(number_rows))?;
928 let result = result.as_binary_view();
929
930 let expected = &BinaryViewArray::from_iter(
931 vec![
932 Some(r#"{"a": {"b": {"c": {"d": 1}}}}"#),
933 Some(r#"{"b": {"c": {"d": 1}}}"#),
934 Some(r#"{"c": {"d": 1}}"#),
935 Some(r#"{"d": 1}"#),
936 None,
937 None,
938 None,
939 ]
940 .into_iter()
941 .map(|x| x.and_then(|s| parse_string_to_jsonb(s).ok())),
942 );
943 assert_eq!(result, expected);
944 Ok(())
945 }
946
947 #[test]
948 fn test_json_get_with_type() {
949 let json_get_with_type = JsonGetWithType::default();
950
951 assert_eq!("json_get", json_get_with_type.name());
952
953 let json_strings = [
954 r#"{"a": {"b": "a"}, "b": "b", "c": "c"}"#,
955 r#"{"a": "d", "b": {"c": "e"}, "c": "f"}"#,
956 r#"{"a": "g", "b": "h", "c": {"a": "g"}}"#,
957 ];
958 let json_struct = test_json_struct();
959
960 let paths = vec![
961 "$.a.b",
962 "$.a",
963 "",
964 "$.kind",
965 "$.payload.code",
966 "$.payload.result.time_cost",
967 "$.payload",
968 "$.payload.success",
969 "$.payload.result",
970 "$.payload.result.error",
971 "$.payload.result.not-exists",
972 "$.payload.not-exists",
973 "$.not-exists",
974 "$",
975 ];
976 let expects = [
977 Some("a"),
978 Some("d"),
979 None,
980 Some("foo"),
981 Some("404"),
982 Some("1.234"),
983 Some(
984 r#"{"code":404,"result":{"error":"not found","time_cost":1.234},"success":false}"#,
985 ),
986 Some("false"),
987 Some(r#"{"error":"not found","time_cost":1.234}"#),
988 Some("not found"),
989 None,
990 None,
991 None,
992 Some(
993 r#"{"kind":"foo","payload":{"code":404,"result":{"error":"not found","time_cost":1.234},"success":false}}"#,
994 ),
995 ];
996
997 let mut jsons = json_strings
998 .iter()
999 .map(|s| {
1000 let value = jsonb::parse_value(s.as_bytes()).unwrap();
1001 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
1002 })
1003 .collect::<Vec<_>>();
1004 let json_struct_arrays =
1005 std::iter::repeat_n(json_struct, expects.len() - jsons.len()).collect::<Vec<_>>();
1006 jsons.extend(json_struct_arrays);
1007
1008 for i in 0..jsons.len() {
1009 let json = &jsons[i];
1010 let path = paths[i];
1011 let expect = expects[i];
1012
1013 let args = ScalarFunctionArgs {
1014 args: vec![
1015 ColumnarValue::Array(json.clone()),
1016 ColumnarValue::Scalar(path.into()),
1017 ColumnarValue::Scalar(ScalarValue::Utf8View(None)),
1018 ],
1019 arg_fields: vec![
1020 test_json_field(json, i >= json_strings.len()),
1021 Arc::new(Field::new("path", DataType::Utf8, false)),
1022 Arc::new(Field::new("with_type", DataType::Utf8View, true)),
1023 ],
1024 number_rows: 1,
1025 return_field: Arc::new(Field::new("x", DataType::Utf8View, false)),
1026 config_options: Arc::new(Default::default()),
1027 };
1028 let result = json_get_with_type
1029 .invoke_with_args(args)
1030 .and_then(|x| x.to_array(1))
1031 .unwrap();
1032
1033 let result = result.as_string_view();
1034 assert_eq!(1, result.len());
1035 let actual = result.is_valid(0).then(|| result.value(0));
1036 assert_eq!(actual, expect);
1037 }
1038
1039 let json_strings = [
1040 r#"{"a": {"b": 2}, "b": 2, "c": 3}"#,
1041 r#"{"a": 4, "b": {"c": 6}, "c": 6}"#,
1042 r#"{"a": 7, "b": 8, "c": {"a": 7}}"#,
1043 ];
1044 let paths = ["$.a.b", "$.a", "$.c", "$.payload.code"];
1045 let expects = [Some(2), Some(4), None, Some(404)];
1046
1047 for (i, (path, expect)) in paths.iter().zip(expects.iter()).enumerate() {
1048 let json = if i < json_strings.len() {
1049 let value = jsonb::parse_value(json_strings[i].as_bytes()).unwrap();
1050 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
1051 } else {
1052 test_json_struct()
1053 };
1054
1055 let args = ScalarFunctionArgs {
1056 args: vec![
1057 ColumnarValue::Array(json),
1058 ColumnarValue::Scalar((*path).into()),
1059 ColumnarValue::Scalar(ScalarValue::Int64(None)),
1060 ],
1061 arg_fields: vec![],
1062 number_rows: 1,
1063 return_field: Arc::new(Field::new("x", DataType::Int64, false)),
1064 config_options: Arc::new(Default::default()),
1065 };
1066 let result = json_get_with_type
1067 .invoke_with_args(args)
1068 .and_then(|x| x.to_array(1))
1069 .unwrap();
1070
1071 let result = result.as_primitive::<Int64Type>();
1072 assert_eq!(1, result.len());
1073 let actual = result.is_valid(0).then(|| result.value(0));
1074 assert_eq!(actual, *expect);
1075 }
1076
1077 let json_strings = [
1078 r#"{"a": {"b": 2.1}, "b": 2.2, "c": 3.3}"#,
1079 r#"{"a": 4.4, "b": {"c": 6.6}, "c": 6.6}"#,
1080 r#"{"a": 7.7, "b": 8.8, "c": {"a": 7.7}}"#,
1081 ];
1082 let paths = ["$.a.b", "$.a", "$.c", "$.payload.result.time_cost"];
1083 let expects = [Some(2.1), Some(4.4), None, Some(1.234)];
1084
1085 for (i, (path, expect)) in paths.iter().zip(expects.iter()).enumerate() {
1086 let json = if i < json_strings.len() {
1087 let value = jsonb::parse_value(json_strings[i].as_bytes()).unwrap();
1088 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
1089 } else {
1090 test_json_struct()
1091 };
1092
1093 let args = ScalarFunctionArgs {
1094 args: vec![
1095 ColumnarValue::Array(json),
1096 ColumnarValue::Scalar((*path).into()),
1097 ColumnarValue::Scalar(ScalarValue::Float64(None)),
1098 ],
1099 arg_fields: vec![],
1100 number_rows: 1,
1101 return_field: Arc::new(Field::new("x", DataType::Float64, false)),
1102 config_options: Arc::new(Default::default()),
1103 };
1104 let result = json_get_with_type
1105 .invoke_with_args(args)
1106 .and_then(|x| x.to_array(1))
1107 .unwrap();
1108
1109 let result = result.as_primitive::<Float64Type>();
1110 assert_eq!(1, result.len());
1111 let actual = result.is_valid(0).then(|| result.value(0));
1112 assert_eq!(actual, *expect);
1113 }
1114
1115 let json_strings = [
1116 r#"{"a": {"b": true}, "b": false, "c": true}"#,
1117 r#"{"a": false, "b": {"c": true}, "c": false}"#,
1118 r#"{"a": true, "b": false, "c": {"a": true}}"#,
1119 ];
1120 let paths = ["$.a.b", "$.a", "$.c", "$.payload.success"];
1121 let expects = [Some(true), Some(false), None, Some(false)];
1122
1123 for (i, (path, expect)) in paths.iter().zip(expects.iter()).enumerate() {
1124 let json = if i < json_strings.len() {
1125 let value = jsonb::parse_value(json_strings[i].as_bytes()).unwrap();
1126 Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef
1127 } else {
1128 test_json_struct()
1129 };
1130
1131 let args = ScalarFunctionArgs {
1132 args: vec![
1133 ColumnarValue::Array(json),
1134 ColumnarValue::Scalar((*path).into()),
1135 ColumnarValue::Scalar(ScalarValue::Boolean(None)),
1136 ],
1137 arg_fields: vec![],
1138 number_rows: 1,
1139 return_field: Arc::new(Field::new("x", DataType::Boolean, false)),
1140 config_options: Arc::new(Default::default()),
1141 };
1142 let result = json_get_with_type
1143 .invoke_with_args(args)
1144 .and_then(|x| x.to_array(1))
1145 .unwrap();
1146
1147 let result = result.as_boolean();
1148 assert_eq!(1, result.len());
1149 let actual = result.is_valid(0).then(|| result.value(0));
1150 assert_eq!(actual, *expect);
1151 }
1152 }
1153}