4283045331

Build Type

push

github

Committed by GitHub

Commit Message

feat: Support timestamp diff (#1074)

Run Details

58 of 58 new or added lines in 2 files covered. (100.0%)

27146 of 37535 relevant lines covered (72.32%)

33460.71 hits per line

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

94.98

/dozer-sql/src/pipeline/expression/tests/execution.rs

use crate::pipeline::builder::SchemaSQLContext;
use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};
use crate::pipeline::expression::mathematical::evaluate_sub;
use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
use crate::pipeline::expression::scalar::common::ScalarFunctionType;
use crate::pipeline::projection::factory::ProjectionProcessorFactory;
use crate::pipeline::tests::utils::get_select;
use dozer_core::node::ProcessorFactory;
use dozer_core::DEFAULT_PORT_HANDLE;
use dozer_types::chrono::DateTime;
use dozer_types::types::{Field, FieldDefinition, FieldType, Record, Schema, SourceDefinition};

#[test]
fn test_column_execution() {
    use dozer_types::ordered_float::OrderedFloat;

    let schema = Schema::empty()
        .field(
            FieldDefinition::new(
                "int_field".to_string(),
                FieldType::Int,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .field(
            FieldDefinition::new(
                "str_field".to_string(),
                FieldType::String,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .field(
            FieldDefinition::new(
                "float_field".to_string(),
                FieldType::Float,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .clone();

    let record = Record::new(
        None,
        vec![
            Field::Int(1337),
            Field::String("test".to_string()),
            Field::Float(OrderedFloat(10.10)),
        ],
        None,
    );

    // Column
    let e = Expression::Column { index: 0 };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Int(1337)
    );

    let e = Expression::Column { index: 1 };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::String("test".to_string())
    );

    let e = Expression::Column { index: 2 };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Float(OrderedFloat(10.10))
    );

    // Literal
    let e = Expression::Literal(Field::Int(1337));
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Int(1337)
    );

    // UnaryOperator
    let e = Expression::UnaryOperator {
        operator: UnaryOperatorType::Not,
        arg: Box::new(Expression::Literal(Field::Boolean(true))),
    };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Boolean(false)
    );

    // BinaryOperator
    let e = Expression::BinaryOperator {
        left: Box::new(Expression::Literal(Field::Boolean(true))),
        operator: BinaryOperatorType::And,
        right: Box::new(Expression::Literal(Field::Boolean(false))),
    };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Boolean(false),
    );

    // ScalarFunction
    let e = Expression::ScalarFunction {
        fun: ScalarFunctionType::Abs,
        args: vec![Expression::Literal(Field::Int(-1))],
    };
    assert_eq!(
        e.evaluate(&record, &schema)
            .unwrap_or_else(|e| panic!("{}", e.to_string())),
        Field::Int(1)
    );
}

