6105410942

Committed 07 Sep 2023 04:28AM UTC coverage: 77.562% (-0.1%) from 77.686%

Build # 6105410942

Build Type

push

github

Committed by

chloeminkyung

Commit Message

feat: onnx image

Run Details

1141 of 1141 new or added lines in 66 files covered. (100.0%)

49957 of 64409 relevant lines covered (77.56%)

50900.25 hits per line

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

90.23

/dozer-sql/src/pipeline/tests/builder_test.rs

use dozer_core::app::{App, AppPipeline};
use dozer_core::appsource::{AppSourceManager, AppSourceMappings};
use dozer_core::channels::SourceChannelForwarder;
use dozer_core::checkpoint::create_checkpoint_factory_for_test;
use dozer_core::dozer_log::storage::Queue;
use dozer_core::epoch::Epoch;
use dozer_core::executor::{DagExecutor, ExecutorOptions};
use dozer_core::executor_operation::ProcessorOperation;
use dozer_core::node::{
    OutputPortDef, OutputPortType, PortHandle, Sink, SinkFactory, Source, SourceFactory,
};
use dozer_core::processor_record::ProcessorRecordStore;
use dozer_core::DEFAULT_PORT_HANDLE;
use dozer_types::errors::internal::BoxedError;
use dozer_types::ingestion_types::IngestionMessage;
use dozer_types::log::debug;
use dozer_types::node::OpIdentifier;
use dozer_types::ordered_float::OrderedFloat;
use dozer_types::types::{
    Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
};

use std::collections::HashMap;

use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use crate::pipeline::builder::statement_to_pipeline;

/// Test Source
#[derive(Debug)]
pub struct TestSourceFactory {
    output_ports: Vec<PortHandle>,
}

impl TestSourceFactory {
    pub fn new(output_ports: Vec<PortHandle>) -> Self {
        Self { output_ports }
    }
}

impl SourceFactory for TestSourceFactory {
    fn get_output_ports(&self) -> Vec<OutputPortDef> {
        self.output_ports
            .iter()
            .map(|e| OutputPortDef::new(*e, OutputPortType::Stateless))
            .collect()
    }

    fn get_output_schema(&self, _port: &PortHandle) -> Result<Schema, BoxedError> {
        Ok(Schema::default()
            .field(
                FieldDefinition::new(
                    String::from("CustomerID"),
                    FieldType::Int,
                    false,
                    SourceDefinition::Dynamic,
                ),
                false,
            )
            .field(
                FieldDefinition::new(
                    String::from("Country"),
                    FieldType::String,
                    false,
                    SourceDefinition::Dynamic,
                ),
                false,
            )
            .field(
                FieldDefinition::new(
                    String::from("Spending"),
                    FieldType::Float,
                    false,
                    SourceDefinition::Dynamic,
                ),
                false,
            )
            .clone())
    }

    fn get_output_port_name(&self, port: &PortHandle) -> String {
        format!("port_{}", port)
    }

    fn build(
        &self,
        _output_schemas: HashMap<PortHandle, Schema>,
    ) -> Result<Box<dyn Source>, BoxedError> {
        Ok(Box::new(TestSource {}))
    }
}

#[derive(Debug)]
pub struct TestSource {}

impl Source for TestSource {
    fn can_start_from(&self, _last_checkpoint: OpIdentifier) -> Result<bool, BoxedError> {
        Ok(false)
    }

    fn start(
        &self,
        fw: &mut dyn SourceChannelForwarder,
        _last_checkpoint: Option<OpIdentifier>,
    ) -> Result<(), BoxedError> {
        for n in 0..10000 {
            fw.send(
                IngestionMessage::new_op(
                    n,
                    0,
                    0,
                    Operation::Insert {
                        new: Record::new(vec![
                            Field::Int(0),
                            Field::String("Italy".to_string()),
                            Field::Float(OrderedFloat(5.5)),
                        ]),
                    },
                ),
                DEFAULT_PORT_HANDLE,
            )
            .unwrap();
        }
        Ok(())
    }
}

#[derive(Debug)]
pub struct TestSinkFactory {
    input_ports: Vec<PortHandle>,
}

