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

76.35

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

use dozer_core::dag::app::App;
use dozer_core::dag::appsource::{AppSource, AppSourceManager};
use dozer_core::dag::channels::SourceChannelForwarder;
use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
use dozer_core::dag::errors::ExecutionError;
use dozer_core::dag::executor::{DagExecutor, ExecutorOptions};
use dozer_core::dag::node::{
    OutputPortDef, OutputPortType, PortHandle, Sink, SinkFactory, Source, SourceFactory,
};
use dozer_core::dag::record_store::RecordReader;
use dozer_core::storage::lmdb_storage::{LmdbEnvironmentManager, SharedTransaction};
use dozer_types::log::debug;
use dozer_types::ordered_float::OrderedFloat;
use dozer_types::types::{Field, FieldDefinition, FieldType, Operation, Record, Schema};

use dozer_core::dag::epoch::Epoch;

use std::collections::HashMap;
use std::fs;

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

use crate::pipeline::builder::{statement_to_pipeline, SchemaSQLContext};

/// 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<SchemaSQLContext> for TestSourceFactory {
    fn get_output_ports(&self) -> Result<Vec<OutputPortDef>, ExecutionError> {
        Ok(self
            .output_ports
            .iter()
            .map(|e| OutputPortDef::new(*e, OutputPortType::Stateless))
            .collect())
    }

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

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

    fn prepare(
        &self,
        _output_schemas: HashMap<PortHandle, (Schema, SchemaSQLContext)>,
    ) -> Result<(), ExecutionError> {
        Ok(())
    }
}

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