#[test]
fn test_alias() {
    let schema = Schema::empty()
        .field(
            FieldDefinition::new(
                String::from("fn"),
                FieldType::Text,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .field(
            FieldDefinition::new(
                String::from("ln"),
                FieldType::String,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .clone();

    let select = get_select("SELECT count(fn) AS alias1, ln as alias2 FROM t1").unwrap();
    let processor_factory = ProjectionProcessorFactory::_new(select.projection);
    let r = processor_factory
        .get_output_schema(
            &DEFAULT_PORT_HANDLE,
            &[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]
                .into_iter()
                .collect(),
        )
        .unwrap()
        .0;

    assert_eq!(
        r,
        Schema::empty()
            .field(
                FieldDefinition::new(
                    String::from("alias1"),
                    FieldType::Text,
                    false,
                    SourceDefinition::Dynamic
                ),
                false,
            )
            .field(
                FieldDefinition::new(
                    String::from("alias2"),
                    FieldType::String,
                    false,
                    SourceDefinition::Dynamic
                ),
                false,
            )
            .clone()
    );
}

#[test]
fn test_wildcard() {
    let schema = Schema::empty()
        .field(
            FieldDefinition::new(
                String::from("fn"),
                FieldType::Text,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .field(
            FieldDefinition::new(
                String::from("ln"),
                FieldType::String,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .clone();

    let select = get_select("SELECT * FROM t1").unwrap();
    let processor_factory = ProjectionProcessorFactory::_new(select.projection);
    let r = processor_factory
        .get_output_schema(
            &DEFAULT_PORT_HANDLE,
            &[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]
                .into_iter()
                .collect(),
        )
        .unwrap()
        .0;

    assert_eq!(
        r,
        Schema::empty()
            .field(
                FieldDefinition::new(
                    String::from("fn"),
                    FieldType::Text,
                    false,
                    SourceDefinition::Dynamic
                ),
                false,
            )
            .field(
                FieldDefinition::new(
                    String::from("ln"),
                    FieldType::String,
                    false,
                    SourceDefinition::Dynamic
                ),
                false,
            )
            .clone()
    );
}

#[test]
fn test_timestamp_difference() {
    let schema = Schema::empty()
        .field(
            FieldDefinition::new(
                String::from("a"),
                FieldType::Timestamp,
                false,
                SourceDefinition::Dynamic,
            ),
            true,
        )
        .field(
            FieldDefinition::new(
                String::from("b"),
                FieldType::Timestamp,
                false,
                SourceDefinition::Dynamic,
            ),
            false,
        )
        .clone();

    let record = Record::new(
        None,
        vec![
            Field::Timestamp(DateTime::parse_from_rfc3339("2020-01-01T00:13:00Z").unwrap()),
            Field::Timestamp(DateTime::parse_from_rfc3339("2020-01-01T00:12:10Z").unwrap()),
        ],
        Some(1),
    );

    let result = evaluate_sub(
        &schema,
        &Expression::Column { index: 0 },
        &Expression::Column { index: 1 },
        &record,
    )
    .unwrap();
    assert_eq!(result, Field::Int(50000));

    let result = evaluate_sub(
        &schema,
        &Expression::Column { index: 1 },
        &Expression::Column { index: 0 },
        &record,
    )
    .unwrap();
    assert_eq!(result, Field::Int(-50000));
}

1	use crate::pipeline::builder::SchemaSQLContext;
2	use crate::pipeline::expression::execution::{Expression, ExpressionExecutor};
3	use crate::pipeline::expression::mathematical::evaluate_sub;
4	use crate::pipeline::expression::operator::{BinaryOperatorType, UnaryOperatorType};
5	use crate::pipeline::expression::scalar::common::ScalarFunctionType;
6	use crate::pipeline::projection::factory::ProjectionProcessorFactory;
7	use crate::pipeline::tests::utils::get_select;
8	use dozer_core::node::ProcessorFactory;
9	use dozer_core::DEFAULT_PORT_HANDLE;
10	use dozer_types::chrono::DateTime;
11	use dozer_types::types::{Field, FieldDefinition, FieldType, Record, Schema, SourceDefinition};	×
12		×
13	#[test]	1✔
14	fn test_column_execution() {	1✔
15	use dozer_types::ordered_float::OrderedFloat;	1✔
16		1✔
17	let schema = Schema::empty()	1✔
18	.field(	1✔
19	FieldDefinition::new(	1✔
20	"int_field".to_string(),	1✔
21	FieldType::Int,	1✔
22	false,	1✔
23	SourceDefinition::Dynamic,	1✔
24	),	1✔
25	false,	1✔
26	)	1✔
27	.field(	1✔
28	FieldDefinition::new(	1✔
29	"str_field".to_string(),	1✔
30	FieldType::String,	1✔
31	false,	1✔
32	SourceDefinition::Dynamic,	1✔
33	),	1✔
34	false,	1✔
35	)	1✔
36	.field(	1✔
37	FieldDefinition::new(	1✔
38	"float_field".to_string(),	1✔
39	FieldType::Float,	1✔
40	false,	1✔
41	SourceDefinition::Dynamic,	1✔
42	),	1✔
43	false,	1✔
44	)	1✔
45	.clone();	1✔
46		1✔
47	let record = Record::new(	1✔
48	None,	1✔
49	vec![	1✔
50	Field::Int(1337),	1✔
51	Field::String("test".to_string()),	1✔
52	Field::Float(OrderedFloat(10.10)),	1✔
53	],	1✔
54	None,	1✔
55	);	1✔
56		1✔
57	// Column	1✔
58	let e = Expression::Column { index: 0 };	1✔
59	assert_eq!(	1✔
60	e.evaluate(&record, &schema)	1✔
61	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
62	Field::Int(1337)	1✔
63	);	1✔
64		×
65	let e = Expression::Column { index: 1 };	1✔
66	assert_eq!(	1✔
67	e.evaluate(&record, &schema)	1✔
68	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
69	Field::String("test".to_string())	1✔
70	);	1✔
71		×
72	let e = Expression::Column { index: 2 };	1✔
73	assert_eq!(	1✔
74	e.evaluate(&record, &schema)	1✔
75	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
76	Field::Float(OrderedFloat(10.10))	1✔
77	);	1✔
78		×
79	// Literal	×
80	let e = Expression::Literal(Field::Int(1337));	1✔
81	assert_eq!(	1✔
82	e.evaluate(&record, &schema)	1✔
83	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
84	Field::Int(1337)	1✔
85	);	1✔
86		×
87	// UnaryOperator	×
88	let e = Expression::UnaryOperator {	1✔
89	operator: UnaryOperatorType::Not,	1✔
90	arg: Box::new(Expression::Literal(Field::Boolean(true))),	1✔
91	};	1✔
92	assert_eq!(	1✔
93	e.evaluate(&record, &schema)	1✔
94	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
95	Field::Boolean(false)	1✔
96	);	1✔
97		×
98	// BinaryOperator	×
99	let e = Expression::BinaryOperator {	1✔
100	left: Box::new(Expression::Literal(Field::Boolean(true))),	1✔
101	operator: BinaryOperatorType::And,	1✔
102	right: Box::new(Expression::Literal(Field::Boolean(false))),	1✔
103	};	1✔
104	assert_eq!(	1✔
105	e.evaluate(&record, &schema)	1✔
106	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
107	Field::Boolean(false),	1✔
108	);	1✔
109		×
110	// ScalarFunction	×
111	let e = Expression::ScalarFunction {	1✔
112	fun: ScalarFunctionType::Abs,	1✔
113	args: vec![Expression::Literal(Field::Int(-1))],	1✔
114	};	1✔
115	assert_eq!(	1✔
116	e.evaluate(&record, &schema)	1✔
117	.unwrap_or_else(\|e\| panic!("{}", e.to_string())),	1✔
118	Field::Int(1)	1✔
119	);	1✔
120	}	1✔
121		×
122	#[test]	1✔
123	fn test_alias() {	1✔
124	let schema = Schema::empty()	1✔
125	.field(	1✔
126	FieldDefinition::new(	1✔
127	String::from("fn"),	1✔
128	FieldType::Text,	1✔
129	false,	1✔
130	SourceDefinition::Dynamic,	1✔
131	),	1✔
132	false,	1✔
133	)	1✔
134	.field(	1✔
135	FieldDefinition::new(	1✔
136	String::from("ln"),	1✔
137	FieldType::String,	1✔
138	false,	1✔
139	SourceDefinition::Dynamic,	1✔
140	),	1✔
141	false,	1✔
142	)	1✔
143	.clone();	1✔
144		1✔
145	let select = get_select("SELECT count(fn) AS alias1, ln as alias2 FROM t1").unwrap();	1✔
146	let processor_factory = ProjectionProcessorFactory::_new(select.projection);	1✔
147	let r = processor_factory	1✔
148	.get_output_schema(	1✔
149	&DEFAULT_PORT_HANDLE,	1✔
150	&[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]	1✔
151	.into_iter()	1✔
152	.collect(),	1✔
153	)	1✔
154	.unwrap()	1✔
155	.0;	1✔
156		1✔
157	assert_eq!(	1✔
158	r,	1✔
159	Schema::empty()	1✔
160	.field(	1✔
161	FieldDefinition::new(	1✔
162	String::from("alias1"),	1✔
163	FieldType::Text,	1✔
164	false,	1✔
165	SourceDefinition::Dynamic	1✔
166	),	1✔
167	false,	1✔
168	)	1✔
169	.field(	1✔
170	FieldDefinition::new(	1✔
171	String::from("alias2"),	1✔
172	FieldType::String,	1✔
173	false,	1✔
174	SourceDefinition::Dynamic	1✔
175	),	1✔
176	false,	1✔
177	)	1✔
178	.clone()	1✔
179	);	1✔
180	}	1✔
181		×
182	#[test]	1✔
183	fn test_wildcard() {	1✔
184	let schema = Schema::empty()	1✔
185	.field(	1✔
186	FieldDefinition::new(	1✔
187	String::from("fn"),	1✔
188	FieldType::Text,	1✔
189	false,	1✔
190	SourceDefinition::Dynamic,	1✔
191	),	1✔
192	false,	1✔
193	)	1✔
194	.field(	1✔
195	FieldDefinition::new(	1✔
196	String::from("ln"),	1✔
197	FieldType::String,	1✔
198	false,	1✔
199	SourceDefinition::Dynamic,	1✔
200	),	1✔
201	false,	1✔
202	)	1✔
203	.clone();	1✔
204		1✔
205	let select = get_select("SELECT * FROM t1").unwrap();	1✔
206	let processor_factory = ProjectionProcessorFactory::_new(select.projection);	1✔
207	let r = processor_factory	1✔
208	.get_output_schema(	1✔
209	&DEFAULT_PORT_HANDLE,	1✔
210	&[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]	1✔
211	.into_iter()	1✔
212	.collect(),	1✔
213	)	1✔
214	.unwrap()	1✔
215	.0;	1✔
216		1✔
217	assert_eq!(	1✔
218	r,	1✔
219	Schema::empty()	1✔
220	.field(	1✔
221	FieldDefinition::new(	1✔
222	String::from("fn"),	1✔
223	FieldType::Text,	1✔
224	false,	1✔
225	SourceDefinition::Dynamic	1✔
226	),	1✔
227	false,	1✔
228	)	1✔
229	.field(	1✔
230	FieldDefinition::new(	1✔
231	String::from("ln"),	1✔
232	FieldType::String,	1✔
233	false,	1✔
234	SourceDefinition::Dynamic	1✔
235	),	1✔
236	false,	1✔
237	)	1✔
238	.clone()	1✔
239	);	1✔
240	}	1✔
241
242	#[test]	1✔
243	fn test_timestamp_difference() {	1✔
244	let schema = Schema::empty()	1✔
245	.field(	1✔
246	FieldDefinition::new(	1✔
247	String::from("a"),	1✔
248	FieldType::Timestamp,	1✔
249	false,	1✔
250	SourceDefinition::Dynamic,	1✔
251	),	1✔
252	true,	1✔
253	)	1✔
254	.field(	1✔
255	FieldDefinition::new(	1✔
256	String::from("b"),	1✔
257	FieldType::Timestamp,	1✔
258	false,	1✔
259	SourceDefinition::Dynamic,	1✔
260	),	1✔
261	false,	1✔
262	)	1✔
263	.clone();	1✔
264		1✔
265	let record = Record::new(	1✔
266	None,	1✔
267	vec![	1✔
268	Field::Timestamp(DateTime::parse_from_rfc3339("2020-01-01T00:13:00Z").unwrap()),	1✔
269	Field::Timestamp(DateTime::parse_from_rfc3339("2020-01-01T00:12:10Z").unwrap()),	1✔
270	],	1✔
271	Some(1),	1✔
272	);	1✔
273		1✔
274	let result = evaluate_sub(	1✔
275	&schema,	1✔
276	&Expression::Column { index: 0 },	1✔
277	&Expression::Column { index: 1 },	1✔
278	&record,	1✔
279	)	1✔
280	.unwrap();	1✔
281	assert_eq!(result, Field::Int(50000));	1✔
282
283	let result = evaluate_sub(	1✔
284	&schema,	1✔
285	&Expression::Column { index: 1 },	1✔
286	&Expression::Column { index: 0 },	1✔
287	&record,	1✔
288	)	1✔
289	.unwrap();	1✔
290	assert_eq!(result, Field::Int(-50000));	1✔
291	}	1✔

getdozer / dozer / 4283045331

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