impl TestSinkFactory {
    pub fn new(input_ports: Vec<PortHandle>) -> Self {
        Self { input_ports }
    }
}

impl SinkFactory for TestSinkFactory {
    fn get_input_ports(&self) -> Vec<PortHandle> {
        self.input_ports.clone()
    }

    fn build(
        &self,
        _input_schemas: HashMap<PortHandle, Schema>,
    ) -> Result<Box<dyn Sink>, BoxedError> {
        Ok(Box::new(TestSink {}))
    }

    fn prepare(&self, _input_schemas: HashMap<PortHandle, Schema>) -> Result<(), BoxedError> {
        Ok(())
    }
}

#[derive(Debug)]
pub struct TestSink {}

impl Sink for TestSink {
    fn process(
        &mut self,
        _from_port: PortHandle,
        _record_store: &ProcessorRecordStore,
        _op: ProcessorOperation,
    ) -> Result<(), BoxedError> {
        Ok(())
    }

    fn commit(&mut self, _epoch_details: &Epoch) -> Result<(), BoxedError> {
        Ok(())
    }

    fn persist(&mut self, _queue: &Queue) -> Result<(), BoxedError> {
        Ok(())
    }

    fn on_source_snapshotting_done(&mut self, _connection_name: String) -> Result<(), BoxedError> {
        Ok(())
    }
}

#[tokio::test]
async fn test_pipeline_builder() {
    let mut pipeline = AppPipeline::new_with_default_flags();
    let context = statement_to_pipeline(
        "SELECT COUNT(Spending), users.Country \
        FROM users \
         WHERE Spending >= 1",
        &mut pipeline,
        Some("results".to_string()),
        vec![],
    )
    .unwrap();

    let table_info = context.output_tables_map.get("results").unwrap();

    let mut asm = AppSourceManager::new();
    asm.add(
        Box::new(TestSourceFactory::new(vec![DEFAULT_PORT_HANDLE])),
        AppSourceMappings::new(
            "mem".to_string(),
            vec![("users".to_string(), DEFAULT_PORT_HANDLE)]
                .into_iter()
                .collect(),
        ),
    )
    .unwrap();

    pipeline.add_sink(
        Box::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])),
        "sink",
        None,
    );
    pipeline.connect_nodes(
        &table_info.node,
        table_info.port,
        "sink",
        DEFAULT_PORT_HANDLE,
    );

    let mut app = App::new(asm);
    app.add_pipeline(pipeline);

    let dag = app.into_dag().unwrap();

    let now = std::time::Instant::now();

    let (_temp_dir, checkpoint_factory, _) = create_checkpoint_factory_for_test(&[]).await;
    DagExecutor::new(
        dag,
        checkpoint_factory,
        Default::default(),
        ExecutorOptions::default(),
    )
    .await
    .unwrap()
    .start(Arc::new(AtomicBool::new(true)), Default::default())
    .unwrap()
    .join()
    .unwrap();

    let elapsed = now.elapsed();
    debug!("Elapsed: {:.2?}", elapsed);
}

