3983272635

Build Type

push

github

Committed by GitHub

Commit Message

fix insert after delete problem in aggregation (#709)

Run Details

52 of 52 new or added lines in 2 files covered. (100.0%)

22141 of 34542 relevant lines covered (64.1%)

46204.35 hits per line

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

89.05

/dozer-core/src/dag/executor/source_node.rs

use std::{
    collections::HashMap,
    path::Path,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};

use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
use dozer_types::log::debug;
use dozer_types::{
    internal_err,
    parking_lot::RwLock,
    types::{Operation, Schema},
};

use crate::dag::{
    channels::SourceChannelForwarder,
    dag::Edge,
    epoch::EpochManager,
    errors::ExecutionError::{self, InternalError},
    executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
    forwarder::{SourceChannelManager, StateWriter},
    node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
    record_store::RecordReader,
};

use super::{node::Node, ExecutorOperation};

#[derive(Debug)]
struct InternalChannelSourceForwarder {
    sender: Sender<(PortHandle, u64, u64, Operation)>,
}

impl InternalChannelSourceForwarder {
    pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {
        Self { sender }
    }
}

impl SourceChannelForwarder for InternalChannelSourceForwarder {
    fn send(
        &mut self,
        txid: u64,
        seq_in_tx: u64,
        op: Operation,
        port: PortHandle,
    ) -> Result<(), ExecutionError> {
        internal_err!(self.sender.send((port, txid, seq_in_tx, op)))
    }
}

/// The sender half of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceSenderNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// The source.
    source: Box<dyn Source>,
    /// Last checkpointed output data sequence number.
    last_checkpoint: (u64, u64),
    /// The forwarder that will be passed to the source for outputig data.
    forwarder: InternalChannelSourceForwarder,
    /// If the execution DAG should be running. Used for terminating the execution DAG.
    running: Arc<AtomicBool>,
}

impl SourceSenderNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `source_factory`: Source factory in description DAG.
    /// - `output_schemas`: Output data schemas.
    /// - `last_checkpoint`: Last checkpointed output of this source.
    /// - `sender`: Channel to send data to.
    /// - `running`: If the execution DAG should still be running.
    pub fn new<T: Clone>(
        node_handle: NodeHandle,
        source_factory: &dyn SourceFactory<T>,
        output_schemas: HashMap<PortHandle, Schema>,
        last_checkpoint: (u64, u64),
        sender: Sender<(PortHandle, u64, u64, Operation)>,
        running: Arc<AtomicBool>,
    ) -> Result<Self, ExecutionError> {
        let source = source_factory.build(output_schemas)?;
        let forwarder = InternalChannelSourceForwarder::new(sender);
        Ok(Self {
            node_handle,
            source,
            last_checkpoint,
            forwarder,
            running,
        })
    }
}

impl Node for SourceSenderNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        let result = self
            .source
            .start(&mut self.forwarder, Some(self.last_checkpoint));
        self.running.store(false, Ordering::SeqCst);
        debug!("[{}-sender] Quit", self.node_handle);
        result
    }
}

/// The listener part of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceListenerNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// Output from corresponding source sender.
    receiver: Receiver<(PortHandle, u64, u64, Operation)>,
    /// Receiving timeout.
    timeout: Duration,
    /// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
    running: Arc<AtomicBool>,
    /// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
    channel_manager: SourceChannelManager,
}

