use std::{collections::HashMap, sync::Arc};
use datafusion::prelude::SessionContext;
use datafusion::sql::TableReference;
use datafusion_table_providers::{
common::DatabaseCatalogProvider, mysql::MySQLTableFactory,
sql::db_connection_pool::mysqlpool::MySQLConnectionPool, util::secrets::to_secret_map,
};
#[tokio::main]
async fn main() {
let mysql_params = to_secret_map(HashMap::from([
(
"connection_string".to_string(),
"mysql://root:password@localhost:3306/mysql_db".to_string(),
),
("sslmode".to_string(), "disabled".to_string()),
]));
let mysql_pool = Arc::new(
MySQLConnectionPool::new(mysql_params)
.await
.expect("unable to create MySQL connection pool"),
);
let table_factory = MySQLTableFactory::new(mysql_pool.clone());
let catalog = DatabaseCatalogProvider::try_new(mysql_pool).await.unwrap();
let ctx = SessionContext::new();
ctx.register_catalog("mysql", Arc::new(catalog));
ctx.register_table(
"companies_v2",
table_factory
.table_provider(TableReference::bare("companies"))
.await
.expect("failed to register table provider"),
)
.expect("failed to register table");
let df = ctx
.sql("SELECT * FROM datafusion.public.companies_v2")
.await
.expect("select failed");
df.show().await.expect("show failed");
let df = ctx
.sql("SELECT * FROM mysql.mysql_db.companies")
.await
.expect("select failed");
df.show().await.expect("show failed");
}