1	use dozer_core::app::{App, AppPipeline};
2	use dozer_core::appsource::{AppSourceManager, AppSourceMappings};
3	use dozer_core::channels::SourceChannelForwarder;
4	use dozer_core::checkpoint::create_checkpoint_factory_for_test;
5	use dozer_core::dozer_log::storage::Queue;
6	use dozer_core::epoch::Epoch;
7	use dozer_core::executor::{DagExecutor, ExecutorOptions};
8	use dozer_core::executor_operation::ProcessorOperation;
9	use dozer_core::node::{
10	OutputPortDef, OutputPortType, PortHandle, Sink, SinkFactory, Source, SourceFactory,
11	};
12	use dozer_core::processor_record::ProcessorRecordStore;
13	use dozer_core::DEFAULT_PORT_HANDLE;
14	use dozer_types::errors::internal::BoxedError;
15	use dozer_types::ingestion_types::IngestionMessage;
16	use dozer_types::log::debug;
17	use dozer_types::node::OpIdentifier;
18	use dozer_types::ordered_float::OrderedFloat;
19	use dozer_types::types::{
20	Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
21	};
22
23	use std::collections::HashMap;
24
25	use std::sync::atomic::AtomicBool;
26	use std::sync::Arc;
27
28	use crate::pipeline::builder::statement_to_pipeline;
29
30	/// Test Source
31	#[derive(Debug)]	×
32	pub struct TestSourceFactory {
33	output_ports: Vec<PortHandle>,
34	}
35
36	impl TestSourceFactory {
37	pub fn new(output_ports: Vec<PortHandle>) -> Self {	1✔
38	Self { output_ports }	1✔
39	}	1✔
40	}
41
42	impl SourceFactory for TestSourceFactory {
43	fn get_output_ports(&self) -> Vec<OutputPortDef> {	3✔
44	self.output_ports	3✔
45	.iter()	3✔
46	.map(\|e\| OutputPortDef::new(*e, OutputPortType::Stateless))	3✔
47	.collect()	3✔
48	}	3✔
49
50	fn get_output_schema(&self, _port: &PortHandle) -> Result<Schema, BoxedError> {	1✔
51	Ok(Schema::default()	1✔
52	.field(	1✔
53	FieldDefinition::new(	1✔
54	String::from("CustomerID"),	1✔
55	FieldType::Int,	1✔
56	false,	1✔
57	SourceDefinition::Dynamic,	1✔
58	),	1✔
59	false,	1✔
60	)	1✔
61	.field(	1✔
62	FieldDefinition::new(	1✔
63	String::from("Country"),	1✔
64	FieldType::String,	1✔
65	false,	1✔
66	SourceDefinition::Dynamic,	1✔
67	),	1✔
68	false,	1✔
69	)	1✔
70	.field(	1✔
71	FieldDefinition::new(	1✔
72	String::from("Spending"),	1✔
73	FieldType::Float,	1✔
74	false,	1✔
75	SourceDefinition::Dynamic,	1✔
76	),	1✔
77	false,	1✔
78	)	1✔
79	.clone())	1✔
80	}	1✔
81
82	fn get_output_port_name(&self, port: &PortHandle) -> String {	×
83	format!("port_{}", port)	×
84	}	×
85
86	fn build(	1✔
87	&self,	1✔
88	_output_schemas: HashMap<PortHandle, Schema>,	1✔
89	) -> Result<Box<dyn Source>, BoxedError> {	1✔
90	Ok(Box::new(TestSource {}))	1✔
91	}	1✔
92	}
93
94	#[derive(Debug)]	×
95	pub struct TestSource {}
96
97	impl Source for TestSource {
98	fn can_start_from(&self, _last_checkpoint: OpIdentifier) -> Result<bool, BoxedError> {	×
99	Ok(false)	×
100	}	×
101
102	fn start(	1✔
103	&self,	1✔
104	fw: &mut dyn SourceChannelForwarder,	1✔
105	_last_checkpoint: Option<OpIdentifier>,	1✔
106	) -> Result<(), BoxedError> {	1✔
107	for n in 0..10000 {	10,001✔
108	fw.send(	10,000✔
109	IngestionMessage::new_op(	10,000✔
110	n,	10,000✔
111	0,	10,000✔
112	0,	10,000✔
113	Operation::Insert {	10,000✔
114	new: Record::new(vec![	10,000✔
115	Field::Int(0),	10,000✔
116	Field::String("Italy".to_string()),	10,000✔
117	Field::Float(OrderedFloat(5.5)),	10,000✔
118	]),	10,000✔
119	},	10,000✔
120	),	10,000✔
121	DEFAULT_PORT_HANDLE,	10,000✔
122	)	10,000✔
123	.unwrap();	10,000✔
124	}	10,000✔
125	Ok(())	1✔
126	}	1✔
127	}
128
129	#[derive(Debug)]	×
130	pub struct TestSinkFactory {
131	input_ports: Vec<PortHandle>,
132	}
133
134	impl TestSinkFactory {
135	pub fn new(input_ports: Vec<PortHandle>) -> Self {	1✔
136	Self { input_ports }	1✔
137	}	1✔
138	}
139
140	impl SinkFactory for TestSinkFactory {
141	fn get_input_ports(&self) -> Vec<PortHandle> {	4✔
142	self.input_ports.clone()	4✔
143	}	4✔
144
145	fn build(	1✔
146	&self,	1✔
147	_input_schemas: HashMap<PortHandle, Schema>,	1✔
148	) -> Result<Box<dyn Sink>, BoxedError> {	1✔
149	Ok(Box::new(TestSink {}))	1✔
150	}	1✔
151
152	fn prepare(&self, _input_schemas: HashMap<PortHandle, Schema>) -> Result<(), BoxedError> {	1✔
153	Ok(())	1✔
154	}	1✔
155	}
156
157	#[derive(Debug)]	×
158	pub struct TestSink {}
159
160	impl Sink for TestSink {
161	fn process(	10,000✔
162	&mut self,	10,000✔
163	_from_port: PortHandle,	10,000✔
164	_record_store: &ProcessorRecordStore,	10,000✔
165	_op: ProcessorOperation,	10,000✔
166	) -> Result<(), BoxedError> {	10,000✔
167	Ok(())	10,000✔
168	}	10,000✔
169
170	fn commit(&mut self, _epoch_details: &Epoch) -> Result<(), BoxedError> {	2✔
171	Ok(())	2✔
172	}	2✔
173
174	fn persist(&mut self, _queue: &Queue) -> Result<(), BoxedError> {	×
175	Ok(())	×
176	}	×
177
178	fn on_source_snapshotting_done(&mut self, _connection_name: String) -> Result<(), BoxedError> {	×
179	Ok(())	×
180	}	×
181	}
182
183	#[tokio::test]	1✔
184	async fn test_pipeline_builder() {	1✔
185	let mut pipeline = AppPipeline::new_with_default_flags();	1✔
186	let context = statement_to_pipeline(	1✔
187	"SELECT COUNT(Spending), users.Country \	1✔
188	FROM users \	1✔
189	WHERE Spending >= 1",	1✔
190	&mut pipeline,	1✔
191	Some("results".to_string()),	1✔
192	vec![],	1✔
193	)	1✔
194	.unwrap();	1✔
195		1✔
196	let table_info = context.output_tables_map.get("results").unwrap();	1✔
197		1✔
198	let mut asm = AppSourceManager::new();	1✔
199	asm.add(	1✔
200	Box::new(TestSourceFactory::new(vec![DEFAULT_PORT_HANDLE])),	1✔
201	AppSourceMappings::new(	1✔
202	"mem".to_string(),	1✔
203	vec![("users".to_string(), DEFAULT_PORT_HANDLE)]	1✔
204	.into_iter()	1✔
205	.collect(),	1✔
206	),	1✔
207	)	1✔
208	.unwrap();	1✔
209		1✔
210	pipeline.add_sink(	1✔
211	Box::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])),	1✔
212	"sink",	1✔
213	None,	1✔
214	);	1✔
215	pipeline.connect_nodes(	1✔
216	&table_info.node,	1✔
217	table_info.port,	1✔
218	"sink",	1✔
219	DEFAULT_PORT_HANDLE,	1✔
220	);	1✔
221		1✔
222	let mut app = App::new(asm);	1✔
223	app.add_pipeline(pipeline);	1✔
224		1✔
225	let dag = app.into_dag().unwrap();	1✔
226		1✔
227	let now = std::time::Instant::now();	1✔
228
229	let (_temp_dir, checkpoint_factory, _) = create_checkpoint_factory_for_test(&[]).await;	11✔
230	DagExecutor::new(	1✔
231	dag,	1✔
232	checkpoint_factory,	1✔
233	Default::default(),	1✔
234	ExecutorOptions::default(),	1✔
235	)	1✔
236	.await	×
237	.unwrap()	1✔
238	.start(Arc::new(AtomicBool::new(true)), Default::default())	1✔
239	.unwrap()	1✔
240	.join()	1✔
241	.unwrap();	1✔
242		1✔
243	let elapsed = now.elapsed();	1✔
244	debug!("Elapsed: {:.2?}", elapsed);	1✔
245	}

getdozer / dozer / 6105410942

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