impl Source for TestSource {
    fn start(
        &self,
        fw: &mut dyn SourceChannelForwarder,
        _from_seq: Option<(u64, u64)>,
    ) -> Result<(), ExecutionError> {
        for n in 0..10000 {
            fw.send(
                n,
                0,
                Operation::Insert {
                    new: Record::new(
                        None,
                        vec![
                            Field::Int(0),
                            Field::String("Italy".to_string()),
                            Field::Float(OrderedFloat(5.5)),
                        ],
                        None,
                    ),
                },
                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<SchemaSQLContext> for TestSinkFactory {
    fn get_input_ports(&self) -> Vec<PortHandle> {
        self.input_ports.clone()
    }

    fn set_input_schema(
        &self,
        _input_schemas: &HashMap<PortHandle, (Schema, SchemaSQLContext)>,
    ) -> Result<(), ExecutionError> {
        Ok(())
    }

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

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

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

impl Sink for TestSink {
    fn init(&mut self, _env: &mut LmdbEnvironmentManager) -> Result<(), ExecutionError> {
        debug!("SINK: Initialising TestSink");
        Ok(())
    }

    fn process(
        &mut self,
        _from_port: PortHandle,
        _op: Operation,
        _state: &SharedTransaction,
        _reader: &HashMap<PortHandle, Box<dyn RecordReader>>,
    ) -> Result<(), ExecutionError> {
        Ok(())
    }

    fn commit(&mut self, _epoch: &Epoch, _tx: &SharedTransaction) -> Result<(), ExecutionError> {
        Ok(())
    }
}

#[test]
fn test_pipeline_builder() {
    let (mut pipeline, (node, node_port)) = statement_to_pipeline(
        "SELECT COUNT(Spending), users.Country \
    FROM users \
    WHERE Spending >= 1",
    )
    .unwrap();

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

    pipeline.add_sink(
        Arc::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])),
        "sink",
    );
    pipeline
        .connect_nodes(&node, Some(node_port), "sink", Some(DEFAULT_PORT_HANDLE))
        .unwrap();

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

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

    let tmp_dir = TempDir::new("example").unwrap_or_else(|_e| panic!("Unable to create temp dir"));
    if tmp_dir.path().exists() {
        fs::remove_dir_all(tmp_dir.path()).unwrap_or_else(|_e| panic!("Unable to remove old dir"));
    }
    fs::create_dir(tmp_dir.path()).unwrap_or_else(|_e| panic!("Unable to create temp dir"));

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

    let tmp_dir = TempDir::new("test").unwrap();
    let mut executor = DagExecutor::new(
        &dag,
        tmp_dir.path(),
        ExecutorOptions::default(),
        Arc::new(AtomicBool::new(true)),
    )
    .unwrap();

    executor
        .start()
        .unwrap_or_else(|e| panic!("Unable to start the Executor: {}", e));
    assert!(executor.join().is_ok());

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

1	use dozer_core::dag::app::App;
2	use dozer_core::dag::appsource::{AppSource, AppSourceManager};
3	use dozer_core::dag::channels::SourceChannelForwarder;
4	use dozer_core::dag::dag::DEFAULT_PORT_HANDLE;
5	use dozer_core::dag::errors::ExecutionError;
6	use dozer_core::dag::executor::{DagExecutor, ExecutorOptions};
7	use dozer_core::dag::node::{
8	OutputPortDef, OutputPortType, PortHandle, Sink, SinkFactory, Source, SourceFactory,
9	};
10	use dozer_core::dag::record_store::RecordReader;
11	use dozer_core::storage::lmdb_storage::{LmdbEnvironmentManager, SharedTransaction};
12	use dozer_types::log::debug;
13	use dozer_types::ordered_float::OrderedFloat;
14	use dozer_types::types::{Field, FieldDefinition, FieldType, Operation, Record, Schema};
15
16	use dozer_core::dag::epoch::Epoch;
17
18	use std::collections::HashMap;
19	use std::fs;
20
21	use std::sync::atomic::AtomicBool;
22	use std::sync::Arc;
23	use tempdir::TempDir;
24
25	use crate::pipeline::builder::{statement_to_pipeline, SchemaSQLContext};
26
27	/// Test Source	×
28	#[derive(Debug)]	×
29	pub struct TestSourceFactory {
30	output_ports: Vec<PortHandle>,
31	}
32
33	impl TestSourceFactory {	×
34	pub fn new(output_ports: Vec<PortHandle>) -> Self {	1✔
35	Self { output_ports }	1✔
36	}	1✔
37	}
38
39	impl SourceFactory<SchemaSQLContext> for TestSourceFactory {	×
40	fn get_output_ports(&self) -> Result<Vec<OutputPortDef>, ExecutionError> {	3✔
41	Ok(self	3✔
42	.output_ports	3✔
43	.iter()	3✔
44	.map(\|e\| OutputPortDef::new(*e, OutputPortType::Stateless))	3✔
45	.collect())	3✔
46	}	3✔
47		×
48	fn get_output_schema(	1✔
49	&self,	1✔
50	_port: &PortHandle,	1✔
51	) -> Result<(Schema, SchemaSQLContext), ExecutionError> {	1✔
52	Ok((	1✔
53	Schema::empty()	1✔
54	.field(	1✔
55	FieldDefinition::new(String::from("CustomerID"), FieldType::Int, false),	1✔
56	false,	1✔
57	)	1✔
58	.field(	1✔
59	FieldDefinition::new(String::from("Country"), FieldType::String, false),	1✔
60	false,	1✔
61	)	1✔
62	.field(	1✔
63	FieldDefinition::new(String::from("Spending"), FieldType::Float, false),	1✔
64	false,	1✔
65	)	1✔
66	.clone(),	1✔
67	SchemaSQLContext {},	1✔
68	))	1✔
69	}	1✔
70		×
71	fn build(	1✔
72	&self,	1✔
73	_output_schemas: HashMap<PortHandle, Schema>,	1✔
74	) -> Result<Box<dyn Source>, ExecutionError> {	1✔
75	Ok(Box::new(TestSource {}))	1✔
76	}	1✔
77		×
78	fn prepare(	×
79	&self,	×
80	_output_schemas: HashMap<PortHandle, (Schema, SchemaSQLContext)>,	×
81	) -> Result<(), ExecutionError> {	×
82	Ok(())	×
83	}	×
84	}	×
85		×
86	#[derive(Debug)]	×
87	pub struct TestSource {}	×
88		×
89	impl Source for TestSource {	×
90	fn start(	1✔
91	&self,	1✔
92	fw: &mut dyn SourceChannelForwarder,	1✔
93	_from_seq: Option<(u64, u64)>,	1✔
94	) -> Result<(), ExecutionError> {	1✔
95	for n in 0..10000 {	10,001✔
96	fw.send(	10,000✔
97	n,	10,000✔
98	0,	10,000✔
99	Operation::Insert {	10,000✔
100	new: Record::new(	10,000✔
101	None,	10,000✔
102	vec![	10,000✔
103	Field::Int(0),	10,000✔
104	Field::String("Italy".to_string()),	10,000✔
105	Field::Float(OrderedFloat(5.5)),	10,000✔
106	],	10,000✔
107	None,	10,000✔
108	),	10,000✔
109	},	10,000✔
110	DEFAULT_PORT_HANDLE,	10,000✔
111	)	10,000✔
112	.unwrap();	10,000✔
113	}	10,000✔
114	Ok(())	1✔
115	}	1✔
116	}	×
117		×
118	#[derive(Debug)]	×
119	pub struct TestSinkFactory {
120	input_ports: Vec<PortHandle>,	×
121	}	×
122		×
123	impl TestSinkFactory {	×
124	pub fn new(input_ports: Vec<PortHandle>) -> Self {	1✔
125	Self { input_ports }	1✔
126	}	1✔
127	}	×
128		×
129	impl SinkFactory<SchemaSQLContext> for TestSinkFactory {	×
130	fn get_input_ports(&self) -> Vec<PortHandle> {	2✔
131	self.input_ports.clone()	2✔
132	}	2✔
133		×
134	fn set_input_schema(	1✔
135	&self,	1✔
136	_input_schemas: &HashMap<PortHandle, (Schema, SchemaSQLContext)>,	1✔
137	) -> Result<(), ExecutionError> {	1✔
138	Ok(())	1✔
139	}	1✔
140		×
141	fn build(	1✔
142	&self,	1✔
143	_input_schemas: HashMap<PortHandle, Schema>,	1✔
144	) -> Result<Box<dyn Sink>, ExecutionError> {	1✔
145	Ok(Box::new(TestSink {}))	1✔
146	}	1✔
147		×
148	fn prepare(	×
149	&self,	×
150	_input_schemas: HashMap<PortHandle, (Schema, SchemaSQLContext)>,	×
151	) -> Result<(), ExecutionError> {	×
152	Ok(())	×
153	}	×
154	}	×
155		×
156	#[derive(Debug)]	×
157	pub struct TestSink {}	×
158		×
159	impl Sink for TestSink {	×
160	fn init(&mut self, _env: &mut LmdbEnvironmentManager) -> Result<(), ExecutionError> {	1✔
161	debug!("SINK: Initialising TestSink");	1✔
162	Ok(())	1✔
163	}	1✔
164		×
165	fn process(	10,000✔
166	&mut self,	10,000✔
167	_from_port: PortHandle,	10,000✔
168	_op: Operation,	10,000✔
169	_state: &SharedTransaction,	10,000✔
170	_reader: &HashMap<PortHandle, Box<dyn RecordReader>>,	10,000✔
171	) -> Result<(), ExecutionError> {	10,000✔
172	Ok(())	10,000✔
173	}	10,000✔
174		×
175	fn commit(&mut self, _epoch: &Epoch, _tx: &SharedTransaction) -> Result<(), ExecutionError> {	5✔
176	Ok(())	5✔
177	}	5✔
178	}	×
179		×
180	#[test]	1✔
181	fn test_pipeline_builder() {	1✔
182	let (mut pipeline, (node, node_port)) = statement_to_pipeline(	1✔
183	"SELECT COUNT(Spending), users.Country \	1✔
184	FROM users \	1✔
185	WHERE Spending >= 1",	1✔
186	)	1✔
187	.unwrap();	1✔
188		1✔
189	let mut asm = AppSourceManager::new();	1✔
190	asm.add(AppSource::new(	1✔
191	"mem".to_string(),	1✔
192	Arc::new(TestSourceFactory::new(vec![DEFAULT_PORT_HANDLE])),	1✔
193	vec![("users".to_string(), DEFAULT_PORT_HANDLE)]	1✔
194	.into_iter()	1✔
195	.collect(),	1✔
196	))	1✔
197	.unwrap();	1✔
198		1✔
199	pipeline.add_sink(	1✔
200	Arc::new(TestSinkFactory::new(vec![DEFAULT_PORT_HANDLE])),	1✔
201	"sink",	1✔
202	);	1✔
203	pipeline	1✔
204	.connect_nodes(&node, Some(node_port), "sink", Some(DEFAULT_PORT_HANDLE))	1✔
205	.unwrap();	1✔
206		1✔
207	let mut app = App::new(asm);	1✔
208	app.add_pipeline(pipeline);	1✔
209		1✔
210	let dag = app.get_dag().unwrap();	1✔
211		1✔
212	let tmp_dir = TempDir::new("example").unwrap_or_else(\|_e\| panic!("Unable to create temp dir"));	1✔
213	if tmp_dir.path().exists() {	1✔
214	fs::remove_dir_all(tmp_dir.path()).unwrap_or_else(\|_e\| panic!("Unable to remove old dir"));	1✔
215	}	1✔
216	fs::create_dir(tmp_dir.path()).unwrap_or_else(\|_e\| panic!("Unable to create temp dir"));	1✔
217		1✔
218	use std::time::Instant;	1✔
219	let now = Instant::now();	1✔
220		1✔
221	let tmp_dir = TempDir::new("test").unwrap();	1✔
222	let mut executor = DagExecutor::new(	1✔
223	&dag,	1✔
224	tmp_dir.path(),	1✔
225	ExecutorOptions::default(),	1✔
226	Arc::new(AtomicBool::new(true)),	1✔
227	)	1✔
228	.unwrap();	1✔
229		1✔
230	executor	1✔
231	.start()	1✔
232	.unwrap_or_else(\|e\| panic!("Unable to start the Executor: {}", e));	1✔
233	assert!(executor.join().is_ok());	1✔
234
235	let elapsed = now.elapsed();	1✔
236	debug!("Elapsed: {:.2?}", elapsed);	1✔
237	}	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