Export the sink described by `sink` from the rendering context.
(
&self,
compute_state: &mut crate::compute_state::ComputeState,
tokens: &BTreeMap<GlobalId, Rc<dyn Any>>,
dependency_ids: BTreeSet<GlobalId>,
sink_id: GlobalId
| 36 | impl<'g, T: RenderTimestamp> Context<'g, T> { |
| 37 | /// Export the sink described by `sink` from the rendering context. |
| 38 | pub(crate) fn export_sink( |
| 39 | &self, |
| 40 | compute_state: &mut crate::compute_state::ComputeState, |
| 41 | tokens: &BTreeMap<GlobalId, Rc<dyn Any>>, |
| 42 | dependency_ids: BTreeSet<GlobalId>, |
| 43 | sink_id: GlobalId, |
| 44 | sink: &ComputeSinkDesc<CollectionMetadata>, |
| 45 | start_signal: StartSignal, |
| 46 | output_probe: &Handle<mz_repr::Timestamp>, |
| 47 | outer_scope: Scope<'g, mz_repr::Timestamp>, |
| 48 | ) { |
| 49 | soft_assert_or_log!( |
| 50 | sink.non_null_assertions.is_strictly_sorted(), |
| 51 | "non-null assertions not sorted" |
| 52 | ); |
| 53 | |
| 54 | // put together tokens that belong to the export |
| 55 | let mut needed_tokens = Vec::new(); |
| 56 | for dep_id in dependency_ids { |
| 57 | if let Some(token) = tokens.get(&dep_id) { |
| 58 | needed_tokens.push(Rc::clone(token)) |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // TODO[btv] - We should determine the key and permutation to use during planning, |
| 63 | // rather than at runtime. |
| 64 | // |
| 65 | // This is basically an inlined version of the old `as_collection`. |
| 66 | let bundle = self |
| 67 | .lookup_id(mz_expr::Id::Global(sink.from)) |
| 68 | .expect("Sink source collection not loaded"); |
| 69 | let (ok_collection, mut err_collection) = if let Some(collection) = &bundle.collection { |
| 70 | collection.clone() |
| 71 | } else { |
| 72 | let (key, _arrangement) = bundle |
| 73 | .arranged |
| 74 | .iter() |
| 75 | .next() |
| 76 | .expect("Invariant violated: at least one collection must be present."); |
| 77 | let unthinned_arity = sink.from_desc.arity(); |
| 78 | let (permutation, thinning) = permutation_for_arrangement(key, unthinned_arity); |
| 79 | let mut mfp = MapFilterProject::new(unthinned_arity); |
| 80 | mfp.permute_fn(|c| permutation[c], thinning.len() + key.len()); |
| 81 | bundle.as_collection_core( |
| 82 | mfp, |
| 83 | Some((key.clone(), None)), |
| 84 | self.until.clone(), |
| 85 | &self.config_set, |
| 86 | ) |
| 87 | }; |
| 88 | |
| 89 | // Attach logging of dataflow errors. |
| 90 | if let Some(logger) = compute_state.compute_logger.clone() { |
| 91 | err_collection = err_collection.log_dataflow_errors(logger, sink_id); |
| 92 | } |
| 93 | |
| 94 | let mut ok_collection = ok_collection.leave(outer_scope); |
| 95 | let mut err_collection = err_collection.leave(outer_scope); |
no test coverage detected