/// Copies the data from the remote tables across all databases in a
/// remote server into a local copy that also has an extra
/// `database_name` column for the database
Copies the data from the remote tables across all databases in a
remote server into a local copy that also has an extra
`database_name` column for the database
Copies the data from the remote tables across all databases in a remote server into a local copy that also has an extra ` database_name ` column for the database
async fn load_remote_system_tables(
    context: &mut SessionContext,
    connection: Connection,
) -> Result<()> {
    // all prefixed with "system."
    let table_names = vec!["queries"];
    let start = Instant::now();
    let mut namespace_client = influxdb_iox_client::namespace::Client::new(connection.clone());
    let db_names: Vec<_> = namespace_client
        .get_namespaces()
        .await
        .context(LoadingDatabaseNamesSnafu)?
        .into_iter()
        .map(|ns| ns.name)
        .collect();
    println!("Loading system tables from {} databases", db_names.len());
    let tasks = db_names
        .into_iter()
        .flat_map(|db_name| {
            let table_names = table_names.clone();
            let connection = connection.clone();
            table_names.into_iter().map(move |table_name| {
                let table_name = table_name.to_string();
                let db_name = db_name.to_string();
                let connection = connection.clone();
                let sql = format!("select * from system.{}", table_name);
                tokio::task::spawn(async move {
                    let mut client = influxdb_iox_client::flight::Client::new(connection);
                    let mut query_results = client
                        .perform_query(ReadInfo {
                            namespace_name: db_name.clone(),
                            sql_query: sql,
                        })
                        .await
                        .context(RunningRemoteQuerySnafu)?;
                    let mut batches = vec![];
                    while let Some(data) = query_results
                        .next()
                        .await
                        .context(RunningRemoteQuerySnafu)?
                    {
                        batches.push(data);
                    }
                    let t: Result<RemoteSystemTable> = Ok(RemoteSystemTable {
                        db_name,
                        table_name,
                        batches,
                    });
                    print!("."); // give some indication of progress
                    use std::io::Write;
                    std::io::stdout().flush().unwrap();
                    t
                })
            })
        })
        .collect::<Vec<_>>();
    // now, get the results and combine them
    let results = futures::future::join_all(tasks).await;
    let mut builder = AggregatedTableBuilder::new();
    results.into_iter().for_each(|result| {
        match result {
            Ok(Ok(table)) => {
                builder.append(table);
            }
            // This is not a fatal error so log it and keep going
            Ok(Err(e)) => {
                println!("WARNING: Error running query: {}", e);
            }
            // This is not a fatal error so log it and keep going
            Err(e) => {
                println!("WARNING: Error running task: {}", e);
            }
        }
    });
    println!();
    println!(" Completed in {:?}", Instant::now() - start);
    builder.build(context);
    Ok(())
}
async fn load_remote_system_tables ( context : & mut SessionContext , connection : Connection , ) -> Result < ( ) > { // all prefixed with "system." let table_names = vec ! [ " " ] ; let start = Instant :: now ( ) ; let mut namespace_client = influxdb_iox_client :: namespace :: Client :: new ( connection . clone ( ) ) ; let db_names : Vec < _ > = namespace_client . get_namespaces ( ) . await . context ( LoadingDatabaseNamesSnafu ) ? . into_iter ( ) . map ( | ns | ns . name ) . collect ( ) ; println ! ( " " db_names len ( ) ) ; let tasks = db_names . into_iter ( ) . flat_map ( | db_name | { let table_names = table_names . clone ( ) ; let connection = connection . clone ( ) ; table_names . into_iter ( ) . map ( move | table_name | { let table_name = table_name . to_string ( ) ; let db_name = db_name . to_string ( ) ; let connection = connection . clone ( ) ; let sql = format ! ( " " table_name ) ; tokio :: task :: spawn ( async move { let mut client = influxdb_iox_client :: flight :: Client :: new ( connection ) ; let mut query_results = client . perform_query ( ReadInfo { namespace_name : db_name . clone ( ) , sql_query : sql , } ) . await . context ( RunningRemoteQuerySnafu ) ? ; let mut batches = vec ! [ ] ; while let Some ( data ) = query_results . next ( ) . await . context ( RunningRemoteQuerySnafu ) ? { batches . push ( data ) ; } let t : Result < RemoteSystemTable > = Ok ( RemoteSystemTable { db_name , table_name , batches , } ) ; print ! ( " " ) ; // give some indication of progress use std :: io :: Write ; std :: io :: stdout ( ) . flush ( ) . unwrap ( ) ; t } ) } ) } ) . collect :: < Vec < _ > > ( ) ; // now, get the results and combine them let results = futures :: future :: join_all ( tasks ) . await ; let mut builder = AggregatedTableBuilder :: new ( ) ; results . into_iter ( ) . for_each ( | result | { match result { Ok ( Ok ( table ) ) => { builder . append ( table ) ; } // This is not a fatal error so log it and keep going Ok ( Err ( e ) ) => { println ! ( " " e ) ; } // This is not a fatal error so log it and keep going Err ( e ) => { println ! ( " " e ) ; } } } ) ; println ! ( ) ; println ! ( " " Instant now ( ) start ) ; builder . build ( context ) ; Ok ( ( ) ) }
Copies the data from the remote tables across all databases in a
remote server into a local copy that also has an extra
`database_name` column for the database
Copies the data from the remote tables across all databases in a remote server into a local copy that also has an extra ` database_name ` column for the database
context
&mut SessionContext
connection
Connection