4007820649

Build # 4007820649

Build Type

github

Committed by GitHub

Commit Message

Merge b71e66da1 into 6c0ac2b2c

Pull Request Pull Request #734: Bump ahash from 0.8.2 to 0.8.3

Run Details

23507 of 35166 relevant lines covered (66.85%)

40241.5 hits per line

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

92.2

/dozer-sql/src/pipeline/aggregation/tests/aggregation_null.rs

use crate::output;
use crate::pipeline::aggregation::factory::AggregationProcessorFactory;
use crate::pipeline::aggregation::tests::aggregation_tests_utils::{
    delete_exp, delete_field, init_input_schema, init_processor, insert_exp, insert_field,
    FIELD_100_INT, FIELD_1_INT, ITALY,
};
use crate::pipeline::builder::SchemaSQLContext;
use crate::pipeline::expression::builder::NameOrAlias;
use crate::pipeline::tests::utils::get_select;
use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
use dozer_core::dag::node::ProcessorFactory;
use dozer_types::types::FieldType::Int;
use dozer_types::types::{
    Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
};
use std::collections::HashMap;

#[test]
fn test_sum_aggregation_null() {
    let schema = init_input_schema(Int, "SUM");
    let (processor, tx) = init_processor(
        "SELECT Country, SUM(Salary) \
        FROM Users \
        WHERE Salary >= 1 GROUP BY Country",
        HashMap::from([(DEFAULT_PORT_HANDLE, schema)]),
    )
    .unwrap();

    // Insert 100 for segment Italy
    /*
        NULL, 100.0
        -------------
        SUM = 100.0
    */
    let inp = Operation::Insert {
        new: Record::new(
            None,
            vec![
                Field::Int(0),
                Field::Null,
                FIELD_100_INT.clone(),
                FIELD_100_INT.clone(),
            ],
            None,
        ),
    };
    let out = output!(processor, inp, tx);
    let exp = vec![Operation::Insert {
        new: Record::new(None, vec![Field::Null, FIELD_100_INT.clone()], None),
    }];
    assert_eq!(out, exp);
}

#[test]
fn test_sum_aggregation_del_and_insert() {
    let schema = init_input_schema(Int, "COUNT");
    let (processor, tx) = init_processor(
        "SELECT Country, COUNT(Salary) \
        FROM Users \
        WHERE Salary >= 1 GROUP BY Country",
        HashMap::from([(DEFAULT_PORT_HANDLE, schema)]),
    )
    .unwrap();

    // Insert 100 for segment Italy
    /*
        Italy, 100.0
        -------------
        COUNT = 1
    */
    let mut inp = insert_field(ITALY, FIELD_100_INT);
    let mut out = output!(processor, inp, tx);
    let mut exp = vec![insert_exp(ITALY, FIELD_1_INT)];
    assert_eq!(out, exp);

    // Delete last record
    /*
        -------------
        COUNT = 0
    */
    inp = delete_field(ITALY, FIELD_100_INT);
    out = output!(processor, inp, tx);
    exp = vec![delete_exp(ITALY, FIELD_1_INT)];
    assert_eq!(out, exp);

    // Insert 100 for segment Italy
    /*
        Italy, 100.0
        -------------
        COUNT = 1
    */
    let inp = insert_field(ITALY, FIELD_100_INT);
    let out = output!(processor, inp, tx);
    let exp = vec![insert_exp(ITALY, FIELD_1_INT)];
    assert_eq!(out, exp);
}

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

    let select = get_select("SELECT ID, SUM(Salary) as Salaries FROM Users GROUP BY ID").unwrap();

    let factory = AggregationProcessorFactory::new(
        NameOrAlias("Users".to_string(), None),
        select.projection,
        select.group_by,
        false,
    );
    let out_schema = factory
        .get_output_schema(
            &DEFAULT_PORT_HANDLE,
            &[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]
                .into_iter()
                .collect(),
        )
        .unwrap()
        .0;

    assert_eq!(
        out_schema,
        Schema::empty()
            .field(
                FieldDefinition::new(
                    String::from("ID"),
                    FieldType::Int,
                    false,
                    SourceDefinition::Dynamic
                ),
                true,
            )
            .field(
                FieldDefinition::new(
                    String::from("Salaries"),
                    FieldType::Int,
                    false,
                    SourceDefinition::Dynamic
                ),
                false,
            )
            .clone()
    );
}

