datafusion_functions_json/
json_length.rs1use std::sync::Arc;
2
3use datafusion::arrow::array::{ArrayRef, UInt64Array, UInt64Builder};
4use datafusion::arrow::datatypes::DataType;
5use datafusion::common::{Result as DataFusionResult, ScalarValue};
6use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility};
7use jiter::Peek;
8
9use crate::common::{get_err, invoke, jiter_json_find, return_type_check, GetError, InvokeResult, JsonPath};
10use crate::common_macros::make_udf_function;
11
12make_udf_function!(
13 JsonLength,
14 json_length,
15 json_data path,
16 r"Get the length of the array or object at the given path."
17);
18
19#[derive(Debug, PartialEq, Eq, Hash)]
20pub(super) struct JsonLength {
21 signature: Signature,
22 aliases: [String; 2],
23}
24
25impl Default for JsonLength {
26 fn default() -> Self {
27 Self {
28 signature: Signature::variadic_any(Volatility::Immutable),
29 aliases: ["json_length".to_string(), "json_len".to_string()],
30 }
31 }
32}
33
34impl ScalarUDFImpl for JsonLength {
35 fn name(&self) -> &str {
36 self.aliases[0].as_str()
37 }
38
39 fn signature(&self) -> &Signature {
40 &self.signature
41 }
42
43 fn return_type(&self, arg_types: &[DataType]) -> DataFusionResult<DataType> {
44 return_type_check(arg_types, self.name(), DataType::UInt64)
45 }
46
47 fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DataFusionResult<ColumnarValue> {
48 invoke::<UInt64Array>(&args.args, jiter_json_length)
49 }
50
51 fn aliases(&self) -> &[String] {
52 &self.aliases
53 }
54
55 fn placement(
56 &self,
57 args: &[datafusion::logical_expr::ExpressionPlacement],
58 ) -> datafusion::logical_expr::ExpressionPlacement {
59 if args.len() >= 2
62 && matches!(args[0], datafusion::logical_expr::ExpressionPlacement::Column)
63 && args[1..]
64 .iter()
65 .all(|arg| matches!(arg, datafusion::logical_expr::ExpressionPlacement::Literal))
66 {
67 datafusion::logical_expr::ExpressionPlacement::MoveTowardsLeafNodes
68 } else {
69 datafusion::logical_expr::ExpressionPlacement::KeepInPlace
70 }
71 }
72}
73
74impl InvokeResult for UInt64Array {
75 type Item = u64;
76
77 type Builder = UInt64Builder;
78
79 const ACCEPT_DICT_RETURN: bool = false;
81
82 fn builder(capacity: usize) -> Self::Builder {
83 UInt64Builder::with_capacity(capacity)
84 }
85
86 fn append_value(builder: &mut Self::Builder, value: Option<Self::Item>) {
87 builder.append_option(value);
88 }
89
90 fn finish(mut builder: Self::Builder) -> DataFusionResult<ArrayRef> {
91 Ok(Arc::new(builder.finish()))
92 }
93
94 fn scalar(value: Option<Self::Item>) -> ScalarValue {
95 ScalarValue::UInt64(value)
96 }
97}
98
99fn jiter_json_length(opt_json: Option<&str>, path: &[JsonPath]) -> Result<u64, GetError> {
100 if let Some((mut jiter, peek)) = jiter_json_find(opt_json, path) {
101 match peek {
102 Peek::Array => {
103 let mut peek_opt = jiter.known_array()?;
104 let mut length: u64 = 0;
105 while let Some(peek) = peek_opt {
106 jiter.known_skip(peek)?;
107 length += 1;
108 peek_opt = jiter.array_step()?;
109 }
110 Ok(length)
111 }
112 Peek::Object => {
113 let mut opt_key = jiter.known_object()?;
114
115 let mut length: u64 = 0;
116 while opt_key.is_some() {
117 jiter.next_skip()?;
118 length += 1;
119 opt_key = jiter.next_key()?;
120 }
121 Ok(length)
122 }
123 _ => get_err!(),
124 }
125 } else {
126 get_err!()
127 }
128}