1#![allow(missing_docs)]
24#![allow(clippy::needless_borrow)]
25
26use std::cell::RefCell;
27use std::collections::{BTreeMap, VecDeque};
28use std::convert::Infallible;
29use std::hash::{Hash, Hasher};
30use std::rc::Rc;
31use std::sync::Arc;
32use std::time::Duration;
33
34use differential_dataflow::lattice::Lattice;
35use differential_dataflow::{AsCollection, Collection, Hashable};
36use futures::stream::StreamExt;
37use mz_ore::cast::CastFrom;
38use mz_ore::collections::CollectionExt;
39use mz_ore::now::NowFn;
40use mz_persist_client::cache::PersistClientCache;
41use mz_repr::{Diff, GlobalId, RelationDesc, Row};
42use mz_storage_types::configuration::StorageConfiguration;
43use mz_storage_types::controller::CollectionMetadata;
44use mz_storage_types::dyncfgs;
45use mz_storage_types::errors::DataflowError;
46use mz_storage_types::sources::{SourceConnection, SourceExport, SourceTimestamp};
47use mz_timely_util::antichain::AntichainExt;
48use mz_timely_util::builder_async::{
49 Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
50};
51use mz_timely_util::capture::PusherCapture;
52use mz_timely_util::operator::ConcatenateFlatten;
53use mz_timely_util::reclock::reclock;
54use timely::PartialOrder;
55use timely::container::CapacityContainerBuilder;
56use timely::dataflow::channels::pact::Pipeline;
57use timely::dataflow::operators::capture::capture::Capture;
58use timely::dataflow::operators::capture::{Event, EventPusher};
59use timely::dataflow::operators::core::Map as _;
60use timely::dataflow::operators::generic::builder_rc::OperatorBuilder as OperatorBuilderRc;
61use timely::dataflow::operators::{Broadcast, CapabilitySet, Inspect, Leave};
62use timely::dataflow::scopes::Child;
63use timely::dataflow::{Scope, Stream};
64use timely::order::TotalOrder;
65use timely::progress::frontier::MutableAntichain;
66use timely::progress::{Antichain, Timestamp};
67use tokio::sync::{Semaphore, watch};
68use tokio_stream::wrappers::WatchStream;
69use tracing::trace;
70
71use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate};
72use crate::metrics::StorageMetrics;
73use crate::metrics::source::SourceMetrics;
74use crate::source::probe;
75use crate::source::reclock::ReclockOperator;
76use crate::source::types::{Probe, SourceMessage, SourceOutput, SourceRender, StackedCollection};
77use crate::statistics::SourceStatistics;
78
79#[derive(Clone)]
82pub struct RawSourceCreationConfig {
83 pub name: String,
85 pub id: GlobalId,
87 pub source_exports: BTreeMap<GlobalId, SourceExport<CollectionMetadata>>,
89 pub worker_id: usize,
91 pub worker_count: usize,
93 pub timestamp_interval: Duration,
96 pub now_fn: NowFn,
98 pub metrics: StorageMetrics,
100 pub as_of: Antichain<mz_repr::Timestamp>,
102 pub resume_uppers: BTreeMap<GlobalId, Antichain<mz_repr::Timestamp>>,
105 pub source_resume_uppers: BTreeMap<GlobalId, Vec<Row>>,
112 pub persist_clients: Arc<PersistClientCache>,
114 pub statistics: BTreeMap<GlobalId, SourceStatistics>,
116 pub shared_remap_upper: Rc<RefCell<Antichain<mz_repr::Timestamp>>>,
118 pub config: StorageConfiguration,
120 pub remap_collection_id: GlobalId,
122 pub remap_metadata: CollectionMetadata,
124 pub busy_signal: Arc<Semaphore>,
127}
128
129#[derive(Clone)]
132pub struct SourceExportCreationConfig {
133 pub id: GlobalId,
135 pub worker_id: usize,
137 pub metrics: StorageMetrics,
139 pub source_statistics: SourceStatistics,
141}
142
143impl RawSourceCreationConfig {
144 pub fn responsible_worker<P: Hash>(&self, partition: P) -> usize {
146 let mut h = std::hash::DefaultHasher::default();
147 (self.id, partition).hash(&mut h);
148 let key = usize::cast_from(h.finish());
149 key % self.worker_count
150 }
151
152 pub fn responsible_for<P: Hash>(&self, partition: P) -> bool {
154 self.responsible_worker(partition) == self.worker_id
155 }
156}
157
158pub fn create_raw_source<'g, G: Scope<Timestamp = ()>, C>(
171 scope: &mut Child<'g, G, mz_repr::Timestamp>,
172 storage_state: &crate::storage_state::StorageState,
173 committed_upper: &Stream<Child<'g, G, mz_repr::Timestamp>, ()>,
174 config: &RawSourceCreationConfig,
175 source_connection: C,
176 start_signal: impl std::future::Future<Output = ()> + 'static,
177) -> (
178 BTreeMap<
179 GlobalId,
180 Collection<
181 Child<'g, G, mz_repr::Timestamp>,
182 Result<SourceOutput<C::Time>, DataflowError>,
183 Diff,
184 >,
185 >,
186 Stream<G, HealthStatusMessage>,
187 Vec<PressOnDropButton>,
188)
189where
190 C: SourceConnection + SourceRender + Clone + 'static,
191{
192 let worker_id = config.worker_id;
193 let id = config.id;
194
195 let mut tokens = vec![];
196
197 let (ingested_upper_tx, ingested_upper_rx) =
198 watch::channel(MutableAntichain::new_bottom(C::Time::minimum()));
199 let (probed_upper_tx, probed_upper_rx) = watch::channel(None);
200
201 let source_metrics = Arc::new(config.metrics.get_source_metrics(id, worker_id));
202
203 let timestamp_desc = source_connection.timestamp_desc();
204
205 let (remap_collection, remap_token) = remap_operator(
206 scope,
207 storage_state,
208 config.clone(),
209 probed_upper_rx,
210 ingested_upper_rx,
211 timestamp_desc,
212 );
213 let remap_collection = remap_collection.inner.broadcast().as_collection();
215 tokens.push(remap_token);
216
217 let committed_upper = reclock_committed_upper(
218 &remap_collection,
219 config.as_of.clone(),
220 committed_upper,
221 id,
222 Arc::clone(&source_metrics),
223 );
224
225 let mut reclocked_exports = BTreeMap::new();
226
227 let reclocked_exports2 = &mut reclocked_exports;
228 let (health, source_tokens) = scope.parent.scoped("SourceTimeDomain", move |scope| {
229 let (exports, source_upper, health_stream, source_tokens) = source_render_operator(
230 scope,
231 config,
232 source_connection,
233 probed_upper_tx,
234 committed_upper,
235 start_signal,
236 );
237
238 for (id, export) in exports {
239 let (reclock_pusher, reclocked) = reclock(&remap_collection, config.as_of.clone());
240 export
241 .inner
242 .map(move |(result, from_time, diff)| {
243 let result = match result {
244 Ok(msg) => Ok(SourceOutput {
245 key: msg.key.clone(),
246 value: msg.value.clone(),
247 metadata: msg.metadata.clone(),
248 from_time: from_time.clone(),
249 }),
250 Err(err) => Err(err.clone()),
251 };
252 (result, from_time.clone(), *diff)
253 })
254 .capture_into(PusherCapture(reclock_pusher));
255 reclocked_exports2.insert(id, reclocked);
256 }
257
258 source_upper.capture_into(FrontierCapture(ingested_upper_tx));
259
260 (health_stream.leave(), source_tokens)
261 });
262
263 tokens.extend(source_tokens);
264
265 (reclocked_exports, health, tokens)
266}
267
268pub struct FrontierCapture<T>(watch::Sender<MutableAntichain<T>>);
269
270impl<T: Timestamp> EventPusher<T, Vec<Infallible>> for FrontierCapture<T> {
271 fn push(&mut self, event: Event<T, Vec<Infallible>>) {
272 match event {
273 Event::Progress(changes) => self.0.send_modify(|frontier| {
274 frontier.update_iter(changes);
275 }),
276 Event::Messages(_, _) => unreachable!(),
277 }
278 }
279}
280
281fn source_render_operator<G, C>(
287 scope: &mut G,
288 config: &RawSourceCreationConfig,
289 source_connection: C,
290 probed_upper_tx: watch::Sender<Option<Probe<C::Time>>>,
291 resume_uppers: impl futures::Stream<Item = Antichain<C::Time>> + 'static,
292 start_signal: impl std::future::Future<Output = ()> + 'static,
293) -> (
294 BTreeMap<GlobalId, StackedCollection<G, Result<SourceMessage, DataflowError>>>,
295 Stream<G, Infallible>,
296 Stream<G, HealthStatusMessage>,
297 Vec<PressOnDropButton>,
298)
299where
300 G: Scope<Timestamp = C::Time>,
301 C: SourceRender + 'static,
302{
303 let source_id = config.id;
304 let worker_id = config.worker_id;
305 let now_fn = config.now_fn.clone();
306 let timestamp_interval = config.timestamp_interval;
307
308 let resume_uppers = resume_uppers.inspect(move |upper| {
309 let upper = upper.pretty();
310 trace!(%upper, "timely-{worker_id} source({source_id}) received resume upper");
311 });
312
313 let (exports, progress, health, probes, tokens) =
314 source_connection.render(scope, config, resume_uppers, start_signal);
315
316 let mut export_collections = BTreeMap::new();
317
318 let source_metrics = config.metrics.get_source_metrics(config.id, worker_id);
319
320 let resume_upper = Antichain::from_iter(
322 config
323 .resume_uppers
324 .values()
325 .flat_map(|f| f.iter().cloned()),
326 );
327 source_metrics
328 .resume_upper
329 .set(mz_persist_client::metrics::encode_ts_metric(&resume_upper));
330
331 let mut health_streams = vec![];
332
333 for (id, export) in exports {
334 let name = format!("SourceGenericStats({})", id);
335 let mut builder = OperatorBuilderRc::new(name, scope.clone());
336
337 let (mut health_output, derived_health) =
338 builder.new_output::<CapacityContainerBuilder<_>>();
339 health_streams.push(derived_health);
340
341 let (mut output, new_export) = builder.new_output::<CapacityContainerBuilder<_>>();
342
343 let mut input = builder.new_input(&export.inner, Pipeline);
344 export_collections.insert(id, new_export.as_collection());
345
346 let bytes_read_counter = config.metrics.source_defs.bytes_read.clone();
347 let source_statistics = config
348 .statistics
349 .get(&id)
350 .expect("statistics initialized")
351 .clone();
352
353 builder.build(move |mut caps| {
354 let mut health_cap = Some(caps.remove(0));
355
356 move |frontiers| {
357 let mut last_status = None;
358 let mut health_output = health_output.activate();
359
360 if frontiers[0].is_empty() {
361 health_cap = None;
362 return;
363 }
364 let health_cap = health_cap.as_mut().unwrap();
365
366 while let Some((cap, data)) = input.next() {
367 for (message, _, _) in data.iter() {
368 let status = match &message {
369 Ok(_) => HealthStatusUpdate::running(),
370 Err(error) => HealthStatusUpdate::stalled(
374 error.to_string(),
375 Some(
376 "retracting the errored value may resume the source"
377 .to_string(),
378 ),
379 ),
380 };
381
382 let status = HealthStatusMessage {
383 id: Some(id),
384 namespace: C::STATUS_NAMESPACE.clone(),
385 update: status,
386 };
387 if last_status.as_ref() != Some(&status) {
388 last_status = Some(status.clone());
389 health_output.session(&health_cap).give(status);
390 }
391
392 match message {
393 Ok(message) => {
394 source_statistics.inc_messages_received_by(1);
395 let key_len = u64::cast_from(message.key.byte_len());
396 let value_len = u64::cast_from(message.value.byte_len());
397 bytes_read_counter.inc_by(key_len + value_len);
398 source_statistics.inc_bytes_received_by(key_len + value_len);
399 }
400 Err(_) => {}
401 }
402 }
403 let mut output = output.activate();
404 output.session(&cap).give_container(data);
405 }
406 }
407 });
408 }
409
410 let probe_stream = match probes {
411 Some(stream) => stream,
412 None => synthesize_probes(source_id, &progress, timestamp_interval, now_fn),
413 };
414
415 probe_stream.broadcast().inspect(move |probe| {
419 let _ = probed_upper_tx.send(Some(probe.clone()));
421 });
422
423 (
424 export_collections,
425 progress,
426 health.concatenate_flatten::<_, CapacityContainerBuilder<_>>(health_streams),
427 tokens,
428 )
429}
430
431fn remap_operator<G, FromTime>(
437 scope: &G,
438 storage_state: &crate::storage_state::StorageState,
439 config: RawSourceCreationConfig,
440 mut probed_upper: watch::Receiver<Option<Probe<FromTime>>>,
441 mut ingested_upper: watch::Receiver<MutableAntichain<FromTime>>,
442 remap_relation_desc: RelationDesc,
443) -> (Collection<G, FromTime, Diff>, PressOnDropButton)
444where
445 G: Scope<Timestamp = mz_repr::Timestamp>,
446 FromTime: SourceTimestamp,
447{
448 let RawSourceCreationConfig {
449 name,
450 id,
451 source_exports: _,
452 worker_id,
453 worker_count,
454 timestamp_interval,
455 remap_metadata,
456 as_of,
457 resume_uppers: _,
458 source_resume_uppers: _,
459 metrics: _,
460 now_fn,
461 persist_clients,
462 statistics: _,
463 shared_remap_upper,
464 config: _,
465 remap_collection_id,
466 busy_signal: _,
467 } = config;
468
469 let read_only_rx = storage_state.read_only_rx.clone();
470 let error_handler = storage_state.error_handler("remap_operator", id);
471
472 let chosen_worker = usize::cast_from(id.hashed() % u64::cast_from(worker_count));
473 let active_worker = chosen_worker == worker_id;
474
475 let operator_name = format!("remap({})", id);
476 let mut remap_op = AsyncOperatorBuilder::new(operator_name, scope.clone());
477 let (remap_output, remap_stream) = remap_op.new_output::<CapacityContainerBuilder<_>>();
478
479 let button = remap_op.build(move |capabilities| async move {
480 if !active_worker {
481 shared_remap_upper.borrow_mut().clear();
484 return;
485 }
486
487 let mut cap_set = CapabilitySet::from_elem(capabilities.into_element());
488
489 let remap_handle = crate::source::reclock::compat::PersistHandle::<FromTime, _>::new(
490 Arc::clone(&persist_clients),
491 read_only_rx,
492 remap_metadata.clone(),
493 as_of.clone(),
494 shared_remap_upper,
495 id,
496 "remap",
497 worker_id,
498 worker_count,
499 remap_relation_desc,
500 remap_collection_id,
501 )
502 .await;
503
504 let remap_handle = match remap_handle {
505 Ok(handle) => handle,
506 Err(e) => {
507 error_handler
508 .report_and_stop(
509 e.context(format!("Failed to create remap handle for source {name}")),
510 )
511 .await
512 }
513 };
514
515 let (mut timestamper, mut initial_batch) = ReclockOperator::new(remap_handle).await;
516
517 trace!(
520 "timely-{worker_id} remap({id}) emitting remap snapshot: trace_updates={:?}",
521 &initial_batch.updates
522 );
523
524 let cap = cap_set.delayed(cap_set.first().unwrap());
525 remap_output.give_container(&cap, &mut initial_batch.updates);
526 drop(cap);
527 cap_set.downgrade(initial_batch.upper);
528
529 let mut ticker = tokio::time::interval(timestamp_interval);
530 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
531
532 let mut prev_probe_ts: Option<mz_repr::Timestamp> = None;
533 let timestamp_interval_ms: u64 = timestamp_interval
534 .as_millis()
535 .try_into()
536 .expect("huge duration");
537
538 while !cap_set.is_empty() {
539 let reclock_to_latest =
542 dyncfgs::STORAGE_RECLOCK_TO_LATEST.get(&config.config.config_set());
543
544 let mut new_probe = None;
548 if reclock_to_latest {
549 new_probe = probed_upper
550 .wait_for(|new_probe| match (prev_probe_ts, new_probe) {
551 (None, Some(_)) => true,
552 (Some(prev_ts), Some(new)) => prev_ts < new.probe_ts,
553 _ => false,
554 })
555 .await
556 .map(|probe| (*probe).clone())
557 .unwrap_or_else(|_| {
558 Some(Probe {
559 probe_ts: now_fn().into(),
560 upstream_frontier: Antichain::new(),
561 })
562 });
563 } else {
564 while prev_probe_ts >= new_probe.as_ref().map(|p| p.probe_ts) {
565 ticker.tick().await;
566 let upstream_frontier = ingested_upper
571 .wait_for(|f| *f.frontier() != [FromTime::minimum()])
572 .await
573 .unwrap()
574 .frontier()
575 .to_owned();
576
577 let now = (now_fn)();
578 let mut probe_ts = now - now % timestamp_interval_ms;
579 if (now % timestamp_interval_ms) != 0 {
580 probe_ts += timestamp_interval_ms;
581 }
582 new_probe = Some(Probe {
583 probe_ts: probe_ts.into(),
584 upstream_frontier,
585 });
586 }
587 };
588
589 let probe = new_probe.expect("known to be Some");
590 prev_probe_ts = Some(probe.probe_ts);
591
592 let binding_ts = probe.probe_ts;
593 let cur_source_upper = probe.upstream_frontier;
594
595 let new_into_upper = Antichain::from_elem(binding_ts.step_forward());
596
597 let mut remap_trace_batch = timestamper
598 .mint(binding_ts, new_into_upper, cur_source_upper.borrow())
599 .await;
600
601 trace!(
602 "timely-{worker_id} remap({id}) minted new bindings: \
603 updates={:?} \
604 source_upper={} \
605 trace_upper={}",
606 &remap_trace_batch.updates,
607 cur_source_upper.pretty(),
608 remap_trace_batch.upper.pretty()
609 );
610
611 let cap = cap_set.delayed(cap_set.first().unwrap());
612 remap_output.give_container(&cap, &mut remap_trace_batch.updates);
613 cap_set.downgrade(remap_trace_batch.upper);
614 }
615 });
616
617 (remap_stream.as_collection(), button.press_on_drop())
618}
619
620fn reclock_committed_upper<G, FromTime>(
624 bindings: &Collection<G, FromTime, Diff>,
625 as_of: Antichain<G::Timestamp>,
626 committed_upper: &Stream<G, ()>,
627 id: GlobalId,
628 metrics: Arc<SourceMetrics>,
629) -> impl futures::stream::Stream<Item = Antichain<FromTime>> + 'static
630where
631 G: Scope,
632 G::Timestamp: Lattice + TotalOrder,
633 FromTime: SourceTimestamp,
634{
635 let (tx, rx) = watch::channel(Antichain::from_elem(FromTime::minimum()));
636 let scope = bindings.scope().clone();
637
638 let name = format!("ReclockCommitUpper({id})");
639 let mut builder = OperatorBuilderRc::new(name, scope);
640
641 let mut bindings = builder.new_input(&bindings.inner, Pipeline);
642 let _ = builder.new_input(committed_upper, Pipeline);
643
644 builder.build(move |_| {
645 use timely::progress::ChangeBatch;
647 let mut accepted_times: ChangeBatch<(G::Timestamp, FromTime)> = ChangeBatch::new();
648 let mut upper = Antichain::from_elem(Timestamp::minimum());
650 let mut ready_times = VecDeque::new();
652 let mut source_upper = MutableAntichain::new();
653
654 move |frontiers| {
655 while let Some((_, data)) = bindings.next() {
657 accepted_times.extend(data.drain(..).map(|(from, mut into, diff)| {
658 into.advance_by(as_of.borrow());
659 ((into, from), diff.into_inner())
660 }));
661 }
662 let new_upper = frontiers[0].frontier();
664 if PartialOrder::less_than(&upper.borrow(), &new_upper) {
665 upper = new_upper.to_owned();
666 let mut pending_times = std::mem::take(&mut accepted_times).into_inner();
669 pending_times.sort_unstable_by(|a, b| a.0.cmp(&b.0));
671 for ((into, from), diff) in pending_times.drain(..) {
672 if !upper.less_equal(&into) {
673 ready_times.push_back((from, into, diff));
674 } else {
675 accepted_times.update((into, from), diff);
676 }
677 }
678 }
679
680 if as_of.iter().all(|t| !upper.less_equal(t)) {
682 let committed_upper = frontiers[1].frontier();
683 if as_of.iter().all(|t| !committed_upper.less_equal(t)) {
684 let reclocked_upper = match committed_upper.as_option() {
725 Some(t_next) => {
726 let idx = ready_times.partition_point(|(_, t, _)| t < t_next);
727 let updates = ready_times
728 .drain(0..idx)
729 .map(|(from_time, _, diff)| (from_time, diff));
730 source_upper.update_iter(updates);
731 source_upper.frontier().to_owned()
734 }
735 None => Antichain::new(),
736 };
737 tx.send_replace(reclocked_upper);
738 }
739 }
740
741 metrics
742 .commit_upper_accepted_times
743 .set(u64::cast_from(accepted_times.len()));
744 metrics
745 .commit_upper_ready_times
746 .set(u64::cast_from(ready_times.len()));
747 }
748 });
749
750 WatchStream::from_changes(rx)
751}
752
753fn synthesize_probes<G>(
759 source_id: GlobalId,
760 progress: &Stream<G, Infallible>,
761 interval: Duration,
762 now_fn: NowFn,
763) -> Stream<G, Probe<G::Timestamp>>
764where
765 G: Scope,
766{
767 let scope = progress.scope();
768
769 let active_worker = usize::cast_from(source_id.hashed()) % scope.peers();
770 let is_active_worker = active_worker == scope.index();
771
772 let mut op = AsyncOperatorBuilder::new("synthesize_probes".into(), scope);
773 let (output, output_stream) = op.new_output();
774 let mut input = op.new_input_for(progress, Pipeline, &output);
775
776 op.build(|caps| async move {
777 if !is_active_worker {
778 return;
779 }
780
781 let [cap] = caps.try_into().expect("one capability per output");
782
783 let mut ticker = probe::Ticker::new(move || interval, now_fn.clone());
784
785 let minimum_frontier = Antichain::from_elem(Timestamp::minimum());
786 let mut frontier = minimum_frontier.clone();
787 loop {
788 tokio::select! {
789 event = input.next() => match event {
790 Some(AsyncEvent::Progress(progress)) => frontier = progress,
791 Some(AsyncEvent::Data(..)) => unreachable!(),
792 None => break,
793 },
794 probe_ts = ticker.tick(), if frontier != minimum_frontier => {
799 let probe = Probe {
800 probe_ts,
801 upstream_frontier: frontier.clone(),
802 };
803 output.give(&cap, probe);
804 }
805 }
806 }
807
808 let probe = Probe {
809 probe_ts: now_fn().into(),
810 upstream_frontier: Antichain::new(),
811 };
812 output.give(&cap, probe);
813 });
814
815 output_stream
816}