1	use crate::output;
2	use crate::pipeline::aggregation::factory::AggregationProcessorFactory;
3	use crate::pipeline::aggregation::tests::aggregation_tests_utils::{
4	delete_exp, delete_field, init_input_schema, init_processor, insert_exp, insert_field,
5	FIELD_100_INT, FIELD_1_INT, ITALY,
6	};
7	use crate::pipeline::builder::SchemaSQLContext;
8	use crate::pipeline::expression::builder::NameOrAlias;
9	use crate::pipeline::tests::utils::get_select;
10	use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
11	use dozer_core::dag::node::ProcessorFactory;
12	use dozer_types::types::FieldType::Int;
13	use dozer_types::types::{
14	Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
15	};	×
16	use std::collections::HashMap;	×
17		×
18	#[test]	1✔
19	fn test_sum_aggregation_null() {	1✔
20	let schema = init_input_schema(Int, "SUM");	1✔
21	let (processor, tx) = init_processor(	1✔
22	"SELECT Country, SUM(Salary) \	1✔
23	FROM Users \	1✔
24	WHERE Salary >= 1 GROUP BY Country",	1✔
25	HashMap::from([(DEFAULT_PORT_HANDLE, schema)]),	1✔
26	)	1✔
27	.unwrap();	1✔
28		1✔
29	// Insert 100 for segment Italy	1✔
30	/*	1✔
31	NULL, 100.0	1✔
32	-------------	1✔
33	SUM = 100.0	1✔
34	*/	1✔
35	let inp = Operation::Insert {	1✔
36	new: Record::new(	1✔
37	None,	1✔
38	vec![	1✔
39	Field::Int(0),	1✔
40	Field::Null,	1✔
41	FIELD_100_INT.clone(),	1✔
42	FIELD_100_INT.clone(),	1✔
43	],	1✔
44	None,	1✔
45	),	1✔
46	};	1✔
47	let out = output!(processor, inp, tx);	1✔
48	let exp = vec![Operation::Insert {	1✔
49	new: Record::new(None, vec![Field::Null, FIELD_100_INT.clone()], None),	1✔
50	}];	1✔
51	assert_eq!(out, exp);	1✔
52	}	1✔
53		×
54	#[test]	1✔
55	fn test_sum_aggregation_del_and_insert() {	1✔
56	let schema = init_input_schema(Int, "COUNT");	1✔
57	let (processor, tx) = init_processor(	1✔
58	"SELECT Country, COUNT(Salary) \	1✔
59	FROM Users \	1✔
60	WHERE Salary >= 1 GROUP BY Country",	1✔
61	HashMap::from([(DEFAULT_PORT_HANDLE, schema)]),	1✔
62	)	1✔
63	.unwrap();	1✔
64		1✔
65	// Insert 100 for segment Italy	1✔
66	/*	1✔
67	Italy, 100.0	1✔
68	-------------	1✔
69	COUNT = 1	1✔
70	*/	1✔
71	let mut inp = insert_field(ITALY, FIELD_100_INT);	1✔
72	let mut out = output!(processor, inp, tx);	1✔
73	let mut exp = vec![insert_exp(ITALY, FIELD_1_INT)];	1✔
74	assert_eq!(out, exp);	1✔
75
76	// Delete last record
77	/*
78	-------------	×
79	COUNT = 0	×
80	*/	×
81	inp = delete_field(ITALY, FIELD_100_INT);	1✔
82	out = output!(processor, inp, tx);	1✔
83	exp = vec![delete_exp(ITALY, FIELD_1_INT)];	1✔
84	assert_eq!(out, exp);	1✔
85
86	// Insert 100 for segment Italy
87	/*
88	Italy, 100.0
89	-------------	×
90	COUNT = 1	×
91	*/	×
92	let inp = insert_field(ITALY, FIELD_100_INT);	1✔
93	let out = output!(processor, inp, tx);	1✔
94	let exp = vec![insert_exp(ITALY, FIELD_1_INT)];	1✔
95	assert_eq!(out, exp);	1✔
96	}	1✔
97		×
98	#[test]	1✔
99	fn test_aggregation_alias() {	1✔
100	let schema = Schema::empty()	1✔
101	.field(	1✔
102	FieldDefinition::new(	1✔
103	String::from("ID"),	1✔
104	FieldType::Int,	1✔
105	false,	1✔
106	SourceDefinition::Dynamic,	1✔
107	),	1✔
108	false,	1✔
109	)	1✔
110	.field(	1✔
111	FieldDefinition::new(	1✔
112	String::from("Salary"),	1✔
113	FieldType::Int,	1✔
114	false,	1✔
115	SourceDefinition::Dynamic,	1✔
116	),	1✔
117	false,	1✔
118	)	1✔
119	.clone();	1✔
120		1✔
121	let select = get_select("SELECT ID, SUM(Salary) as Salaries FROM Users GROUP BY ID").unwrap();	1✔
122		1✔
123	let factory = AggregationProcessorFactory::new(	1✔
124	NameOrAlias("Users".to_string(), None),	1✔
125	select.projection,	1✔
126	select.group_by,	1✔
127	false,	1✔
128	);	1✔
129	let out_schema = factory	1✔
130	.get_output_schema(	1✔
131	&DEFAULT_PORT_HANDLE,	1✔
132	&[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext::default()))]	1✔
133	.into_iter()	1✔
134	.collect(),	1✔
135	)	1✔
136	.unwrap()	1✔
137	.0;	1✔
138		1✔
139	assert_eq!(	1✔
140	out_schema,	1✔
141	Schema::empty()	1✔
142	.field(	1✔
143	FieldDefinition::new(	1✔
144	String::from("ID"),	1✔
145	FieldType::Int,	1✔
146	false,	1✔
147	SourceDefinition::Dynamic	1✔
148	),	1✔
149	true,	1✔
150	)	1✔
151	.field(	1✔
152	FieldDefinition::new(	1✔
153	String::from("Salaries"),	1✔
154	FieldType::Int,	1✔
155	false,	1✔
156	SourceDefinition::Dynamic	1✔
157	),	1✔
158	false,	1✔
159	)	1✔
160	.clone()	1✔
161	);	1✔
162	}	1✔

getdozer / dozer / 4007820649

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