5630015928

pending completion

Build # 5630015928

Build Type

push

github

Committed by

web-flow

Commit Message

Bump version (#1779)

Run Details

42841 of 55898 relevant lines covered (76.64%)

32850.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.72

/dozer-sql/src/pipeline/expression/json_functions.rs

use crate::pipeline::errors::PipelineError;
use crate::pipeline::errors::PipelineError::{
    InvalidArgument, InvalidFunction, InvalidFunctionArgument, InvalidValue,
};
use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};

use crate::jsonpath::{JsonPathFinder, JsonPathInst};
use dozer_types::json_types::JsonValue;
use dozer_types::types::{Field, Record, Schema};
use std::fmt::{Display, Formatter};
use std::str::FromStr;

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
pub enum JsonFunctionType {
    JsonValue,
    JsonQuery,
}

impl Display for JsonFunctionType {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            JsonFunctionType::JsonValue => f.write_str("JSON_VALUE".to_string().as_str()),
            JsonFunctionType::JsonQuery => f.write_str("JSON_QUERY".to_string().as_str()),
        }
    }
}

impl JsonFunctionType {
    pub(crate) fn new(name: &str) -> Result<JsonFunctionType, PipelineError> {
        match name {
            "json_value" => Ok(JsonFunctionType::JsonValue),
            "json_query" => Ok(JsonFunctionType::JsonQuery),
            _ => Err(InvalidFunction(name.to_string())),
        }
    }

    pub(crate) fn evaluate(
        &self,
        schema: &Schema,
        args: &Vec<Expression>,
        record: &Record,
    ) -> Result<Field, PipelineError> {
        match self {
            JsonFunctionType::JsonValue => self.evaluate_json_value(schema, args, record),
            JsonFunctionType::JsonQuery => self.evaluate_json_query(schema, args, record),
        }
    }

    pub(crate) fn evaluate_json_value(
        &self,
        schema: &Schema,
        args: &Vec<Expression>,
        record: &Record,
    ) -> Result<Field, PipelineError> {
        if args.len() > 2 {
            return Err(InvalidFunctionArgument(
                self.to_string(),
                args[2].evaluate(record, schema)?,
                2,
            ));
        }
        let json_input = args[0].evaluate(record, schema)?;
        let path = args[1]
            .evaluate(record, schema)?
            .to_string()
            .ok_or(InvalidArgument(args[1].to_string(schema)))?;

        Ok(Field::Json(self.evaluate_json(json_input, path)?))
    }

    pub(crate) fn evaluate_json_query(
        &self,
        schema: &Schema,
        args: &Vec<Expression>,
        record: &Record,
    ) -> Result<Field, PipelineError> {
        let mut path = String::from("$");
        if args.len() < 2 && !args.is_empty() {
            Ok(Field::Json(
                self.evaluate_json(args[0].evaluate(record, schema)?, path)?,
            ))
        } else if args.len() == 2 {
            let json_input = args[0].evaluate(record, schema)?;
            path = args[1]
                .evaluate(record, schema)?
                .to_string()
                .ok_or(InvalidArgument(args[1].to_string(schema)))?;

            Ok(Field::Json(self.evaluate_json(json_input, path)?))
        } else {
            Err(InvalidFunctionArgument(
                self.to_string(),
                args[2].evaluate(record, schema)?,
                2,
            ))
        }
    }

    pub(crate) fn evaluate_json(
        &self,
        json_input: Field,
        path: String,
    ) -> Result<JsonValue, PipelineError> {
        let json_val = match json_input.to_json() {
            Some(json) => json,
            None => JsonValue::Null,
        };

        let finder = JsonPathFinder::new(
            Box::from(json_val),
            Box::from(JsonPathInst::from_str(path.as_str()).map_err(InvalidArgument)?),
        );

        match finder.find() {
            JsonValue::Null => Ok(JsonValue::Null),
            JsonValue::Array(a) => {
                if a.is_empty() {
                    Ok(JsonValue::Array(vec![]))
                } else if a.len() == 1 {
                    let item = match a.first() {
                        Some(i) => i,
                        None => return Err(InvalidValue("Invalid length of array".to_string())),
                    };
                    Ok(item.to_owned())
                } else {
                    let mut array_val = vec![];
                    for item in a {
                        array_val.push(item);
                    }
                    Ok(JsonValue::Array(array_val))
                }
            }
            _ => Err(InvalidValue(path)),
        }
    }
}

