3978628498

Build Type

github

Committed by GitHub

Commit Message

Merge 8775fcda7 into e2f9ad287

Pull Request Pull Request #705: chore: support for generic schema context in `Sink`, `Processor` and `Source` factories

Run Details

572 of 572 new or added lines in 35 files covered. (100.0%)

22294 of 34850 relevant lines covered (63.97%)

40332.28 hits per line

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

97.4

/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::{
    init_input_schema, init_processor, FIELD_100_INT,
};
use crate::pipeline::builder::SchemaSQLContext;
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};
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_aggregation_alias() {
    let schema = Schema::empty()
        .field(
            FieldDefinition::new(String::from("ID"), FieldType::Int, false),
            false,
        )
        .field(
            FieldDefinition::new(String::from("Salary"), FieldType::Int, false),
            false,
        )
        .clone();

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

    let factory = AggregationProcessorFactory::new(select.projection, select.group_by, false);
    let out_schema = factory
        .get_output_schema(
            &DEFAULT_PORT_HANDLE,
            &[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext {}))]
                .into_iter()
                .collect(),
        )
        .unwrap()
        .0;

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

1	use crate::output;
2	use crate::pipeline::aggregation::factory::AggregationProcessorFactory;
3	use crate::pipeline::aggregation::tests::aggregation_tests_utils::{
4	init_input_schema, init_processor, FIELD_100_INT,
5	};
6	use crate::pipeline::builder::SchemaSQLContext;
7	use crate::pipeline::tests::utils::get_select;
8	use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
9	use dozer_core::dag::node::ProcessorFactory;
10	use dozer_types::types::FieldType::Int;
11	use dozer_types::types::{Field, FieldDefinition, FieldType, Operation, Record, Schema};
12	use std::collections::HashMap;
13		×
14	#[test]	1✔
15	fn test_sum_aggregation_null() {	1✔
16	let schema = init_input_schema(Int, "SUM");	1✔
17	let (processor, tx) = init_processor(	1✔
18	"SELECT Country, SUM(Salary) \	1✔
19	FROM Users \	1✔
20	WHERE Salary >= 1 GROUP BY Country",	1✔
21	HashMap::from([(DEFAULT_PORT_HANDLE, schema)]),	1✔
22	)	1✔
23	.unwrap();	1✔
24		1✔
25	// Insert 100 for segment Italy	1✔
26	/*	1✔
27	NULL, 100.0	1✔
28	-------------	1✔
29	SUM = 100.0	1✔
30	*/	1✔
31	let inp = Operation::Insert {	1✔
32	new: Record::new(	1✔
33	None,	1✔
34	vec![	1✔
35	Field::Int(0),	1✔
36	Field::Null,	1✔
37	FIELD_100_INT.clone(),	1✔
38	FIELD_100_INT.clone(),	1✔
39	],	1✔
40	None,	1✔
41	),	1✔
42	};	1✔
43	let out = output!(processor, inp, tx);	1✔
44	let exp = vec![Operation::Insert {	1✔
45	new: Record::new(None, vec![Field::Null, FIELD_100_INT.clone()], None),	1✔
46	}];	1✔
47	assert_eq!(out, exp);	1✔
48	}	1✔
49		×
50	#[test]	1✔
51	fn test_aggregation_alias() {	1✔
52	let schema = Schema::empty()	1✔
53	.field(	1✔
54	FieldDefinition::new(String::from("ID"), FieldType::Int, false),	1✔
55	false,	1✔
56	)	1✔
57	.field(	1✔
58	FieldDefinition::new(String::from("Salary"), FieldType::Int, false),	1✔
59	false,	1✔
60	)	1✔
61	.clone();	1✔
62		1✔
63	let select = get_select("SELECT ID, SUM(Salary) as Salaries FROM Users GROUP BY ID").unwrap();	1✔
64		1✔
65	let factory = AggregationProcessorFactory::new(select.projection, select.group_by, false);	1✔
66	let out_schema = factory	1✔
67	.get_output_schema(	1✔
68	&DEFAULT_PORT_HANDLE,	1✔
69	&[(DEFAULT_PORT_HANDLE, (schema, SchemaSQLContext {}))]	1✔
70	.into_iter()	1✔
71	.collect(),	1✔
72	)	1✔
73	.unwrap()	1✔
74	.0;	1✔
75		1✔
76	assert_eq!(	1✔
77	out_schema,	1✔
78	Schema::empty()	1✔
79	.field(	1✔
80	FieldDefinition::new(String::from("ID"), FieldType::Int, false),	1✔
81	true,	1✔
82	)	1✔
83	.field(	1✔
84	FieldDefinition::new(String::from("Salaries"), FieldType::Int, false),	1✔
85	false,	1✔
86	)	1✔
87	.clone()	1✔
88	);	1✔
89	}	1✔

getdozer / dozer / 3978628498

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