pipeline/etl/
value.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15pub mod array;
16pub mod map;
17pub mod time;
18
19use std::result::Result as StdResult;
20
21pub use array::Array;
22use jsonb::{Number as JsonbNumber, Object as JsonbObject, Value as JsonbValue};
23use jsonpath_rust::parser::{parse_json_path, JsonPathIndex};
24use jsonpath_rust::path::{JsonLike, Path};
25use jsonpath_rust::{jsp_idx, jsp_obj, JsonPath, JsonPathParserError, JsonPathStr};
26pub use map::Map;
27use regex::Regex;
28use snafu::{OptionExt, ResultExt};
29pub use time::Timestamp;
30
31use crate::error::{
32    Error, Result, UnsupportedNumberTypeSnafu, ValueDefaultValueUnsupportedSnafu,
33    ValueInvalidResolutionSnafu, ValueParseBooleanSnafu, ValueParseFloatSnafu, ValueParseIntSnafu,
34    ValueParseTypeSnafu, ValueUnsupportedYamlTypeSnafu, ValueYamlKeyMustBeStringSnafu,
35};
36use crate::etl::PipelineMap;
37
38/// Value can be used as type
39/// acts as value: the enclosed value is the actual value
40/// acts as type: the enclosed value is the default value
41
42#[derive(Debug, Clone, PartialEq, Default)]
43pub enum Value {
44    // as value: null
45    // as type: no type specified
46    #[default]
47    Null,
48
49    Int8(i8),
50    Int16(i16),
51    Int32(i32),
52    Int64(i64),
53
54    Uint8(u8),
55    Uint16(u16),
56    Uint32(u32),
57    Uint64(u64),
58
59    Float32(f32),
60    Float64(f64),
61
62    Boolean(bool),
63    String(String),
64
65    Timestamp(Timestamp),
66
67    /// We only consider object and array to be json types.
68    Array(Array),
69    Map(Map),
70}
71
72impl Value {
73    pub fn is_null(&self) -> bool {
74        matches!(self, Value::Null)
75    }
76
77    pub fn parse_str_type(t: &str) -> Result<Self> {
78        let mut parts = t.splitn(2, ',');
79        let head = parts.next().unwrap_or_default();
80        let tail = parts.next().map(|s| s.trim().to_string());
81        match head.to_lowercase().as_str() {
82            "int8" => Ok(Value::Int8(0)),
83            "int16" => Ok(Value::Int16(0)),
84            "int32" => Ok(Value::Int32(0)),
85            "int64" => Ok(Value::Int64(0)),
86
87            "uint8" => Ok(Value::Uint8(0)),
88            "uint16" => Ok(Value::Uint16(0)),
89            "uint32" => Ok(Value::Uint32(0)),
90            "uint64" => Ok(Value::Uint64(0)),
91
92            "float32" => Ok(Value::Float32(0.0)),
93            "float64" => Ok(Value::Float64(0.0)),
94
95            "boolean" => Ok(Value::Boolean(false)),
96            "string" => Ok(Value::String("".to_string())),
97
98            "timestamp" | "epoch" | "time" => match tail {
99                Some(resolution) if !resolution.is_empty() => match resolution.as_str() {
100                    time::NANOSECOND_RESOLUTION | time::NANO_RESOLUTION | time::NS_RESOLUTION => {
101                        Ok(Value::Timestamp(Timestamp::Nanosecond(0)))
102                    }
103                    time::MICROSECOND_RESOLUTION | time::MICRO_RESOLUTION | time::US_RESOLUTION => {
104                        Ok(Value::Timestamp(Timestamp::Microsecond(0)))
105                    }
106                    time::MILLISECOND_RESOLUTION | time::MILLI_RESOLUTION | time::MS_RESOLUTION => {
107                        Ok(Value::Timestamp(Timestamp::Millisecond(0)))
108                    }
109                    time::SECOND_RESOLUTION | time::SEC_RESOLUTION | time::S_RESOLUTION => {
110                        Ok(Value::Timestamp(Timestamp::Second(0)))
111                    }
112                    _ => ValueInvalidResolutionSnafu {
113                        resolution,
114                        valid_resolution: time::VALID_RESOLUTIONS.join(","),
115                    }
116                    .fail(),
117                },
118                _ => Ok(Value::Timestamp(Timestamp::Nanosecond(0))),
119            },
120
121            // We only consider object and array to be json types. and use Map to represent json
122            // TODO(qtang): Needs to be defined with better semantics
123            "json" => Ok(Value::Map(Map::default())),
124
125            _ => ValueParseTypeSnafu { t }.fail(),
126        }
127    }
128
129    /// only support string, bool, number, null
130    pub fn parse_str_value(&self, v: &str) -> Result<Self> {
131        match self {
132            Value::Int8(_) => v
133                .parse::<i8>()
134                .map(Value::Int8)
135                .context(ValueParseIntSnafu { ty: "int8", v }),
136            Value::Int16(_) => v
137                .parse::<i16>()
138                .map(Value::Int16)
139                .context(ValueParseIntSnafu { ty: "int16", v }),
140            Value::Int32(_) => v
141                .parse::<i32>()
142                .map(Value::Int32)
143                .context(ValueParseIntSnafu { ty: "int32", v }),
144            Value::Int64(_) => v
145                .parse::<i64>()
146                .map(Value::Int64)
147                .context(ValueParseIntSnafu { ty: "int64", v }),
148
149            Value::Uint8(_) => v
150                .parse::<u8>()
151                .map(Value::Uint8)
152                .context(ValueParseIntSnafu { ty: "uint8", v }),
153            Value::Uint16(_) => v
154                .parse::<u16>()
155                .map(Value::Uint16)
156                .context(ValueParseIntSnafu { ty: "uint16", v }),
157            Value::Uint32(_) => v
158                .parse::<u32>()
159                .map(Value::Uint32)
160                .context(ValueParseIntSnafu { ty: "uint32", v }),
161            Value::Uint64(_) => v
162                .parse::<u64>()
163                .map(Value::Uint64)
164                .context(ValueParseIntSnafu { ty: "uint64", v }),
165
166            Value::Float32(_) => v
167                .parse::<f32>()
168                .map(Value::Float32)
169                .context(ValueParseFloatSnafu { ty: "float32", v }),
170            Value::Float64(_) => v
171                .parse::<f64>()
172                .map(Value::Float64)
173                .context(ValueParseFloatSnafu { ty: "float64", v }),
174
175            Value::Boolean(_) => v
176                .parse::<bool>()
177                .map(Value::Boolean)
178                .context(ValueParseBooleanSnafu { ty: "boolean", v }),
179            Value::String(_) => Ok(Value::String(v.to_string())),
180
181            Value::Null => Ok(Value::Null),
182
183            _ => ValueDefaultValueUnsupportedSnafu {
184                value: format!("{:?}", self),
185            }
186            .fail(),
187        }
188    }
189
190    /// only support string, bool, number, null
191    pub fn to_str_value(&self) -> String {
192        match self {
193            Value::Int8(v) => format!("{}", v),
194            Value::Int16(v) => format!("{}", v),
195            Value::Int32(v) => format!("{}", v),
196            Value::Int64(v) => format!("{}", v),
197
198            Value::Uint8(v) => format!("{}", v),
199            Value::Uint16(v) => format!("{}", v),
200            Value::Uint32(v) => format!("{}", v),
201            Value::Uint64(v) => format!("{}", v),
202
203            Value::Float32(v) => format!("{}", v),
204            Value::Float64(v) => format!("{}", v),
205
206            Value::Boolean(v) => format!("{}", v),
207            Value::String(v) => v.to_string(),
208
209            v => v.to_string(),
210        }
211    }
212
213    pub fn to_str_type(&self) -> &str {
214        match self {
215            Value::Int8(_) => "int8",
216            Value::Int16(_) => "int16",
217            Value::Int32(_) => "int32",
218            Value::Int64(_) => "int64",
219
220            Value::Uint8(_) => "uint8",
221            Value::Uint16(_) => "uint16",
222            Value::Uint32(_) => "uint32",
223            Value::Uint64(_) => "uint64",
224
225            Value::Float32(_) => "float32",
226            Value::Float64(_) => "float64",
227
228            Value::Boolean(_) => "boolean",
229            Value::String(_) => "string",
230
231            Value::Timestamp(_) => "epoch",
232
233            Value::Array(_) | Value::Map(_) => "json",
234
235            Value::Null => "null",
236        }
237    }
238
239    pub fn get(&self, key: &str) -> Option<&Self> {
240        match self {
241            Value::Map(map) => map.get(key),
242            _ => None,
243        }
244    }
245
246    pub fn as_str(&self) -> Option<&str> {
247        match self {
248            Value::String(v) => Some(v),
249            _ => None,
250        }
251    }
252
253    pub fn as_i64(&self) -> Option<i64> {
254        match self {
255            Value::Uint32(v) => Some(*v as i64),
256            Value::Uint16(v) => Some(*v as i64),
257            Value::Uint8(v) => Some(*v as i64),
258            Value::Int64(v) => Some(*v),
259            Value::Int32(v) => Some(*v as i64),
260            Value::Int16(v) => Some(*v as i64),
261            Value::Int8(v) => Some(*v as i64),
262            _ => None,
263        }
264    }
265
266    pub fn as_u64(&self) -> Option<u64> {
267        match self {
268            Value::Uint64(v) => Some(*v),
269            Value::Uint32(v) => Some(*v as u64),
270            Value::Uint16(v) => Some(*v as u64),
271            Value::Uint8(v) => Some(*v as u64),
272            _ => None,
273        }
274    }
275
276    pub fn as_f64(&self) -> Option<f64> {
277        match self {
278            Value::Float32(v) => Some(*v as f64),
279            Value::Float64(v) => Some(*v),
280            Value::Uint64(v) => Some(*v as f64),
281            Value::Uint32(v) => Some(*v as f64),
282            Value::Uint16(v) => Some(*v as f64),
283            Value::Uint8(v) => Some(*v as f64),
284            Value::Int64(v) => Some(*v as f64),
285            Value::Int32(v) => Some(*v as f64),
286            Value::Int16(v) => Some(*v as f64),
287            Value::Int8(v) => Some(*v as f64),
288            _ => None,
289        }
290    }
291
292    // ref https://github.com/serde-rs/json/blob/master/src/value/mod.rs#L779
293    pub fn pointer(&self, pointer: &str) -> Option<&Value> {
294        if pointer.is_empty() {
295            return Some(self);
296        }
297        if !pointer.starts_with('/') {
298            return None;
299        }
300        pointer
301            .split('/')
302            .skip(1)
303            .map(|x| x.replace("~1", "/").replace("~0", "~"))
304            .try_fold(self, |target, token| match target {
305                Value::Map(map) => map.get(&token),
306                Value::Array(list) => parse_index(&token).and_then(|x| list.get(x)),
307                _ => None,
308            })
309    }
310
311    // ref https://github.com/serde-rs/json/blob/master/src/value/mod.rs#L834
312    pub fn pointer_mut(&mut self, pointer: &str) -> Option<&mut Value> {
313        if pointer.is_empty() {
314            return Some(self);
315        }
316        if !pointer.starts_with('/') {
317            return None;
318        }
319        pointer
320            .split('/')
321            .skip(1)
322            .map(|x| x.replace("~1", "/").replace("~0", "~"))
323            .try_fold(self, |target, token| match target {
324                Value::Map(map) => map.get_mut(&token),
325                Value::Array(list) => parse_index(&token).and_then(move |x| list.get_mut(x)),
326                _ => None,
327            })
328    }
329}
330
331// ref https://github.com/serde-rs/json/blob/master/src/value/mod.rs#L259
332fn parse_index(s: &str) -> Option<usize> {
333    if s.starts_with('+') || (s.starts_with('0') && s.len() != 1) {
334        return None;
335    }
336    s.parse().ok()
337}
338
339impl std::fmt::Display for Value {
340    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
341        let str = match self {
342            Value::Null => "null".to_string(),
343
344            Value::Int8(v) => format!("int8({})", v),
345            Value::Int16(v) => format!("int16({})", v),
346            Value::Int32(v) => format!("int32({})", v),
347            Value::Int64(v) => format!("int64({})", v),
348
349            Value::Uint8(v) => format!("uint8({})", v),
350            Value::Uint16(v) => format!("uint16({})", v),
351            Value::Uint32(v) => format!("uint32({})", v),
352            Value::Uint64(v) => format!("uint64({})", v),
353
354            Value::Float32(v) => format!("float32({})", v),
355            Value::Float64(v) => format!("float64({})", v),
356
357            Value::Boolean(v) => format!("boolean({})", v),
358            Value::String(v) => format!("string({})", v),
359
360            Value::Timestamp(v) => format!("epoch({})", v),
361
362            Value::Array(v) => format!("{}", v),
363            Value::Map(v) => format!("{}", v),
364        };
365
366        write!(f, "{}", str)
367    }
368}
369
370impl TryFrom<simd_json::value::OwnedValue> for Value {
371    type Error = Error;
372
373    fn try_from(v: simd_json::value::OwnedValue) -> Result<Self> {
374        match v {
375            simd_json::value::OwnedValue::Static(v) => match v {
376                simd_json::value::StaticNode::Null => Ok(Value::Null),
377                simd_json::value::StaticNode::Bool(v) => Ok(Value::Boolean(v)),
378                simd_json::value::StaticNode::I64(v) => Ok(Value::Int64(v)),
379                simd_json::value::StaticNode::U64(v) => Ok(Value::Uint64(v)),
380                simd_json::value::StaticNode::F64(v) => Ok(Value::Float64(v)),
381            },
382            simd_json::OwnedValue::String(s) => Ok(Value::String(s)),
383            simd_json::OwnedValue::Array(values) => {
384                let mut re = Vec::with_capacity(values.len());
385                for v in values.into_iter() {
386                    re.push(Value::try_from(v)?);
387                }
388                Ok(Value::Array(Array { values: re }))
389            }
390            simd_json::OwnedValue::Object(map) => {
391                let mut values = PipelineMap::new();
392                for (k, v) in map.into_iter() {
393                    values.insert(k, Value::try_from(v)?);
394                }
395                Ok(Value::Map(Map { values }))
396            }
397        }
398    }
399}
400
401impl TryFrom<serde_json::Value> for Value {
402    type Error = Error;
403
404    fn try_from(v: serde_json::Value) -> Result<Self> {
405        match v {
406            serde_json::Value::Null => Ok(Value::Null),
407            serde_json::Value::Bool(v) => Ok(Value::Boolean(v)),
408            serde_json::Value::Number(v) => {
409                if let Some(v) = v.as_i64() {
410                    Ok(Value::Int64(v))
411                } else if let Some(v) = v.as_u64() {
412                    Ok(Value::Uint64(v))
413                } else if let Some(v) = v.as_f64() {
414                    Ok(Value::Float64(v))
415                } else {
416                    UnsupportedNumberTypeSnafu { value: v }.fail()
417                }
418            }
419            serde_json::Value::String(v) => Ok(Value::String(v)),
420            serde_json::Value::Array(v) => {
421                let mut values = Vec::with_capacity(v.len());
422                for v in v {
423                    values.push(Value::try_from(v)?);
424                }
425                Ok(Value::Array(Array { values }))
426            }
427            serde_json::Value::Object(v) => {
428                let mut values = PipelineMap::new();
429                for (k, v) in v {
430                    values.insert(k, Value::try_from(v)?);
431                }
432                Ok(Value::Map(Map { values }))
433            }
434        }
435    }
436}
437
438impl TryFrom<&yaml_rust::Yaml> for Value {
439    type Error = Error;
440
441    fn try_from(v: &yaml_rust::Yaml) -> Result<Self> {
442        match v {
443            yaml_rust::Yaml::Null => Ok(Value::Null),
444            yaml_rust::Yaml::Boolean(v) => Ok(Value::Boolean(*v)),
445            yaml_rust::Yaml::Integer(v) => Ok(Value::Int64(*v)),
446            yaml_rust::Yaml::Real(v) => match v.parse::<f64>() {
447                Ok(v) => Ok(Value::Float64(v)),
448                Err(e) => Err(e).context(ValueParseFloatSnafu { ty: "float64", v }),
449            },
450            yaml_rust::Yaml::String(v) => Ok(Value::String(v.to_string())),
451            yaml_rust::Yaml::Array(arr) => {
452                let mut values = vec![];
453                for v in arr {
454                    values.push(Value::try_from(v)?);
455                }
456                Ok(Value::Array(Array { values }))
457            }
458            yaml_rust::Yaml::Hash(v) => {
459                let mut values = PipelineMap::new();
460                for (k, v) in v {
461                    let key = k
462                        .as_str()
463                        .with_context(|| ValueYamlKeyMustBeStringSnafu { value: v.clone() })?;
464                    values.insert(key.to_string(), Value::try_from(v)?);
465                }
466                Ok(Value::Map(Map { values }))
467            }
468            _ => ValueUnsupportedYamlTypeSnafu { value: v.clone() }.fail(),
469        }
470    }
471}
472
473impl From<&Value> for JsonbValue<'_> {
474    fn from(value: &Value) -> Self {
475        match value {
476            Value::Null => JsonbValue::Null,
477            Value::Boolean(v) => JsonbValue::Bool(*v),
478
479            Value::Int8(v) => JsonbValue::Number(JsonbNumber::Int64(*v as i64)),
480            Value::Int16(v) => JsonbValue::Number(JsonbNumber::Int64(*v as i64)),
481            Value::Int32(v) => JsonbValue::Number(JsonbNumber::Int64(*v as i64)),
482            Value::Int64(v) => JsonbValue::Number(JsonbNumber::Int64(*v)),
483
484            Value::Uint8(v) => JsonbValue::Number(JsonbNumber::UInt64(*v as u64)),
485            Value::Uint16(v) => JsonbValue::Number(JsonbNumber::UInt64(*v as u64)),
486            Value::Uint32(v) => JsonbValue::Number(JsonbNumber::UInt64(*v as u64)),
487            Value::Uint64(v) => JsonbValue::Number(JsonbNumber::UInt64(*v)),
488            Value::Float32(v) => JsonbValue::Number(JsonbNumber::Float64(*v as f64)),
489            Value::Float64(v) => JsonbValue::Number(JsonbNumber::Float64(*v)),
490            Value::String(v) => JsonbValue::String(v.clone().into()),
491            Value::Timestamp(v) => JsonbValue::String(v.to_string().into()),
492            Value::Array(arr) => {
493                let mut vals: Vec<JsonbValue> = Vec::with_capacity(arr.len());
494                for val in arr.iter() {
495                    vals.push(val.into());
496                }
497                JsonbValue::Array(vals)
498            }
499            Value::Map(obj) => {
500                let mut map = JsonbObject::new();
501                for (k, v) in obj.iter() {
502                    let val: JsonbValue = v.into();
503                    map.insert(k.to_string(), val);
504                }
505                JsonbValue::Object(map)
506            }
507        }
508    }
509}
510
511impl From<Value> for JsonbValue<'_> {
512    fn from(value: Value) -> Self {
513        match value {
514            Value::Null => JsonbValue::Null,
515            Value::Boolean(v) => JsonbValue::Bool(v),
516
517            Value::Int8(v) => JsonbValue::Number(JsonbNumber::Int64(v as i64)),
518            Value::Int16(v) => JsonbValue::Number(JsonbNumber::Int64(v as i64)),
519            Value::Int32(v) => JsonbValue::Number(JsonbNumber::Int64(v as i64)),
520            Value::Int64(v) => JsonbValue::Number(JsonbNumber::Int64(v)),
521
522            Value::Uint8(v) => JsonbValue::Number(JsonbNumber::UInt64(v as u64)),
523            Value::Uint16(v) => JsonbValue::Number(JsonbNumber::UInt64(v as u64)),
524            Value::Uint32(v) => JsonbValue::Number(JsonbNumber::UInt64(v as u64)),
525            Value::Uint64(v) => JsonbValue::Number(JsonbNumber::UInt64(v)),
526            Value::Float32(v) => JsonbValue::Number(JsonbNumber::Float64(v as f64)),
527            Value::Float64(v) => JsonbValue::Number(JsonbNumber::Float64(v)),
528            Value::String(v) => JsonbValue::String(v.into()),
529            Value::Timestamp(v) => JsonbValue::String(v.to_string().into()),
530            Value::Array(arr) => {
531                let mut vals: Vec<JsonbValue> = Vec::with_capacity(arr.len());
532                for val in arr.into_iter() {
533                    vals.push(val.into());
534                }
535                JsonbValue::Array(vals)
536            }
537            Value::Map(obj) => {
538                let mut map = JsonbObject::new();
539                for (k, v) in obj.values.into_iter() {
540                    let val: JsonbValue = v.into();
541                    map.insert(k, val);
542                }
543                JsonbValue::Object(map)
544            }
545        }
546    }
547}
548
549impl From<String> for Value {
550    fn from(value: String) -> Self {
551        Value::String(value)
552    }
553}
554
555impl From<&str> for Value {
556    fn from(value: &str) -> Self {
557        Value::String(value.to_string())
558    }
559}
560
561impl From<i64> for Value {
562    fn from(value: i64) -> Self {
563        Value::Int64(value)
564    }
565}
566
567impl From<f64> for Value {
568    fn from(value: f64) -> Self {
569        Value::Float64(value)
570    }
571}
572
573impl From<Vec<String>> for Value {
574    fn from(value: Vec<String>) -> Self {
575        Value::Array(Array {
576            values: value.into_iter().map(Value::String).collect(),
577        })
578    }
579}
580
581impl From<Vec<Self>> for Value {
582    fn from(value: Vec<Self>) -> Self {
583        Value::Array(Array { values: value })
584    }
585}
586
587impl From<bool> for Value {
588    fn from(value: bool) -> Self {
589        Value::Boolean(value)
590    }
591}
592
593impl JsonLike for Value {
594    fn get(&self, key: &str) -> Option<&Self> {
595        self.get(key)
596    }
597
598    fn itre(&self, pref: String) -> Vec<jsonpath_rust::JsonPathValue<Self>> {
599        let res = match self {
600            Value::Array(elems) => {
601                let mut res = vec![];
602                for (idx, el) in elems.iter().enumerate() {
603                    res.push(jsonpath_rust::JsonPathValue::Slice(
604                        el,
605                        jsonpath_rust::jsp_idx(&pref, idx),
606                    ));
607                }
608                res
609            }
610            Value::Map(elems) => {
611                let mut res = vec![];
612                for (key, el) in elems.iter() {
613                    res.push(jsonpath_rust::JsonPathValue::Slice(
614                        el,
615                        jsonpath_rust::jsp_obj(&pref, key),
616                    ));
617                }
618                res
619            }
620            _ => vec![],
621        };
622        if res.is_empty() {
623            vec![jsonpath_rust::JsonPathValue::NoValue]
624        } else {
625            res
626        }
627    }
628
629    fn array_len(&self) -> jsonpath_rust::JsonPathValue<'static, Self> {
630        match self {
631            Value::Array(elems) => {
632                jsonpath_rust::JsonPathValue::NewValue(Value::Int64(elems.len() as i64))
633            }
634            _ => jsonpath_rust::JsonPathValue::NoValue,
635        }
636    }
637
638    fn init_with_usize(cnt: usize) -> Self {
639        Value::Int64(cnt as i64)
640    }
641
642    fn deep_flatten(&self, pref: String) -> Vec<(&Self, String)> {
643        let mut acc = vec![];
644        match self {
645            Value::Map(elems) => {
646                for (f, v) in elems.iter() {
647                    let pref = jsp_obj(&pref, f);
648                    acc.push((v, pref.clone()));
649                    acc.append(&mut v.deep_flatten(pref));
650                }
651            }
652            Value::Array(elems) => {
653                for (i, v) in elems.iter().enumerate() {
654                    let pref = jsp_idx(&pref, i);
655                    acc.push((v, pref.clone()));
656                    acc.append(&mut v.deep_flatten(pref));
657                }
658            }
659            _ => (),
660        }
661        acc
662    }
663
664    fn deep_path_by_key<'a>(
665        &'a self,
666        key: jsonpath_rust::path::ObjectField<'a, Self>,
667        pref: String,
668    ) -> Vec<(&'a Self, String)> {
669        let mut result: Vec<(&'a Value, String)> = jsonpath_rust::JsonPathValue::vec_as_pair(
670            key.find(jsonpath_rust::JsonPathValue::new_slice(self, pref.clone())),
671        );
672        match self {
673            Value::Map(elems) => {
674                let mut next_levels: Vec<(&'a Value, String)> = elems
675                    .iter()
676                    .flat_map(|(k, v)| v.deep_path_by_key(key.clone(), jsp_obj(&pref, k)))
677                    .collect();
678                result.append(&mut next_levels);
679                result
680            }
681            Value::Array(elems) => {
682                let mut next_levels: Vec<(&'a Value, String)> = elems
683                    .iter()
684                    .enumerate()
685                    .flat_map(|(i, v)| v.deep_path_by_key(key.clone(), jsp_idx(&pref, i)))
686                    .collect();
687                result.append(&mut next_levels);
688                result
689            }
690            _ => result,
691        }
692    }
693
694    fn as_u64(&self) -> Option<u64> {
695        match self {
696            Value::Uint64(v) => Some(*v),
697            Value::Uint32(v) => Some(*v as u64),
698            Value::Uint16(v) => Some(*v as u64),
699            Value::Uint8(v) => Some(*v as u64),
700            Value::Int64(v) if *v >= 0 => Some(*v as u64),
701            Value::Int32(v) if *v >= 0 => Some(*v as u64),
702            Value::Int16(v) if *v >= 0 => Some(*v as u64),
703            Value::Int8(v) if *v >= 0 => Some(*v as u64),
704            Value::Float64(v) if *v >= 0.0 => Some(*v as u64),
705            Value::Float32(v) if *v >= 0.0 => Some(*v as u64),
706            _ => None,
707        }
708    }
709
710    fn is_array(&self) -> bool {
711        matches!(self, Value::Array(_))
712    }
713
714    fn as_array(&self) -> Option<&Vec<Self>> {
715        match self {
716            Value::Array(arr) => Some(&arr.values),
717            _ => None,
718        }
719    }
720
721    fn size(left: Vec<&Self>, right: Vec<&Self>) -> bool {
722        if let Some(v) = right.first() {
723            let sz = match v {
724                Value::Int64(n) => *n as usize,
725                Value::Int32(n) => *n as usize,
726                Value::Int16(n) => *n as usize,
727                Value::Int8(n) => *n as usize,
728
729                Value::Uint64(n) => *n as usize,
730                Value::Uint32(n) => *n as usize,
731                Value::Uint16(n) => *n as usize,
732                Value::Uint8(n) => *n as usize,
733                Value::Float32(n) => *n as usize,
734                Value::Float64(n) => *n as usize,
735                _ => return false,
736            };
737            for el in left.iter() {
738                match el {
739                    Value::String(v) if v.len() == sz => true,
740                    Value::Array(elems) if elems.len() == sz => true,
741                    Value::Map(fields) if fields.len() == sz => true,
742                    _ => return false,
743                };
744            }
745            return true;
746        }
747        false
748    }
749
750    fn sub_set_of(left: Vec<&Self>, right: Vec<&Self>) -> bool {
751        if left.is_empty() {
752            return true;
753        }
754        if right.is_empty() {
755            return false;
756        }
757
758        if let Some(elems) = left.first().and_then(|e| e.as_array()) {
759            if let Some(Value::Array(right_elems)) = right.first() {
760                if right_elems.is_empty() {
761                    return false;
762                }
763
764                for el in elems {
765                    let mut res = false;
766
767                    for r in right_elems.iter() {
768                        if el.eq(r) {
769                            res = true
770                        }
771                    }
772                    if !res {
773                        return false;
774                    }
775                }
776                return true;
777            }
778        }
779        false
780    }
781
782    fn any_of(left: Vec<&Self>, right: Vec<&Self>) -> bool {
783        if left.is_empty() {
784            return true;
785        }
786        if right.is_empty() {
787            return false;
788        }
789
790        if let Some(Value::Array(elems)) = right.first() {
791            if elems.is_empty() {
792                return false;
793            }
794
795            for el in left.iter() {
796                if let Some(left_elems) = el.as_array() {
797                    for l in left_elems.iter() {
798                        for r in elems.iter() {
799                            if l.eq(r) {
800                                return true;
801                            }
802                        }
803                    }
804                } else {
805                    for r in elems.iter() {
806                        if el.eq(&r) {
807                            return true;
808                        }
809                    }
810                }
811            }
812        }
813
814        false
815    }
816
817    fn regex(left: Vec<&Self>, right: Vec<&Self>) -> bool {
818        if left.is_empty() || right.is_empty() {
819            return false;
820        }
821
822        match right.first() {
823            Some(Value::String(str)) => {
824                if let Ok(regex) = Regex::new(str) {
825                    for el in left.iter() {
826                        if let Some(v) = el.as_str() {
827                            if regex.is_match(v) {
828                                return true;
829                            }
830                        }
831                    }
832                }
833                false
834            }
835            _ => false,
836        }
837    }
838
839    fn inside(left: Vec<&Self>, right: Vec<&Self>) -> bool {
840        if left.is_empty() {
841            return false;
842        }
843
844        match right.first() {
845            Some(Value::Array(elems)) => {
846                for el in left.iter() {
847                    if elems.contains(el) {
848                        return true;
849                    }
850                }
851                false
852            }
853            Some(Value::Map(elems)) => {
854                for el in left.iter() {
855                    for r in elems.values() {
856                        if el.eq(&r) {
857                            return true;
858                        }
859                    }
860                }
861                false
862            }
863            _ => false,
864        }
865    }
866
867    fn less(left: Vec<&Self>, right: Vec<&Self>) -> bool {
868        if left.len() == 1 && right.len() == 1 {
869            match (left.first(), right.first()) {
870                (Some(l), Some(r)) => l
871                    .as_f64()
872                    .and_then(|v1| r.as_f64().map(|v2| v1 < v2))
873                    .unwrap_or(false),
874                _ => false,
875            }
876        } else {
877            false
878        }
879    }
880
881    fn eq(left: Vec<&Self>, right: Vec<&Self>) -> bool {
882        if left.len() != right.len() {
883            false
884        } else {
885            left.iter().zip(right).all(|(a, b)| a.eq(&b))
886        }
887    }
888
889    fn array(data: Vec<Self>) -> Self {
890        Value::Array(Array { values: data })
891    }
892
893    fn null() -> Self {
894        Value::Null
895    }
896
897    // ref https://github.com/besok/jsonpath-rust/blob/main/src/path/mod.rs#L423
898    fn reference<T>(
899        &self,
900        path: T,
901    ) -> std::result::Result<std::option::Option<&Value>, JsonPathParserError>
902    where
903        T: Into<JsonPathStr>,
904    {
905        Ok(self.pointer(&path_to_json_path(path.into())?))
906    }
907
908    // https://github.com/besok/jsonpath-rust/blob/main/src/path/mod.rs#L430
909    fn reference_mut<T>(
910        &mut self,
911        path: T,
912    ) -> std::result::Result<std::option::Option<&mut Value>, JsonPathParserError>
913    where
914        T: Into<JsonPathStr>,
915    {
916        Ok(self.pointer_mut(&path_to_json_path(path.into())?))
917    }
918}
919
920// ref https://github.com/besok/jsonpath-rust/blob/main/src/path/mod.rs#L438
921fn path_to_json_path(path: JsonPathStr) -> StdResult<String, JsonPathParserError> {
922    convert_part(&parse_json_path(path.as_str())?)
923}
924
925// https://github.com/besok/jsonpath-rust/blob/main/src/path/mod.rs#L442
926fn convert_part(path: &JsonPath) -> StdResult<String, JsonPathParserError> {
927    match path {
928        JsonPath::Chain(elems) => elems
929            .iter()
930            .map(convert_part)
931            .collect::<StdResult<String, JsonPathParserError>>(),
932
933        JsonPath::Index(JsonPathIndex::Single(v)) => Ok(format!("/{}", v)),
934        JsonPath::Field(e) => Ok(format!("/{}", e)),
935        JsonPath::Root => Ok("".to_string()),
936        e => Err(JsonPathParserError::InvalidJsonPath(e.to_string())),
937    }
938}
pipeline/etl/value.rs

pipeline/etl/
value.rs