1	use crate::pipeline::errors::PipelineError;
2	use crate::pipeline::errors::PipelineError::{
3	InvalidArgument, InvalidFunction, InvalidFunctionArgument, InvalidValue,
4	};
5	use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};
6
7	use crate::jsonpath::{JsonPathFinder, JsonPathInst};
8	use dozer_types::json_types::JsonValue;
9	use dozer_types::types::{Field, Record, Schema};
10	use std::fmt::{Display, Formatter};
11	use std::str::FromStr;
12
13	#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]	15✔
14	pub enum JsonFunctionType {
15	JsonValue,
16	JsonQuery,
17	}
18
19	impl Display for JsonFunctionType {
20	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {	×
21	match self {	×
22	JsonFunctionType::JsonValue => f.write_str("JSON_VALUE".to_string().as_str()),	×
23	JsonFunctionType::JsonQuery => f.write_str("JSON_QUERY".to_string().as_str()),	×
24	}
25	}	×
26	}
27
28	impl JsonFunctionType {
29	pub(crate) fn new(name: &str) -> Result<JsonFunctionType, PipelineError> {	30✔
30	match name {	30✔
31	"json_value" => Ok(JsonFunctionType::JsonValue),	30✔
32	"json_query" => Ok(JsonFunctionType::JsonQuery),	12✔
33	_ => Err(InvalidFunction(name.to_string())),	×
34	}
35	}	30✔
36
37	pub(crate) fn evaluate(	15✔
38	&self,	15✔
39	schema: &Schema,	15✔
40	args: &Vec<Expression>,	15✔
41	record: &Record,	15✔
42	) -> Result<Field, PipelineError> {	15✔
43	match self {	15✔
44	JsonFunctionType::JsonValue => self.evaluate_json_value(schema, args, record),	9✔
45	JsonFunctionType::JsonQuery => self.evaluate_json_query(schema, args, record),	6✔
46	}
47	}	15✔
48
49	pub(crate) fn evaluate_json_value(	9✔
50	&self,	9✔
51	schema: &Schema,	9✔
52	args: &Vec<Expression>,	9✔
53	record: &Record,	9✔
54	) -> Result<Field, PipelineError> {	9✔
55	if args.len() > 2 {	9✔
56	return Err(InvalidFunctionArgument(
57	self.to_string(),	×
58	args[2].evaluate(record, schema)?,	×
59	2,
60	));
61	}	9✔
62	let json_input = args[0].evaluate(record, schema)?;	9✔
63	let path = args[1]	9✔
64	.evaluate(record, schema)?	9✔
65	.to_string()	9✔
66	.ok_or(InvalidArgument(args[1].to_string(schema)))?;	9✔
67
68	Ok(Field::Json(self.evaluate_json(json_input, path)?))	9✔
69	}	9✔
70
71	pub(crate) fn evaluate_json_query(	6✔
72	&self,	6✔
73	schema: &Schema,	6✔
74	args: &Vec<Expression>,	6✔
75	record: &Record,	6✔
76	) -> Result<Field, PipelineError> {	6✔
77	let mut path = String::from("$");	6✔
78	if args.len() < 2 && !args.is_empty() {	6✔
79	Ok(Field::Json(
80	self.evaluate_json(args[0].evaluate(record, schema)?, path)?,	1✔
81	))
82	} else if args.len() == 2 {	5✔
83	let json_input = args[0].evaluate(record, schema)?;	5✔
84	path = args[1]	5✔
85	.evaluate(record, schema)?	5✔
86	.to_string()	5✔
87	.ok_or(InvalidArgument(args[1].to_string(schema)))?;	5✔
88
89	Ok(Field::Json(self.evaluate_json(json_input, path)?))	5✔
90	} else {
91	Err(InvalidFunctionArgument(
92	self.to_string(),	×
93	args[2].evaluate(record, schema)?,	×
94	2,
95	))
96	}
97	}	6✔
98
99	pub(crate) fn evaluate_json(	15✔
100	&self,	15✔
101	json_input: Field,	15✔
102	path: String,	15✔
103	) -> Result<JsonValue, PipelineError> {	15✔
104	let json_val = match json_input.to_json() {	15✔
105	Some(json) => json,	15✔
106	None => JsonValue::Null,	×
107	};
108
109	let finder = JsonPathFinder::new(	15✔
110	Box::from(json_val),	15✔
111	Box::from(JsonPathInst::from_str(path.as_str()).map_err(InvalidArgument)?),	15✔
112	);
113
114	match finder.find() {	15✔
115	JsonValue::Null => Ok(JsonValue::Null),	1✔
116	JsonValue::Array(a) => {	14✔
117	if a.is_empty() {	14✔
118	Ok(JsonValue::Array(vec![]))	×
119	} else if a.len() == 1 {	14✔
120	let item = match a.first() {	12✔
121	Some(i) => i,	12✔
122	None => return Err(InvalidValue("Invalid length of array".to_string())),	×
123	};
124	Ok(item.to_owned())	12✔
125	} else {
126	let mut array_val = vec![];	2✔
127	for item in a {	10✔
128	array_val.push(item);	8✔
129	}	8✔
130	Ok(JsonValue::Array(array_val))	2✔
131	}
132	}
133	_ => Err(InvalidValue(path)),	×
134	}
135	}	15✔
136	}

getdozer / dozer / 5630015928

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous