fn import_into_dataflow(
        &mut self,
        id: &GlobalId,
        dataflow: &mut DataflowDesc,
    ) -> Result<(), CoordError> {
        maybe_grow(|| {
            // Avoid importing the item redundantly.
            if dataflow.is_imported(id) {
                return Ok(());
            }

            // A valid index is any index on `id` that is known to the dataflow
            // layer, as indicated by its presence in `self.indexes`.
            let valid_index = self.catalog.enabled_indexes()[id]
                .iter()
                .find(|(id, _keys)| self.indexes.contains_key(*id));
            if let Some((index_id, keys)) = valid_index {
                let index_desc = IndexDesc {
                    on_id: *id,
                    key: keys.to_vec(),
                };
                let desc = self
                    .catalog
                    .get_by_id(id)
                    .desc()
                    .expect("indexes can only be built on items with descs");
                dataflow.import_index(*index_id, index_desc, desc.typ().clone(), *id);
            } else {
                let entry = self.catalog.get_by_id(id);
                match entry.item() {
                    CatalogItem::Table(_) => {
                        let source_description = self.catalog.source_description_for(*id).unwrap();
                        let persist_details = None;
                        dataflow.import_source(*id, source_description, persist_details);
                    }
                    CatalogItem::Source(source) => {
                        if source.requires_single_materialization() {
                            let source_type = RematerializedSourceType::for_source(source);
                            let dependent_indexes = self.catalog.dependent_indexes(*id);
                            // If this source relies on any pre-existing indexes (i.e., indexes
                            // that we're not building as part of this `DataflowBuilder`), we're
                            // attempting to reinstantiate a single-use source.
                            let intersection = self.indexes.intersection(dependent_indexes);
                            if !intersection.is_empty() {
                                let existing_indexes = intersection
                                    .iter()
                                    .map(|id| self.catalog.get_by_id(id).name().item.clone())
                                    .collect();
                                return Err(CoordError::InvalidRematerialization {
                                    base_source: entry.name().item.clone(),
                                    existing_indexes,
                                    source_type,
                                });
                            }
                        }

                        let source_description = self.catalog.source_description_for(*id).unwrap();

                        let persist_desc = self
                            .persister
                            .load_source_persist_desc(&source)
                            .map_err(CoordError::Persistence)?;

                        dataflow.import_source(*id, source_description, persist_desc);
                    }
                    CatalogItem::View(view) => {
                        let expr = view.optimized_expr.clone();
                        self.import_view_into_dataflow(id, &expr, dataflow)?;
                    }
                    _ => unreachable!(),
                }
            }
            Ok(())
        })
    }
/// Imports the view, source, or table with `id` into the provided
/// dataflow description.
Imports the view, source, or table with `id` into the provided
dataflow description.
Imports the view source or table with ` id ` into the provided dataflow description .
fn import_into_dataflow(
        &mut self,
        id: &GlobalId,
        dataflow: &mut DataflowDesc,
    ) -> Result<(), CoordError> {
        maybe_grow(|| {
            // Avoid importing the item redundantly.
            if dataflow.is_imported(id) {
                return Ok(());
            }
            // A valid index is any index on `id` that is known to the dataflow
            // layer, as indicated by its presence in `self.indexes`.
            let valid_index = self.catalog.enabled_indexes()[id]
                .iter()
                .find(|(id, _keys)| self.indexes.contains_key(*id));
            if let Some((index_id, keys)) = valid_index {
                let index_desc = IndexDesc {
                    on_id: *id,
                    key: keys.to_vec(),
                };
                let desc = self
                    .catalog
                    .get_by_id(id)
                    .desc()
                    .expect("indexes can only be built on items with descs");
                dataflow.import_index(*index_id, index_desc, desc.typ().clone(), *id);
            } else {
                let entry = self.catalog.get_by_id(id);
                match entry.item() {
                    CatalogItem::Table(_) => {
                        let source_description = self.catalog.source_description_for(*id).unwrap();
                        let persist_details = None;
                        dataflow.import_source(*id, source_description, persist_details);
                    }
                    CatalogItem::Source(source) => {
                        if source.requires_single_materialization() {
                            let source_type = RematerializedSourceType::for_source(source);
                            let dependent_indexes = self.catalog.dependent_indexes(*id);
                            // If this source relies on any pre-existing indexes (i.e., indexes
                            // that we're not building as part of this `DataflowBuilder`), we're
                            // attempting to reinstantiate a single-use source.
                            let intersection = self.indexes.intersection(dependent_indexes);
                            if !intersection.is_empty() {
                                let existing_indexes = intersection
                                    .iter()
                                    .map(|id| self.catalog.get_by_id(id).name().item.clone())
                                    .collect();
                                return Err(CoordError::InvalidRematerialization {
                                    base_source: entry.name().item.clone(),
                                    existing_indexes,
                                    source_type,
                                });
                            }
                        }
                        let source_description = self.catalog.source_description_for(*id).unwrap();
                        let persist_desc = self
                            .persister
                            .load_source_persist_desc(&source)
                            .map_err(CoordError::Persistence)?;
                        dataflow.import_source(*id, source_description, persist_desc);
                    }
                    CatalogItem::View(view) => {
                        let expr = view.optimized_expr.clone();
                        self.import_view_into_dataflow(id, &expr, dataflow)?;
                    }
                    _ => unreachable!(),
                }
            }
            Ok(())
        })
    }
