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

pipeline/etl/
value.rs
