Method export_index_iterative

src/compute/src/render.rs:761–853

(
        &self,
        outer: Scope<'outer, mz_repr::Timestamp>,
        compute_state: &mut ComputeState,
        tokens: &BTreeMap<GlobalId, Rc<dyn std::any::Any>>,
        dependency_ids: BTreeSe

    T: RenderTimestamp,
{
    pub(crate) fn export_index_iterative<'outer>(
        &self,
        outer: Scope<'outer, mz_repr::Timestamp>,
        compute_state: &mut ComputeState,
        tokens: &BTreeMap<GlobalId, Rc<dyn std::any::Any>>,
        dependency_ids: BTreeSet<GlobalId>,
        idx_id: GlobalId,
        idx: &IndexDesc,
        output_probe: &MzProbeHandle<mz_repr::Timestamp>,
    ) {
        // put together tokens that belong to the export
        let mut needed_tokens = Vec::new();
        for dep_id in dependency_ids {
            if let Some(token) = tokens.get(&dep_id) {
                needed_tokens.push(Rc::clone(token));
            }
        }
        let bundle = self.lookup_id(Id::Global(idx_id)).unwrap_or_else(|| {
            panic!(
                "Arrangement alarmingly absent! id: {:?}",
                Id::Global(idx_id)
            )
        });

        match bundle.arrangement(&idx.key) {
            Some(ArrangementFlavor::Local(oks, errs)) => {
                // TODO: The following as_collection/leave/arrange sequence could be optimized.
                //   * Combine as_collection and leave into a single function.
                //   * Use columnar to extract columns from the batches to implement leave.
                let mut oks = oks
                    .as_collection(|k, v| (k.to_row(), v.to_row()))
                    .leave(outer)
                    .mz_arrange::<
                        ColumnationChunker<_>,
                        RowRowBatcher<_, _>,
                        RowRowBuilder<_, _>,
                        _,
                    >(
                        "Arrange export iterative",
                    );

                let mut errs = errs
                    .as_collection(|k, v| (k.clone(), v.clone()))
                    .leave(outer)
                    .mz_arrange::<ColumnationChunker<_>, ErrBatcher<_, _>, ErrBuilder<_, _>, _>(
                        "Arrange export iterative err",
                    );

                // Ensure that the frontier does not advance past the expiration time, if set.
                // Otherwise, we might write down incorrect data.
                if let Some(&expiration) = self.dataflow_expiration.as_option() {
                    oks.stream = oks.stream.expire_stream_at(
                        &format!("{}_export_index_iterative_oks", self.debug_name),
                        expiration,
                    );
                    errs.stream = errs.stream.expire_stream_at(
                        &format!("{}_export_index_iterative_err", self.debug_name),
                        expiration,