forked from datafusion-contrib/datafusion-functions-json
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathjson_object_keys.rs
More file actions
146 lines (123 loc) · 4.33 KB
/
json_object_keys.rs
File metadata and controls
146 lines (123 loc) · 4.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
use std::any::Any;
use std::sync::Arc;
use datafusion::arrow::array::{ArrayRef, ListBuilder, StringBuilder};
use datafusion::arrow::datatypes::{DataType, Field};
use datafusion::common::{Result as DataFusionResult, ScalarValue};
use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility};
use jiter::Peek;
use crate::common::{get_err, invoke, jiter_json_find, return_type_check, GetError, InvokeResult, JsonPath};
use crate::common_macros::make_udf_function;
make_udf_function!(
JsonObjectKeys,
json_object_keys,
json_data path,
r"Get the keys of a JSON object as an array."
);
#[derive(Debug, PartialEq, Eq, Hash)]
pub(super) struct JsonObjectKeys {
signature: Signature,
aliases: [String; 2],
}
impl Default for JsonObjectKeys {
fn default() -> Self {
Self {
signature: Signature::variadic_any(Volatility::Immutable),
aliases: ["json_object_keys".to_string(), "json_keys".to_string()],
}
}
}
impl ScalarUDFImpl for JsonObjectKeys {
fn as_any(&self) -> &dyn Any {
self
}
fn name(&self) -> &str {
self.aliases[0].as_str()
}
fn signature(&self) -> &Signature {
&self.signature
}
fn return_type(&self, arg_types: &[DataType]) -> DataFusionResult<DataType> {
return_type_check(
arg_types,
self.name(),
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
)
}
fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DataFusionResult<ColumnarValue> {
invoke::<BuildListArray>(&args.args, jiter_json_object_keys)
}
fn aliases(&self) -> &[String] {
&self.aliases
}
fn placement(
&self,
args: &[datafusion::logical_expr::ExpressionPlacement],
) -> datafusion::logical_expr::ExpressionPlacement {
// If the first argument is a column and the remaining arguments are literals (a path)
// then we can push this UDF down to the leaf nodes.
if args.len() >= 2
&& matches!(args[0], datafusion::logical_expr::ExpressionPlacement::Column)
&& args[1..]
.iter()
.all(|arg| matches!(arg, datafusion::logical_expr::ExpressionPlacement::Literal))
{
datafusion::logical_expr::ExpressionPlacement::MoveTowardsLeafNodes
} else {
datafusion::logical_expr::ExpressionPlacement::KeepInPlace
}
}
}
/// Struct used to build a `ListArray` from the result of `jiter_json_object_keys`.
#[derive(Debug)]
struct BuildListArray;
impl InvokeResult for BuildListArray {
type Item = Vec<String>;
type Builder = ListBuilder<StringBuilder>;
const ACCEPT_DICT_RETURN: bool = true;
fn builder(capacity: usize) -> Self::Builder {
let values_builder = StringBuilder::new();
ListBuilder::with_capacity(values_builder, capacity)
}
fn append_value(builder: &mut Self::Builder, value: Option<Self::Item>) {
builder.append_option(value.map(|v| v.into_iter().map(Some)));
}
fn finish(mut builder: Self::Builder) -> DataFusionResult<ArrayRef> {
Ok(Arc::new(builder.finish()))
}
fn scalar(value: Option<Self::Item>) -> ScalarValue {
keys_to_scalar(value)
}
}
fn keys_to_scalar(opt_keys: Option<Vec<String>>) -> ScalarValue {
let values_builder = StringBuilder::new();
let mut builder = ListBuilder::new(values_builder);
if let Some(keys) = opt_keys {
for value in keys {
builder.values().append_value(value);
}
builder.append(true);
} else {
builder.append(false);
}
let array = builder.finish();
ScalarValue::List(Arc::new(array))
}
fn jiter_json_object_keys(opt_json: Option<&str>, path: &[JsonPath]) -> Result<Vec<String>, GetError> {
if let Some((mut jiter, peek)) = jiter_json_find(opt_json, path) {
match peek {
Peek::Object => {
let mut opt_key = jiter.known_object()?;
let mut keys = Vec::new();
while let Some(key) = opt_key {
keys.push(key.to_string());
jiter.next_skip()?;
opt_key = jiter.next_key()?;
}
Ok(keys)
}
_ => get_err!(),
}
} else {
get_err!()
}
}