Rust SDK

The Rust SDK crate spiceai is the easiest way to query Spice.ai from Rust. It is developed in the spice-rs repository.

It uses Apache Arrow Flight to efficiently stream data to the client and Apache Arrow Records as data frames.

Requirements

• Rust 1.93.1 or later (the crate is edition 2024)
• Tokio (every client method is async)

Installation

Add Spice SDK

cargo add spiceai

Note the crate is named spiceai, while the repository is named spice-rs. All imports use spiceai.

Usage

1. Create a client by providing your API key to ClientBuilder. Get your free API key at spice.ai.

use spiceai::ClientBuilder;

#[tokio::main]
async fn main() {
  let client = ClientBuilder::new()
    .api_key("API_KEY")
    .use_spiceai_cloud()
    .build()
    .await
    .unwrap();
}

ClientBuilder also accepts .flight_url() to set a custom endpoint, plus .user_agent(), .max_retries() (default 3), and .cache_control(). .flight_url() and .use_spiceai_cloud() are last-writer-wins, in either order.

For a Cloud connection with no other configuration, spiceai::Client::new("API_KEY").await is equivalent.

2. Execute a query and get back a stream of Apache Arrow record batches.

let mut flight_data_stream = client.sql("SELECT * FROM tpch.lineitem LIMIT 10;").await.expect("Error executing query");

The binding must be mut, because advancing the stream takes a mutable borrow.

3. Iterate through the stream to access the records. StreamExt must be in scope; the crate re-exports it.

use spiceai::StreamExt;

while let Some(batch) = flight_data_stream.next().await {
    match batch {
        Ok(batch) => {
            /* process batch */
            println!("{:?}", batch)
        },
        Err(e) => {
            /* handle error */
        },
    };
}

Parameterized queries

sql_with_params binds parameters supplied as an Arrow RecordBatch whose fields are named $1, $2, and so on. For a single row of scalar values, sql_with_bindings builds that batch from a QueryParameters list:

use spiceai::QueryParameters;

let mut stream = client
    .sql_with_bindings(
        "SELECT * FROM taxi_trips WHERE VendorID = $1 AND fare_amount > $2 LIMIT 10;",
        QueryParameters::new().push(1_i32).push(1.0_f64),
    )
    .await
    .expect("Error executing query");

From 5.0.0, QueryParameter also converts from chrono and std::time values: DateTime<Tz>, NaiveDateTime, NaiveDate, NaiveTime, SystemTime, std::time::Duration and chrono::TimeDelta. All except NaiveDate use QueryParameter::try_from, which returns an error for a value that Arrow cannot hold, such as a timestamp outside 1677 to 2262. The crate re-exports chrono as spiceai::chrono.

use std::time::Duration;
use spiceai::chrono::{TimeZone, Utc};
use spiceai::{QueryParameter, QueryParameters};

let since = Utc.with_ymd_and_hms(2024, 1, 31, 0, 0, 0).unwrap();
let params = QueryParameters::new()
    .push(QueryParameter::try_from(since).expect("Error converting the timestamp"))
    .push(QueryParameter::try_from(Duration::from_secs(30 * 60)).expect("Error converting the duration"));

let mut stream = client
    .sql_with_bindings(
        "SELECT * FROM trips WHERE pickup_time >= $1 AND trip_duration > $2",
        params,
    )
    .await
    .expect("Error executing query");

Asynchronous queries

query and query_with_bindings submit a query for asynchronous execution and return a QueryJob handle instead of a stream. They require the runtime to be running in distributed (scheduler) mode.

These methods go over HTTP rather than Flight. From 5.0.0, the HTTP endpoint follows the Flight endpoint: http://localhost:8090 for the local default, and https://data.spiceai.io with .use_spiceai_cloud(). For any other Flight endpoint, set .http_url(). Without it, query returns QueryError::HttpError before submitting anything. results() does not block: it returns QueryError::NotReady while the job is still pending or running, so wait on the handle first with wait() or wait_timeout().

let client = ClientBuilder::new()
  .http_url("http://localhost:8090")
  .build()
  .await
  .unwrap();

let job = client.query("SELECT * FROM taxi_trips").await.expect("Error submitting query");
job.wait().await.expect("Error waiting for the query");
let batches = job.results().await.expect("Error reading results");

Usage with local Spice runtime

Follow the quickstart guide to install and run spice locally. ClientBuilder defaults to the local runtime, so no endpoint is needed:

use spiceai::ClientBuilder;

#[tokio::main]
async fn main() {
  let client = ClientBuilder::new()
    .build()
    .await
    .unwrap();

  let data = client.sql("SELECT trip_distance, total_amount FROM taxi_trips ORDER BY trip_distance DESC LIMIT 10;").await;
}

Default endpoints

ClientBuilder defaults to the local runtime; Client::new(api_key) defaults to Spice Cloud. Override the Flight endpoint with .flight_url() and the HTTP endpoint with .http_url().

Contributing

Contribute to or file an issue with the spiceai crate at: https://github.com/spiceai/spice-rs