impl SourceListenerNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `receiver`: Channel that the data comes in.
    /// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
    /// - `base_path`: Base path of persisted data for the last execution of the description DAG.
    /// - `output_ports`: Output port definition of the source in description DAG.
    /// - `record_readers`: Record readers of all stateful ports.
    /// - `senders`: Output channels from this processor.
    /// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
    /// - `running`: If the execution DAG should still be running.
    /// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
    /// - `output_schemas`: Output data schemas.
    /// - `retention_queue_size`: Size of retention queue (used by RecordWriter)
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        node_handle: NodeHandle,
        receiver: Receiver<(PortHandle, u64, u64, Operation)>,
        timeout: Duration,
        base_path: &Path,
        output_ports: &[OutputPortDef],
        record_readers: Arc<
            RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,
        >,
        senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,
        edges: &[Edge],
        running: Arc<AtomicBool>,
        commit_sz: u32,
        max_duration_between_commits: Duration,
        epoch_manager: Arc<EpochManager>,
        output_schemas: HashMap<PortHandle, Schema>,
        start_seq: (u64, u64),
        retention_queue_size: usize,
    ) -> Result<Self, ExecutionError> {
        let state_meta = init_component(&node_handle, base_path, |_| Ok(()))?;
        let (master_tx, port_databases) =
            create_ports_databases_and_fill_downstream_record_readers(
                &node_handle,
                edges,
                state_meta.env,
                output_ports,
                &mut record_readers.write(),
            )?;
        let channel_manager = SourceChannelManager::new(
            node_handle.clone(),
            senders,
            StateWriter::new(
                state_meta.meta_db,
                port_databases,
                master_tx,
                output_schemas,
                retention_queue_size,
            )?,
            true,
            commit_sz,
            max_duration_between_commits,
            epoch_manager,
            start_seq,
        );
        Ok(Self {
            node_handle,
            receiver,
            timeout,
            running,
            channel_manager,
        })
    }
}

impl SourceListenerNode {
    /// Returns if the node should terminate.
    fn send_and_trigger_commit_if_needed(
        &mut self,
        data: Option<(PortHandle, u64, u64, Operation)>,
    ) -> Result<bool, ExecutionError> {
        // First check if termination was requested.
        let terminating = !self.running.load(Ordering::SeqCst);
        // If this commit was not requested with termination at the start, we shouldn't terminate either.
        let terminating = match data {
            Some((port, txid, seq_in_tx, op)) => self
                .channel_manager
                .send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,
            None => self.channel_manager.trigger_commit_if_needed(terminating)?,
        };
        if terminating {
            self.channel_manager.terminate()?;
            debug!("[{}-listener] Quitting", &self.node_handle);
        }
        Ok(terminating)
    }
}

impl Node for SourceListenerNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        loop {
            match self.receiver.recv_timeout(self.timeout) {
                Ok(data) => {
                    if self.send_and_trigger_commit_if_needed(Some(data))? {
                        return Ok(());
                    }
                }
                Err(e) => {
                    if self.send_and_trigger_commit_if_needed(None)? {
                        return Ok(());
                    }
                    // Channel disconnected but running flag not set to false, the source sender must have panicked.
                    if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {
                        return Err(ExecutionError::ChannelDisconnected);
                    }
                }
            }
        }
    }
}

