Polars with ODBC

polars_odbc.rs

use arrow2::error::Result;
use arrow2::io::odbc::api::Cursor;
use arrow2::io::odbc::{api, read};
use polars::prelude::*;

const QUERY: &str = include_str!("../query.sql");

fn main() -> Result<()> {
    let connector = ".......";

    let env = api::Environment::new()?;
    let connection = env.connect_with_connection_string(connector)?;
    let mut prep = connection.prepare(QUERY)?;

    let fields = read::infer_schema(&prep)?;

    let mut df = fields
        .iter()
        .map(|s| {
            let typ = DataType::from(s.data_type());
            Series::new_empty(&s.name, &typ)
        })
        .collect::<Vec<_>>();

    let max_batch_size = 100;
    let buffer = read::buffer_from_metadata(&prep, max_batch_size)?;

    let cursor = prep.execute(())?.unwrap();
    let mut cursor = cursor.bind_buffer(buffer)?;

    while let Some(batch) = cursor.fetch()? {
        for ((idx, field), df_elem) in (0..batch.num_cols()).zip(fields.iter()).zip(df.iter_mut()) {
            let column_view = batch.column(idx);
            let arr = Arc::from(read::deserialize(column_view, field.data_type.clone()));
            let series = Series::try_from((field.name.as_str(), vec![arr])).unwrap();
            df_elem.append(&series).unwrap();
        }
    }

    let dataframe = DataFrame::new(df).unwrap();
    dbg!(dataframe);
    Ok(())
}