Method import_into_dataflow

src/adapter/src/optimize/dataflows.rs:308–383

Imports the view, source, or table with `id` into the provided dataflow description. [`OptimizerFeatures`] is used while running the [`Monotonic`] analysis. Panics if `id` refers to a non-importable item, such as an index or sink.

(
        &mut self,
        id: &GlobalId,
        dataflow: &mut DataflowDesc,
        features: &OptimizerFeatures,
    )

    ///
    /// Panics if `id` refers to a non-importable item, such as an index or sink.
    pub fn import_into_dataflow(
        &mut self,
        id: &GlobalId,
        dataflow: &mut DataflowDesc,
        features: &OptimizerFeatures,
    ) -> Result<(), OptimizerError> {
        maybe_grow(|| {
            // Avoid importing the item redundantly.
            if dataflow.is_imported(id) {
                return Ok(());
            }

            let monotonic = self.monotonic_object(*id, features);

            // A valid index is any index on `id` that is known to index oracle.
            // Here, we import all indexes that belong to all imported collections. Later,
            // `prune_and_annotate_dataflow_index_imports` runs at the end of the MIR
            // pipeline, and removes unneeded index imports based on the optimized plan.
            let mut valid_indexes = self.indexes_on(*id).peekable();
            if valid_indexes.peek().is_some() {
                for (index_id, idx) in valid_indexes {
                    let index_desc = IndexDesc {
                        on_id: *id,
                        key: idx.keys.to_vec(),
                    };
                    let entry = self.catalog.get_entry(id);
                    let desc = entry
                        .relation_desc()
                        .expect("indexes can only be built on items with descs");
                    dataflow.import_index(
                        index_id,
                        index_desc,
                        ReprRelationType::from(desc.typ()),
                        monotonic,
                    );
                }
            } else {
                drop(valid_indexes);
                let entry = self.catalog.get_entry(id);
                // Note that the following match should be kept in sync with `sufficient_collections`.
                match entry.item() {
                    CatalogItem::Table(table) => {
                        dataflow.import_source(*id, table.desc_for(id).into_typ(), monotonic);
                    }
                    CatalogItem::Source(source) => {
                        dataflow.import_source(*id, source.desc.typ().clone(), monotonic);
                    }
                    CatalogItem::View(view) => {
                        let expr = view.locally_optimized_expr.as_ref();
                        self.import_view_into_dataflow(id, expr, dataflow, features)?;
                    }
                    CatalogItem::MaterializedView(mview) if mview.replacement_target.is_some() => {
                        // Can't read from replacements, use the view definition directly.
                        let expr = mview.locally_optimized_expr.as_ref();
                        self.import_view_into_dataflow(id, expr, dataflow, features)?;
                    }
                    CatalogItem::MaterializedView(mview) => {
                        dataflow.import_source(*id, mview.desc_for(id).into_typ(), monotonic);