async fn load_schema(
    catalog: &Arc<dyn Catalog>,
    namespace: &str,
    schema: &NamespaceSchema,
) -> Result<CatalogNamespaceSchema, Error> {
    let mut repos = catalog.repositories().await;
    let kafka_topic = repos.kafka_topics().create_or_get(KAFKA_NAME).await?;
    let query_pool = repos.query_pools().create_or_get(QUERY_POOL).await?;
    // ensure there's a sequencer for this partition so it can be used later
    let _sequencer = repos
        .sequencers()
        .create_or_get(&kafka_topic, KafkaPartition::new(KAFKA_PARTITION))
        .await?;

    let namespace = match repos
        .namespaces()
        .create(namespace, "inf", kafka_topic.id, query_pool.id)
        .await
    {
        Ok(n) => n,
        Err(iox_catalog::interface::Error::NameExists { .. }) => repos
            .namespaces()
            .get_by_name(namespace)
            .await?
            .ok_or(Error::NamespaceNotFound)?,
        e => e?,
    };
    println!("namespace {} loaded into local catalog", &namespace.name);

    for (table_name, table_schema) in &schema.tables {
        let table = repos
            .tables()
            .create_or_get(table_name, namespace.id)
            .await?;
        for (column_name, column_schema) in &table_schema.columns {
            let column_type = column_schema.column_type as i16;
            let column_type: ColumnType = column_type
                .try_into()
                .expect("column type from remote not valid");
            let _column = repos
                .columns()
                .create_or_get(column_name, table.id, column_type)
                .await?;
        }
        let column_names = &table_schema
            .columns
            .keys()
            .map(|k| k.to_string())
            .collect::<Vec<_>>()
            .join(", ");
        println!(
            "table {} with columns {} loaded into local catalog",
            table_name, column_names
        );
    }

    let full_inserted_schema = get_schema_by_name(&namespace.name, repos.as_mut()).await?;

    Ok(full_inserted_schema)
}
// loads the protobuf namespace schema returned from a remote IOx server into the passed in
// catalog. It does this based on namespace, table, and column names, not IDs. It also inserts
// a kafka topic and query pool for the namespace to use, which aren't for real use, but just
// to make the loaded schema work.
loads the protobuf namespace schema returned from a remote IOx server into the passed in
catalog. It does this based on namespace, table, and column names, not IDs. It also inserts
a kafka topic and query pool for the namespace to use, which aren't for real use, but just
to make the loaded schema work.
loads the protobuf namespace schema returned from a remote IOx server into the passed in catalog . It does this based on namespace table and column names not IDs . It also inserts a kafka topic and query pool for the namespace to use which aren ' t for real use but just to make the loaded schema work .
async fn load_schema(
    catalog: &Arc<dyn Catalog>,
    namespace: &str,
    schema: &NamespaceSchema,
) -> Result<CatalogNamespaceSchema, Error> {
    let mut repos = catalog.repositories().await;
    let kafka_topic = repos.kafka_topics().create_or_get(KAFKA_NAME).await?;
    let query_pool = repos.query_pools().create_or_get(QUERY_POOL).await?;
    // ensure there's a sequencer for this partition so it can be used later
    let _sequencer = repos
        .sequencers()
        .create_or_get(&kafka_topic, KafkaPartition::new(KAFKA_PARTITION))
        .await?;
    let namespace = match repos
        .namespaces()
        .create(namespace, "inf", kafka_topic.id, query_pool.id)
        .await
    {
        Ok(n) => n,
        Err(iox_catalog::interface::Error::NameExists { .. }) => repos
            .namespaces()
            .get_by_name(namespace)
            .await?
            .ok_or(Error::NamespaceNotFound)?,
        e => e?,
    };
    println!("namespace {} loaded into local catalog", &namespace.name);
    for (table_name, table_schema) in &schema.tables {
        let table = repos
            .tables()
            .create_or_get(table_name, namespace.id)
            .await?;
        for (column_name, column_schema) in &table_schema.columns {
            let column_type = column_schema.column_type as i16;
            let column_type: ColumnType = column_type
                .try_into()
                .expect("column type from remote not valid");
            let _column = repos
                .columns()
                .create_or_get(column_name, table.id, column_type)
                .await?;
        }
        let column_names = &table_schema
            .columns
            .keys()
            .map(|k| k.to_string())
            .collect::<Vec<_>>()
            .join(", ");
        println!(
            "table {} with columns {} loaded into local catalog",
            table_name, column_names
        );
    }
    let full_inserted_schema = get_schema_by_name(&namespace.name, repos.as_mut()).await?;
    Ok(full_inserted_schema)
}
async fn load_schema ( catalog : & Arc < dyn Catalog > , namespace : & str , schema : & NamespaceSchema , ) -> Result < CatalogNamespaceSchema , Error > { let mut repos = catalog . repositories ( ) . await ; let kafka_topic = repos . kafka_topics ( ) . create_or_get ( KAFKA_NAME ) . await ? ; let query_pool = repos . query_pools ( ) . create_or_get ( QUERY_POOL ) . await ? ; // ensure there's a sequencer for this partition so it can be used later let _sequencer = repos . sequencers ( ) . create_or_get ( & kafka_topic , KafkaPartition :: new ( KAFKA_PARTITION ) ) . await ? ; let namespace = match repos . namespaces ( ) . create ( namespace , " " , kafka_topic . id , query_pool . id ) . await { Ok ( n ) => n , Err ( iox_catalog :: interface :: Error :: NameExists { .. } ) => repos . namespaces ( ) . get_by_name ( namespace ) . await ? . ok_or ( Error :: NamespaceNotFound ) ? , e => e ? , } ; println ! ( " " namespace name ) ; for ( table_name , table_schema ) in & schema . tables { let table = repos . tables ( ) . create_or_get ( table_name , namespace . id ) . await ? ; for ( column_name , column_schema ) in & table_schema . columns { let column_type = column_schema . column_type as i16 ; let column_type : ColumnType = column_type . try_into ( ) . expect ( " " ) ; let _column = repos . columns ( ) . create_or_get ( column_name , table . id , column_type ) . await ? ; } let column_names = & table_schema . columns . keys ( ) . map ( | k | k . to_string ( ) ) . collect :: < Vec < _ > > ( ) . join ( " " ) ; println ! ( " " table_name column_names ) ; } let full_inserted_schema = get_schema_by_name ( & namespace . name , repos . as_mut ( ) ) . await ? ; Ok ( full_inserted_schema ) }
loads the protobuf namespace schema returned from a remote IOx server into the passed in
catalog.
loads the protobuf namespace schema returned from a remote IOx server into the passed in catalog .
catalog
&Arc<dyn Catalog>
namespace
&str
schema
&NamespaceSchema
catalog
&Arc<dyn Catalog>
namespace
&str
schema
&NamespaceSchema