6010722164

Committed 29 Aug 2023 10:20AM UTC coverage: 76.689% (-1.4%) from 78.07%

Build # 6010722164

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Create unit tests workflow (#1910)

* chore: Update for Rust 1.72.0

Rust 1.72.0 has introduced a bunch of new lints. Here we fix them all.

`let ... else` finally gets formatted.

* chire: Create unit tests workflow

* Rename and remove useless steps

* remove env vars

* Add concurrency group

* Test unit workflow on 4 cores

* Add mysql service to unit tests

---------

Co-authored-by: chubei <914745487@qq.com>

Run Details

49006 of 63902 relevant lines covered (76.69%)

48444.14 hits per line

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

53.25

/dozer-sql/src/pipeline/product/table/factory.rs

use std::collections::HashMap;

use dozer_core::{
    node::{OutputPortDef, OutputPortType, PortHandle, Processor, ProcessorFactory},
    processor_record::ProcessorRecordStore,
    DEFAULT_PORT_HANDLE,
};
use dozer_types::{errors::internal::BoxedError, types::Schema};
use sqlparser::ast::TableFactor;

use crate::pipeline::{
    errors::{PipelineError, ProductError},
    expression::builder::extend_schema_source_def,
};
use crate::pipeline::{
    expression::builder::NameOrAlias, window::builder::string_from_sql_object_name,
};

use super::processor::TableProcessor;

#[derive(Debug)]
pub struct TableProcessorFactory {
    id: String,
    relation: TableFactor,
}

impl TableProcessorFactory {
    pub fn new(id: String, relation: TableFactor) -> Self {
        Self { id, relation }
    }
}

impl ProcessorFactory for TableProcessorFactory {
    fn id(&self) -> String {
        self.id.clone()
    }

    fn type_name(&self) -> String {
        "Table".to_string()
    }

    fn get_input_ports(&self) -> Vec<PortHandle> {
        vec![DEFAULT_PORT_HANDLE]
    }

    fn get_output_ports(&self) -> Vec<OutputPortDef> {
        vec![OutputPortDef::new(
            DEFAULT_PORT_HANDLE,
            OutputPortType::Stateless,
        )]
    }

    fn get_output_schema(
        &self,
        _output_port: &PortHandle,
        input_schemas: &HashMap<PortHandle, Schema>,
    ) -> Result<Schema, BoxedError> {
        if let Some(input_schema) = input_schemas.get(&DEFAULT_PORT_HANDLE) {
            let table = get_name_or_alias(&self.relation)?;
            let extended_input_schema = extend_schema_source_def(input_schema, &table);
            Ok(extended_input_schema)
        } else {
            Err(PipelineError::InvalidPortHandle(DEFAULT_PORT_HANDLE).into())
        }
    }

    fn build(
        &self,
        _input_schemas: HashMap<PortHandle, dozer_types::types::Schema>,
        _output_schemas: HashMap<PortHandle, dozer_types::types::Schema>,
        _record_store: &ProcessorRecordStore,
    ) -> Result<Box<dyn Processor>, BoxedError> {
        Ok(Box::new(TableProcessor::new(self.id.clone())))
    }
}

pub fn get_name_or_alias(relation: &TableFactor) -> Result<NameOrAlias, PipelineError> {
    match relation {
        TableFactor::Table { name, alias, .. } => {
            let table_name = string_from_sql_object_name(name);
            if let Some(table_alias) = alias {
                let alias = table_alias.name.value.clone();
                return Ok(NameOrAlias(table_name, Some(alias)));
            }
            Ok(NameOrAlias(table_name, None))
        }
        TableFactor::Derived { alias, .. } => {
            if let Some(table_alias) = alias {
                let alias = table_alias.name.value.clone();
                return Ok(NameOrAlias("dozer_derived".to_string(), Some(alias)));
            }
            Ok(NameOrAlias("dozer_derived".to_string(), None))
        }
        TableFactor::TableFunction { .. } => Err(PipelineError::ProductError(
            ProductError::UnsupportedTableFunction,
        )),
        TableFactor::UNNEST { .. } => {
            Err(PipelineError::ProductError(ProductError::UnsupportedUnnest))
        }
        TableFactor::NestedJoin { alias, .. } => {
            if let Some(table_alias) = alias {
                let alias = table_alias.name.value.clone();
                return Ok(NameOrAlias("dozer_nested".to_string(), Some(alias)));
            }
            Ok(NameOrAlias("dozer_nested".to_string(), None))
        }
        TableFactor::Pivot { .. } => {
            Err(PipelineError::ProductError(ProductError::UnsupportedPivot))
        }
    }
}

1	use std::collections::HashMap;
2
3	use dozer_core::{
4	node::{OutputPortDef, OutputPortType, PortHandle, Processor, ProcessorFactory},
5	processor_record::ProcessorRecordStore,
6	DEFAULT_PORT_HANDLE,
7	};
8	use dozer_types::{errors::internal::BoxedError, types::Schema};
9	use sqlparser::ast::TableFactor;
10
11	use crate::pipeline::{
12	errors::{PipelineError, ProductError},
13	expression::builder::extend_schema_source_def,
14	};
15	use crate::pipeline::{
16	expression::builder::NameOrAlias, window::builder::string_from_sql_object_name,
17	};
18
19	use super::processor::TableProcessor;
20
21	#[derive(Debug)]	×
22	pub struct TableProcessorFactory {	×
23	id: String,
24	relation: TableFactor,
25	}
26
27	impl TableProcessorFactory {
28	pub fn new(id: String, relation: TableFactor) -> Self {	464✔
29	Self { id, relation }	464✔
30	}	464✔
31	}	×
32
33	impl ProcessorFactory for TableProcessorFactory {
34	fn id(&self) -> String {	×
35	self.id.clone()	×
36	}	×
37		×
38	fn type_name(&self) -> String {	×
39	"Table".to_string()	×
40	}	×
41		×
42	fn get_input_ports(&self) -> Vec<PortHandle> {	1,768✔
43	vec![DEFAULT_PORT_HANDLE]	1,768✔
44	}	1,768✔
45		×
46	fn get_output_ports(&self) -> Vec<OutputPortDef> {	1,326✔
47	vec![OutputPortDef::new(	1,326✔
48	DEFAULT_PORT_HANDLE,	1,326✔
49	OutputPortType::Stateless,	1,326✔
50	)]	1,326✔
51	}	1,326✔
52		×
53	fn get_output_schema(
54	&self,
55	_output_port: &PortHandle,
56	input_schemas: &HashMap<PortHandle, Schema>,
57	) -> Result<Schema, BoxedError> {
58	if let Some(input_schema) = input_schemas.get(&DEFAULT_PORT_HANDLE) {	442✔
59	let table = get_name_or_alias(&self.relation)?;	442✔
60	let extended_input_schema = extend_schema_source_def(input_schema, &table);	442✔
61	Ok(extended_input_schema)	442✔
62	} else {	×
63	Err(PipelineError::InvalidPortHandle(DEFAULT_PORT_HANDLE).into())	×
64	}	×
65	}	442✔
66		×
67	fn build(	442✔
68	&self,	442✔
69	_input_schemas: HashMap<PortHandle, dozer_types::types::Schema>,	442✔
70	_output_schemas: HashMap<PortHandle, dozer_types::types::Schema>,	442✔
71	_record_store: &ProcessorRecordStore,	442✔
72	) -> Result<Box<dyn Processor>, BoxedError> {	442✔
73	Ok(Box::new(TableProcessor::new(self.id.clone())))	442✔
74	}	442✔
75	}	×
76
77	pub fn get_name_or_alias(relation: &TableFactor) -> Result<NameOrAlias, PipelineError> {	977✔
78	match relation {	977✔
79	TableFactor::Table { name, alias, .. } => {	942✔
80	let table_name = string_from_sql_object_name(name);	942✔
81	if let Some(table_alias) = alias {	942✔
82	let alias = table_alias.name.value.clone();	344✔
83	return Ok(NameOrAlias(table_name, Some(alias)));	344✔
84	}	598✔
85	Ok(NameOrAlias(table_name, None))	598✔
86	}	×
87	TableFactor::Derived { alias, .. } => {	35✔
88	if let Some(table_alias) = alias {	35✔
89	let alias = table_alias.name.value.clone();	28✔
90	return Ok(NameOrAlias("dozer_derived".to_string(), Some(alias)));	28✔
91	}	7✔
92	Ok(NameOrAlias("dozer_derived".to_string(), None))	7✔
93	}	×
94	TableFactor::TableFunction { .. } => Err(PipelineError::ProductError(	×
95	ProductError::UnsupportedTableFunction,	×
96	)),	×
97	TableFactor::UNNEST { .. } => {	×
98	Err(PipelineError::ProductError(ProductError::UnsupportedUnnest))	×
99	}	×
100	TableFactor::NestedJoin { alias, .. } => {	×
101	if let Some(table_alias) = alias {	×
102	let alias = table_alias.name.value.clone();	×
103	return Ok(NameOrAlias("dozer_nested".to_string(), Some(alias)));	×
104	}	×
105	Ok(NameOrAlias("dozer_nested".to_string(), None))	×
106	}	×
107	TableFactor::Pivot { .. } => {
108	Err(PipelineError::ProductError(ProductError::UnsupportedPivot))	×
109	}	×
110	}
111	}	977✔

getdozer / dozer / 6010722164

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