1	use std::{
2	collections::HashMap,
3	path::Path,
4	sync::{
5	atomic::{AtomicBool, Ordering},
6	Arc,
7	},
8	time::Duration,
9	};
10
11	use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
12	use dozer_types::log::debug;
13	use dozer_types::{
14	internal_err,
15	parking_lot::RwLock,
16	types::{Operation, Schema},
17	};
18
19	use crate::dag::{
20	channels::SourceChannelForwarder,
21	dag::Edge,
22	epoch::EpochManager,
23	errors::ExecutionError::{self, InternalError},
24	executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
25	forwarder::{SourceChannelManager, StateWriter},
26	node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
27	record_store::RecordReader,
28	};
29
30	use super::{node::Node, ExecutorOperation};
31
32	#[derive(Debug)]	×
33	struct InternalChannelSourceForwarder {
34	sender: Sender<(PortHandle, u64, u64, Operation)>,
35	}
36
37	impl InternalChannelSourceForwarder {
38	pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {	212✔
39	Self { sender }	212✔
40	}	212✔
41	}
42
43	impl SourceChannelForwarder for InternalChannelSourceForwarder {
44	fn send(	3,616,129✔
45	&mut self,	3,616,129✔
46	txid: u64,	3,616,129✔
47	seq_in_tx: u64,	3,616,129✔
48	op: Operation,	3,616,129✔
49	port: PortHandle,	3,616,129✔
50	) -> Result<(), ExecutionError> {	3,616,129✔
51	internal_err!(self.sender.send((port, txid, seq_in_tx, op)))	11✔
52	}	3,616,129✔
53	}
54
55	/// The sender half of a source in the execution DAG.
56	#[derive(Debug)]	×
57	pub struct SourceSenderNode {
58	/// Node handle in description DAG.
59	node_handle: NodeHandle,
60	/// The source.
61	source: Box<dyn Source>,
62	/// Last checkpointed output data sequence number.
63	last_checkpoint: (u64, u64),
64	/// The forwarder that will be passed to the source for outputig data.
65	forwarder: InternalChannelSourceForwarder,
66	/// If the execution DAG should be running. Used for terminating the execution DAG.
67	running: Arc<AtomicBool>,
68	}
69
70	impl SourceSenderNode {
71	/// # Arguments
72	///
73	/// - `node_handle`: Node handle in description DAG.
74	/// - `source_factory`: Source factory in description DAG.
75	/// - `output_schemas`: Output data schemas.
76	/// - `last_checkpoint`: Last checkpointed output of this source.
77	/// - `sender`: Channel to send data to.
78	/// - `running`: If the execution DAG should still be running.
79	pub fn new<T: Clone>(	64✔
80	node_handle: NodeHandle,	64✔
81	source_factory: &dyn SourceFactory<T>,	64✔
82	output_schemas: HashMap<PortHandle, Schema>,	64✔
83	last_checkpoint: (u64, u64),	64✔
84	sender: Sender<(PortHandle, u64, u64, Operation)>,	64✔
85	running: Arc<AtomicBool>,	64✔
86	) -> Result<Self, ExecutionError> {	64✔
87	let source = source_factory.build(output_schemas)?;	64✔
88	let forwarder = InternalChannelSourceForwarder::new(sender);	63✔
89	Ok(Self {	63✔
90	node_handle,	63✔
91	source,	63✔
92	last_checkpoint,	63✔
93	forwarder,	63✔
94	running,	63✔
95	})	63✔
96	}	64✔
97	}
98
99	impl Node for SourceSenderNode {
100	fn run(mut self) -> Result<(), ExecutionError> {	212✔
101	let result = self	212✔
102	.source	212✔
103	.start(&mut self.forwarder, Some(self.last_checkpoint));	212✔
104	self.running.store(false, Ordering::SeqCst);	212✔
105	debug!("[{}-sender] Quit", self.node_handle);	212✔
106	result	212✔
107	}	212✔
108	}
109
110	/// The listener part of a source in the execution DAG.
111	#[derive(Debug)]	×
112	pub struct SourceListenerNode {
113	/// Node handle in description DAG.
114	node_handle: NodeHandle,
115	/// Output from corresponding source sender.
116	receiver: Receiver<(PortHandle, u64, u64, Operation)>,
117	/// Receiving timeout.
118	timeout: Duration,
119	/// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
120	running: Arc<AtomicBool>,
121	/// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
122	channel_manager: SourceChannelManager,
123	}
124
125	impl SourceListenerNode {
126	/// # Arguments
127	///
128	/// - `node_handle`: Node handle in description DAG.
129	/// - `receiver`: Channel that the data comes in.
130	/// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
131	/// - `base_path`: Base path of persisted data for the last execution of the description DAG.
132	/// - `output_ports`: Output port definition of the source in description DAG.
133	/// - `record_readers`: Record readers of all stateful ports.
134	/// - `senders`: Output channels from this processor.
135	/// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
136	/// - `running`: If the execution DAG should still be running.
137	/// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
138	/// - `output_schemas`: Output data schemas.
139	/// - `retention_queue_size`: Size of retention queue (used by RecordWriter)
140	#[allow(clippy::too_many_arguments)]	×
141	pub(crate) fn new(	154✔
142	node_handle: NodeHandle,	154✔
143	receiver: Receiver<(PortHandle, u64, u64, Operation)>,	154✔
144	timeout: Duration,	154✔
145	base_path: &Path,	154✔
146	output_ports: &[OutputPortDef],	154✔
147	record_readers: Arc<	154✔
148	RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,	154✔
149	>,	154✔
150	senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,	154✔
151	edges: &[Edge],	154✔
152	running: Arc<AtomicBool>,	154✔
153	commit_sz: u32,	154✔
154	max_duration_between_commits: Duration,	154✔
155	epoch_manager: Arc<EpochManager>,	154✔
156	output_schemas: HashMap<PortHandle, Schema>,	154✔
157	start_seq: (u64, u64),	154✔
158	retention_queue_size: usize,	154✔
159	) -> Result<Self, ExecutionError> {	154✔
160	let state_meta = init_component(&node_handle, base_path, \|_\| Ok(()))?;	154✔
161	let (master_tx, port_databases) =	153✔
162	create_ports_databases_and_fill_downstream_record_readers(	153✔
163	&node_handle,	153✔
164	edges,	153✔
165	state_meta.env,	153✔
166	output_ports,	153✔
167	&mut record_readers.write(),	153✔
168	)?;	153✔
169	let channel_manager = SourceChannelManager::new(	153✔
170	node_handle.clone(),	153✔
171	senders,	153✔
172	StateWriter::new(	153✔
173	state_meta.meta_db,	153✔
174	port_databases,	153✔
175	master_tx,	153✔
176	output_schemas,	153✔
177	retention_queue_size,	153✔
178	)?,	153✔
179	true,	×
180	commit_sz,	153✔
181	max_duration_between_commits,	153✔
182	epoch_manager,	153✔
183	start_seq,	153✔
184	);	153✔
185	Ok(Self {	153✔
186	node_handle,	153✔
187	receiver,	153✔
188	timeout,	153✔
189	running,	153✔
190	channel_manager,	153✔
191	})	153✔
192	}	154✔
193	}	×
194		×
195	impl SourceListenerNode {	×
196	/// Returns if the node should terminate.	×
197	fn send_and_trigger_commit_if_needed(	3,589,285✔
198	&mut self,	3,589,285✔
199	data: Option<(PortHandle, u64, u64, Operation)>,	3,589,285✔
200	) -> Result<bool, ExecutionError> {	3,589,285✔
201	// First check if termination was requested.	3,589,285✔
202	let terminating = !self.running.load(Ordering::SeqCst);	3,589,285✔
203	// If this commit was not requested with termination at the start, we shouldn't terminate either.	×
204	let terminating = match data {	3,589,285✔
205	Some((port, txid, seq_in_tx, op)) => self	3,588,709✔
206	.channel_manager	3,588,709✔
207	.send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,	3,588,709✔
208	None => self.channel_manager.trigger_commit_if_needed(terminating)?,	576✔
209	};	×
210	if terminating {	3,589,278✔
211	self.channel_manager.terminate()?;	2,025✔
212	debug!("[{}-listener] Quitting", &self.node_handle);	2,025✔
213	}	3,587,253✔
214	Ok(terminating)	3,587,397✔
215	}	3,587,404✔
216	}	×
217		×
218	impl Node for SourceListenerNode {	×
219	fn run(mut self) -> Result<(), ExecutionError> {	153✔
220	loop {	3,589,146✔
221	match self.receiver.recv_timeout(self.timeout) {	3,589,146✔
222	Ok(data) => {	3,588,469✔
223	if self.send_and_trigger_commit_if_needed(Some(data))? {	3,588,469✔
224	return Ok(());	7✔
225	}	3,588,455✔
226	}	×
227	Err(e) => {	677✔
228	if self.send_and_trigger_commit_if_needed(None)? {	677✔
229	return Ok(());	137✔
230	}	540✔
231	// Channel disconnected but running flag not set to false, the source sender must have panicked.	540✔
232	if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {	540✔
233	return Err(ExecutionError::ChannelDisconnected);	2✔
234	}	538✔
235	}
236	}
237	}
238	}	153✔
239	}

getdozer / dozer / 3983272635

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