use std::sync::Arc;
use crate::datasources::{exon_listing_table_options::ExonListingConfig, ScanFunction};
use datafusion::{
datasource::{function::TableFunctionImpl, TableProvider},
error::Result,
execution::context::SessionContext,
logical_expr::Expr,
};
use exon_common::TableSchema;
use super::table_provider::{ListingBCFTable, ListingBCFTableOptions};
#[derive(Default)]
pub struct BCFScanFunction {
ctx: SessionContext,
}
impl std::fmt::Debug for BCFScanFunction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BCFScanFunction").finish()
}
}
impl BCFScanFunction {
pub fn new(ctx: SessionContext) -> Self {
Self { ctx }
}
}
impl TableFunctionImpl for BCFScanFunction {
fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
let listing_scan_function = ScanFunction::try_from(exprs)?;
let options = ListingBCFTableOptions::default();
let schema = futures::executor::block_on(async {
let schema = options
.infer_schema(&self.ctx.state(), &listing_scan_function.listing_table_url)
.await?;
Ok::<TableSchema, datafusion::error::DataFusionError>(schema)
})?;
let config =
ExonListingConfig::new_with_options(listing_scan_function.listing_table_url, options);
let listing_table = ListingBCFTable::new(config, schema);
Ok(Arc::new(listing_table))
}
}