pub async fn list_schemas(
    catalog: &dyn Catalog,
) -> Result<impl Iterator<Item = (Namespace, NamespaceSchema)>> {
    let mut repos = catalog.repositories().await;

    // In order to obtain a point-in-time snapshot, first fetch the columns,
    // then the tables, and then resolve the namespace IDs to Namespace in order
    // to construct the schemas.
    //
    // The set of columns returned forms the state snapshot, with the subsequent
    // queries resolving only what is needed to construct schemas for the
    // retrieved columns (ignoring any newly added tables/namespaces since the
    // column snapshot was taken).

    // First fetch all the columns - this is the state snapshot of the catalog
    // schemas.
    let columns = repos.columns().list().await?;

    // Construct the set of table IDs these columns belong to.
    let retain_table_ids = columns.iter().map(|c| c.table_id).collect::<HashSet<_>>();

    // Fetch all tables, and filter for those that are needed to construct
    // schemas for "columns" only.
    //
    // Discard any tables that have no columns or have been created since
    // the "columns" snapshot was retrieved, and construct a map of ID->Table.
    let tables = repos
        .tables()
        .list()
        .await?
        .into_iter()
        .filter_map(|t| {
            if !retain_table_ids.contains(&t.id) {
                return None;
            }

            Some((t.id, t))
        })
        .collect::<HashMap<_, _>>();

    // Drop the table ID set as it will not be referenced again.
    drop(retain_table_ids);

    // Do all the I/O to fetch the namespaces in the background, while this
    // thread constructs the NamespaceId->TableSchema map below.
    let namespaces = tokio::spawn(async move { repos.namespaces().list().await });

    // A set of tables within a single namespace.
    type NamespaceTables = BTreeMap<String, TableSchema>;

    let mut joined = HashMap::<NamespaceId, NamespaceTables>::default();
    for column in columns {
        // Resolve the table this column references
        let table = tables.get(&column.table_id).expect("no table for column");

        let table_schema = joined
            // Find or create a record in the joined <NamespaceId, Tables> map
            // for this namespace ID.
            .entry(table.namespace_id)
            .or_default()
            // Fetch the schema record for this table, or create an empty one.
            .entry(table.name.clone())
            .or_insert_with(|| TableSchema::new(column.table_id));

        table_schema.add_column(&column);
    }

    // The table map is no longer needed - immediately reclaim the memory.
    drop(tables);

    // Convert the Namespace instances into NamespaceSchema instances.
    let iter = namespaces
        .await
        .expect("namespace list task panicked")?
        .into_iter()
        // Ignore any namespaces that did not exist when the "columns" snapshot
        // was created, or have no tables/columns (and therefore have no entry
        // in "joined").
        .filter_map(move |v| {
            let mut ns = NamespaceSchema::new(v.id, v.kafka_topic_id, v.query_pool_id);
            ns.tables = joined.remove(&v.id)?;
            Some((v, ns))
        });

    Ok(iter)
}
/// Fetch all [`NamespaceSchema`] in the catalog.
///
/// This method performs the minimal number of queries needed to build the
/// result set. No table lock is obtained, nor are queries executed within a
/// transaction, but this method does return a point-in-time snapshot of the
/// catalog state.
Fetch all [`NamespaceSchema`] in the catalog.
This method performs the minimal number of queries needed to build the
result set. No table lock is obtained, nor are queries executed within a
transaction, but this method does return a point-in-time snapshot of the
catalog state.
Fetch all [ ` NamespaceSchema ` ] in the catalog . This method performs the minimal number of queries needed to build the result set . No table lock is obtained nor are queries executed within a transaction but this method does return a point - in - time snapshot of the catalog state .
pub async fn list_schemas(
    catalog: &dyn Catalog,
) -> Result<impl Iterator<Item = (Namespace, NamespaceSchema)>> {
    let mut repos = catalog.repositories().await;
    // In order to obtain a point-in-time snapshot, first fetch the columns,
    // then the tables, and then resolve the namespace IDs to Namespace in order
    // to construct the schemas.
    //
    // The set of columns returned forms the state snapshot, with the subsequent
    // queries resolving only what is needed to construct schemas for the
    // retrieved columns (ignoring any newly added tables/namespaces since the
    // column snapshot was taken).
    // First fetch all the columns - this is the state snapshot of the catalog
    // schemas.
    let columns = repos.columns().list().await?;
    // Construct the set of table IDs these columns belong to.
    let retain_table_ids = columns.iter().map(|c| c.table_id).collect::<HashSet<_>>();
    // Fetch all tables, and filter for those that are needed to construct
    // schemas for "columns" only.
    //
    // Discard any tables that have no columns or have been created since
    // the "columns" snapshot was retrieved, and construct a map of ID->Table.
    let tables = repos
        .tables()
        .list()
        .await?
        .into_iter()
        .filter_map(|t| {
            if !retain_table_ids.contains(&t.id) {
                return None;
            }
            Some((t.id, t))
        })
        .collect::<HashMap<_, _>>();
    // Drop the table ID set as it will not be referenced again.
    drop(retain_table_ids);
    // Do all the I/O to fetch the namespaces in the background, while this
    // thread constructs the NamespaceId->TableSchema map below.
    let namespaces = tokio::spawn(async move { repos.namespaces().list().await });
    // A set of tables within a single namespace.
    type NamespaceTables = BTreeMap<String, TableSchema>;
    let mut joined = HashMap::<NamespaceId, NamespaceTables>::default();
    for column in columns {
        // Resolve the table this column references
        let table = tables.get(&column.table_id).expect("no table for column");
        let table_schema = joined
            // Find or create a record in the joined <NamespaceId, Tables> map
            // for this namespace ID.
            .entry(table.namespace_id)
            .or_default()
            // Fetch the schema record for this table, or create an empty one.
            .entry(table.name.clone())
            .or_insert_with(|| TableSchema::new(column.table_id));
        table_schema.add_column(&column);
    }
    // The table map is no longer needed - immediately reclaim the memory.
    drop(tables);
    // Convert the Namespace instances into NamespaceSchema instances.
    let iter = namespaces
        .await
        .expect("namespace list task panicked")?
        .into_iter()
        // Ignore any namespaces that did not exist when the "columns" snapshot
        // was created, or have no tables/columns (and therefore have no entry
        // in "joined").
        .filter_map(move |v| {
            let mut ns = NamespaceSchema::new(v.id, v.kafka_topic_id, v.query_pool_id);
            ns.tables = joined.remove(&v.id)?;
            Some((v, ns))
        });
    Ok(iter)
}
pub async fn list_schemas ( catalog : & dyn Catalog , ) -> Result < impl Iterator < Item = ( Namespace , NamespaceSchema ) > > { let mut repos = catalog . repositories ( ) . await ; // In order to obtain a point-in-time snapshot, first fetch the columns, // then the tables, and then resolve the namespace IDs to Namespace in order // to construct the schemas. // // The set of columns returned forms the state snapshot, with the subsequent // queries resolving only what is needed to construct schemas for the // retrieved columns (ignoring any newly added tables/namespaces since the // column snapshot was taken). // First fetch all the columns - this is the state snapshot of the catalog // schemas. let columns = repos . columns ( ) . list ( ) . await ? ; // Construct the set of table IDs these columns belong to. let retain_table_ids = columns . iter ( ) . map ( | c | c . table_id ) . collect :: < HashSet < _ > > ( ) ; // Fetch all tables, and filter for those that are needed to construct // schemas for "columns" only. // // Discard any tables that have no columns or have been created since // the "columns" snapshot was retrieved, and construct a map of ID->Table. let tables = repos . tables ( ) . list ( ) . await ? . into_iter ( ) . filter_map ( | t | { if ! retain_table_ids . contains ( & t . id ) { return None ; } Some ( ( t . id , t ) ) } ) . collect :: < HashMap < _ , _ > > ( ) ; // Drop the table ID set as it will not be referenced again. drop ( retain_table_ids ) ; // Do all the I/O to fetch the namespaces in the background, while this // thread constructs the NamespaceId->TableSchema map below. let namespaces = tokio :: spawn ( async move { repos . namespaces ( ) . list ( ) . await } ) ; // A set of tables within a single namespace. type NamespaceTables = BTreeMap < String , TableSchema > ; let mut joined = HashMap :: < NamespaceId , NamespaceTables > :: default ( ) ; for column in columns { // Resolve the table this column references let table = tables . get ( & column . table_id ) . expect ( " " ) ; let table_schema = joined // Find or create a record in the joined <NamespaceId, Tables> map // for this namespace ID. . entry ( table . namespace_id ) . or_default ( ) // Fetch the schema record for this table, or create an empty one. . entry ( table . name . clone ( ) ) . or_insert_with ( | | TableSchema :: new ( column . table_id ) ) ; table_schema . add_column ( & column ) ; } // The table map is no longer needed - immediately reclaim the memory. drop ( tables ) ; // Convert the Namespace instances into NamespaceSchema instances. let iter = namespaces . await . expect ( " " ) ? . into_iter ( ) // Ignore any namespaces that did not exist when the "columns" snapshot // was created, or have no tables/columns (and therefore have no entry // in "joined"). . filter_map ( move | v | { let mut ns = NamespaceSchema :: new ( v . id , v . kafka_topic_id , v . query_pool_id ) ; ns . tables = joined . remove ( & v . id ) ? ; Some ( ( v , ns ) ) } ) ; Ok ( iter ) }
Fetch all [`NamespaceSchema`] in the catalog.
Fetch all [ ` NamespaceSchema ` ] in the catalog .
catalog
&dyn Catalog
catalog
&dyn Catalog