Example for reading from a delta table

[theelderbeever]

As the title says. Looking for an example on how to read from a deltalake table. Is this even possible yet? I have started with something like below. Do I need to implement my own TableProvider somewhere or does anyone know if one already exists?

    let mut table = match open_table("data/metrics").await {
        Ok(tbl) => tbl,
        Err(_) => {
            let ops = DeltaOps::try_from_uri("data/metrics").await?;
            ops.create()
                .with_table_name("metrics")
                .with_columns(VectorMetric::columns())
                .await?
        }
    };

    let mut ctx = SessionContext::new();

    ctx.register_table("metrics", Arc::new(table)).unwrap();

[simonvandel]

There's an implementation of TableProvider here: https://github.com/delta-io/delta-rs/blob/21e369f309941d069636d8b7c8d50b30e2646ea4/rust/src/delta_datafusion/mod.rs#L428

It's behind the datafusion cargo feature.

[theelderbeever]

I have the datafusion feature added but I still get that the TableProvider trait isn't implemented for DeltaTable. Its also not a pub trait in that file so I can't bring it into scope. Any thoughts?

the trait bound deltalake::DeltaTable: datafusion::datasource::TableProvider is not satisfied
the following other types implement trait datafusion::datasource::TableProvider:
datafusion::datasource::empty::EmptyTable
datafusion::datasource::listing::ListingTable
datafusion::datasource::streaming::StreamingTable
datafusion::datasource::ViewTable
datafusion::test_util::TestTableProvider
datafusion::datasource::MemTable
required for the cast from std::sync::Arc<deltalake::DeltaTable> to std::sync::Arc<(dyn datafusion::datasource::TableProvider + 'static)>

[theelderbeever]

Looks like it was a version issue. Datafusion 32.0.0 isn't compatible with delta-rs 0.16.0.

[AbrarNitk]

Have faced the same issue, It worked for me with the below config.

[dependencies]
tokio = { version = "1.37", features = ["full"] }
deltalake = { version = "0.17", features = ["s3", "datafusion", "datafusion-ext"] }
datafusion = { version = "35.0.0" }

[CHRISCARLON]

Have faced the same issue, It worked for me with the below config.

[dependencies]
tokio = { version = "1.37", features = ["full"] }
deltalake = { version = "0.17", features = ["s3", "datafusion", "datafusion-ext"] }
datafusion = { version = "35.0.0" }

Can confirm that this works with later versions of deltalake too.

I didn't have to specify datafusion directly either.

[dependencies] calamine = "0.25.0" comfy-table = "7.1.0" clap = { version = "4.5.7", features = ["derive"] } reqwest = { version = "0.12", features = ["blocking", "json"] } colored = "2.1.0" serde = "1.0.207" serde_json = "1.0.124" indicatif = "0.17.8" bytes = "1.7.2" csv = "1.3.0" aws-config = "1.5.8" aws-credential-types = "1.2.0" aws-sdk-sts = "1.39.0" aws-types = "1.3.3" **deltalake = { version = "0.20.1", features = [ "s3", "datafusion", "datafusion-ext", ] }** deltalake-aws = "0.3.0" tokio = { version = "1.39.3", features = ["full"] }

This actually took me a few hours to figure out so posting a code snippet below which works in case anyone else comes across this issue in the future!

@theelderbeever be keen to know what sort of stuff you're doing with DeltaLake tables :)

use aws_config;
use aws_config::BehaviorVersion;
use aws_sdk_sts::config::ProvideCredentials;
use colored::Colorize;
use deltalake::arrow::record_batch::RecordBatch;
use deltalake::datafusion::execution::context::SessionContext;
use deltalake::{open_table_with_storage_options, DeltaOps};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

// Load AWS Creds into a hashmap for use with delta lake reader
pub async fn get_aws_config() -> Result<HashMap<String, String>, Box<dyn std::error::Error>> {
    // Locate aws creds
    let config = aws_config::defaults(BehaviorVersion::latest())
        .retry_config(aws_config::retry::RetryConfig::standard().with_max_attempts(5))
        .timeout_config(
            aws_config::timeout::TimeoutConfig::builder()
                .operation_timeout(Duration::from_secs(30))
                .build(),
        )
        .load()
        .await;

    // Create new hashmap to store creds
    let mut aws_info = HashMap::new();

    // Add credentials to HashMap if available
    if let Some(creds_provider) = config.credentials_provider() {
        match creds_provider.provide_credentials().await {
            Ok(creds) => {
                aws_info.insert(
                    "AWS_ACCESS_KEY_ID".to_string(),
                    creds.access_key_id().to_string(),
                );
                aws_info.insert(
                    "AWS_SECRET_ACCESS_KEY".to_string(),
                    creds.secret_access_key().to_string(),
                );
                if let Some(session_token) = creds.session_token() {
                    aws_info.insert("AWS_SESSION_TOKEN".to_string(), session_token.to_string());
                }
            }
            Err(e) => return Err(format!("Failed to retrieve credentials: {}", e).into()),
        }
    } else {
        return Err("No credentials provider found in the configuration".into());
    }

    Ok(aws_info)
}

// Read DeltaLake table stored in S3
pub async fn load_remote_delta_lake_table_info(
    s3_uri: &str,
    credential_hash_map: HashMap<String, String>,
) -> Result<(), Box<dyn std::error::Error>> {
    // load credentials
    let storage_options: HashMap<String, String> = credential_hash_map;

    // register aws backend
    deltalake_aws::register_handlers(None);

    // open delta lake table
    let table = match open_table_with_storage_options(s3_uri, storage_options).await {
        Ok(tbl) => tbl,
        Err(_) => {
            let ops = DeltaOps::try_from_uri(s3_uri).await?;
            ops.create().with_table_name("data").await?
        }
    };

    // Start DataFusion context
    let ctx = SessionContext::new();

    // Register table
    // Register table
    ctx.register_table("data", Arc::new(table))
        .map_err(|e| format!("Failed to register table: {}", e))?;

    // Create batches
    let batches = ctx
        .sql("SELECT * FROM data")
        .await
        .map_err(|e| format!("SQL Query Failed: {}", e))?
        .collect()
        .await
        .map_err(|e| format!("Could not read results: {}", e))?;

    for batch in batches {
        println!("{}", "DeltaLake Output: Columns and RecordBatch.".green());
        process_record_batch(&batch);
        println!("Record Batch: {:?}", batch);
    }

    Ok(())
}

// Process DeltaLake table stored in S3
// Processes the returned Arrow RecordBatch
fn process_record_batch(batch: &RecordBatch) {
    println!("Number of columns: {}", batch.num_columns());
    println!("Number of rows: {}", batch.num_rows());

    let schema = batch.schema();

    for i in 0..batch.num_columns() {
        let field = schema.field(i);

        println!(
            "Column {}: '{}' (Type: {:?})",
            i,
            field.name(),
            field.data_type()
        );
    }
}

[theelderbeever]

@CHRISCARLON nice work getting that sorted out!

re:deltalake
We have since shelved the projects I was using them on but, the use cases were twofold.

1. Building out analytical analytic datasets for on chain data. Got intro'd to deltalake through a trial with Databicks. Due to more experience with Rust, the need to write custom function handlers (mathematics on uint256 datatypes), and a propensity to self host I was going build it out.
2. I have built out internal billing system which offloads a lot of our OLTP data that we have collected in a timeseries database to parquet tables for archival and further processing. Utilizing deltalake represented a UX/performance/integrity improvement over the raw parquet tables.

Pretty much only looking at doing #2 and possibly also streaming from Redpanda directly to delta tables. All that said I need to look at Iceberg as an option instead with some of the recent goings on in the datastore world.