fn import_into_dataflow ( & mut self , id : & GlobalId , dataflow : & mut DataflowDesc , ) -> Result < ( ) , CoordError > { maybe_grow ( | | { // Avoid importing the item redundantly. if dataflow . is_imported ( id ) { return Ok ( ( ) ) ; } // A valid index is any index on `id` that is known to the dataflow // layer, as indicated by its presence in `self.indexes`. let valid_index = self . catalog . enabled_indexes ( ) [ id ] . iter ( ) . find ( | ( id , _keys ) | self . indexes . contains_key ( * id ) ) ; if let Some ( ( index_id , keys ) ) = valid_index { let index_desc = IndexDesc { on_id : * id , key : keys . to_vec ( ) , } ; let desc = self . catalog . get_by_id ( id ) . desc ( ) . expect ( " " ) ; dataflow . import_index ( * index_id , index_desc , desc . typ ( ) . clone ( ) , * id ) ; } else { let entry = self . catalog . get_by_id ( id ) ; match entry . item ( ) { CatalogItem :: Table ( _ ) => { let source_description = self . catalog . source_description_for ( * id ) . unwrap ( ) ; let persist_details = None ; dataflow . import_source ( * id , source_description , persist_details ) ; } CatalogItem :: Source ( source ) => { if source . requires_single_materialization ( ) { let source_type = RematerializedSourceType :: for_source ( source ) ; let dependent_indexes = self . catalog . dependent_indexes ( * id ) ; // If this source relies on any pre-existing indexes (i.e., indexes // that we're not building as part of this `DataflowBuilder`), we're // attempting to reinstantiate a single-use source. let intersection = self . indexes . intersection ( dependent_indexes ) ; if ! intersection . is_empty ( ) { let existing_indexes = intersection . iter ( ) . map ( | id | self . catalog . get_by_id ( id ) . name ( ) . item . clone ( ) ) . collect ( ) ; return Err ( CoordError :: InvalidRematerialization { base_source : entry . name ( ) . item . clone ( ) , existing_indexes , source_type , } ) ; } } let source_description = self . catalog . source_description_for ( * id ) . unwrap ( ) ; let persist_desc = self . persister . load_source_persist_desc ( & source ) . map_err ( CoordError :: Persistence ) ? ; dataflow . import_source ( * id , source_description , persist_desc ) ; } CatalogItem :: View ( view ) => { let expr = view . optimized_expr . clone ( ) ; self . import_view_into_dataflow ( id , & expr , dataflow ) ? ; } _ => unreachable ! ( ) , } } Ok ( ( ) ) } ) }
Imports the view, source, or table with `id` into the provided
dataflow description.
Imports the view source or table with ` id ` into the provided dataflow description .
&mut self
id
&GlobalId
dataflow
&mut DataflowDesc