(
&mut self,
dataflow: DataflowDescription<RenderPlan, CollectionMetadata>,
)
| 543 | } |
| 544 | |
| 545 | fn handle_create_dataflow( |
| 546 | &mut self, |
| 547 | dataflow: DataflowDescription<RenderPlan, CollectionMetadata>, |
| 548 | ) { |
| 549 | let dataflow_index = Rc::new(self.timely_worker.next_dataflow_index()); |
| 550 | let as_of = dataflow.as_of.clone().unwrap(); |
| 551 | |
| 552 | let dataflow_expiration = dataflow |
| 553 | .time_dependence |
| 554 | .as_ref() |
| 555 | .map(|time_dependence| { |
| 556 | self.determine_dataflow_expiration(time_dependence, &dataflow.until) |
| 557 | }) |
| 558 | .unwrap_or_default(); |
| 559 | |
| 560 | // Add the dataflow expiration to `until`. |
| 561 | let until = dataflow.until.meet(&dataflow_expiration); |
| 562 | |
| 563 | if dataflow.is_transient() { |
| 564 | debug!( |
| 565 | name = %dataflow.debug_name, |
| 566 | import_ids = %dataflow.display_import_ids(), |
| 567 | export_ids = %dataflow.display_export_ids(), |
| 568 | as_of = ?as_of.elements(), |
| 569 | time_dependence = ?dataflow.time_dependence, |
| 570 | expiration = ?dataflow_expiration.elements(), |
| 571 | expiration_datetime = ?dataflow_expiration |
| 572 | .as_option() |
| 573 | .map(|t| mz_ore::now::to_datetime(t.into())), |
| 574 | plan_until = ?dataflow.until.elements(), |
| 575 | until = ?until.elements(), |
| 576 | "creating dataflow", |
| 577 | ); |
| 578 | } else { |
| 579 | info!( |
| 580 | name = %dataflow.debug_name, |
| 581 | import_ids = %dataflow.display_import_ids(), |
| 582 | export_ids = %dataflow.display_export_ids(), |
| 583 | as_of = ?as_of.elements(), |
| 584 | time_dependence = ?dataflow.time_dependence, |
| 585 | expiration = ?dataflow_expiration.elements(), |
| 586 | expiration_datetime = ?dataflow_expiration |
| 587 | .as_option() |
| 588 | .map(|t| mz_ore::now::to_datetime(t.into())), |
| 589 | plan_until = ?dataflow.until.elements(), |
| 590 | until = ?until.elements(), |
| 591 | "creating dataflow", |
| 592 | ); |
| 593 | }; |
| 594 | |
| 595 | let subscribe_copy_ids: BTreeSet<_> = dataflow |
| 596 | .subscribe_ids() |
| 597 | .chain(dataflow.copy_to_ids()) |
| 598 | .collect(); |
| 599 | |
| 600 | // Initialize compute and logging state for each object. |
| 601 | for object_id in dataflow.export_ids() { |
| 602 | let is_subscribe_or_copy = subscribe_copy_ids.contains(&object_id); |
no test coverage detected