Reading

A Scan reads one exact Table version as a stream of Arrow record batches, with no SQL engine in between. Because the version never changes, running the same Scan again returns the same rows. You choose the columns, filter the rows, and compute values inside the Scan, so only the data your application needs reaches it.

For the step-by-step task, see Read rows. This page covers the Rust options and how a Scan behaves.

Open and consume a ScanCopy Anchor Link

Open a Table through a session, then call table.scan(options). The Scan fetches the Table's metadata and storage access once, then fixes its columns, filter, output schema, and backend. From then on:

• scan.schema() returns the output schema without another request.
• scan.into_stream() returns the batch stream. Reading starts when you poll it, and dropping it stops the Scan.
• scan.resolved_address() returns the exact path@v<version> that the Scan reads.
• Cloning a Scan starts another read of the same version with the same options.

Embedded scans yield to your application's async runtime while waiting for scan planning or the next batch. Dropping a pending read cancels it.

Scan optionsCopy Anchor Link

Without partitioned, a Scan returns rows in stored order.

Compute outputsCopy Anchor Link

Set projection to compute values before the rows reach you. This example returns id and is_front, and never reads the image or embedding columns:

let tables = client.project("demo")?.table_session("main").await?;
let table = tables.table("frames@main").await?;
let scan = table
    .scan(ScanOptions {
        projection: Some(vec![
            NamedExpression {
                name: "id".into(),
                expression: Expression::new(col("frame_id"))?,
            },
            NamedExpression {
                name: "is_front".into(),
                expression: Expression::new(eq(col("camera"), lit("front")))?,
            },
        ]),
        limit: Some(100),
        ..Default::default()
    })
    .await?;
println!("output schema: {}", scan.schema());
let mut batches = scan.into_stream();
let mut rows = 0;
while let Some(batch) = batches.next_batch().await {
    rows += batch?.num_rows();
}

Build expressions with the builders re-exported in spiral::vortex::expr (from Vortex, the columnar file format Spiral stores Tables in), and wrap each one with Expression::new. Expressions support fields, literals, explicit casts, arithmetic, comparisons, boolean logic, null checks, and SPQL function calls.

• Each output reads the Table's own fields. It cannot read another output.
• Output names must be unique and not empty.
• An empty projection returns batches with zero columns and the original row counts.
• A filter uses the same expressions and also reads Table fields, not output names.

Operand types must match exactly; add an explicit cast when they differ. Type errors and unsupported functions fail when the Scan opens, before any rows stream. Errors that depend on the data, such as a failed cast or an integer overflow, end the batch stream.

Call SPQL functionsCopy Anchor Link

SPQL is Spiral's function library. Call a function with Expression::call, its full name, and the argument expressions, for example Expression::call("spql.str.upper", vec![camera]). Pass parameters such as a format or a unit as literal expressions. Calls work in both outputs and filters.

Read blob payloadsCopy Anchor Link

A Table can store images, video, and other large values as blobs. In outputs, spql.io.fetch returns a blob's payload as an inline blob, whose bytes spql.blob.bytes returns, and spql.io.stat returns its metadata. They work for blobs stored inline in the Table and for blobs stored in the Table's own storage, and the Scan reads both with the same storage access it uses for rows. A full read checks the blob's declared size and digest; a range read returns bytes at offsets within the blob.

A Scan cannot read a blob that points outside the Table's storage, because it has no storage access there. Reads from Locations and Connections are planned.

Read in parallelCopy Anchor Link

For a training loader or a bulk export, set partitioned and max_streams, then read the streams from scan.into_streams() on separate tasks:

let tables = client.project("demo")?.table_session("main").await?;
let table = tables.table("frames").await?;
let scan = table
    .scan(ScanOptions {
        partitioned: true,
        max_streams: Some(4),
        columns: Some(vec!["frame_id".into(), "camera".into()]),
        ..Default::default()
    })
    .await?;
let lanes = scan.into_streams().into_iter().map(|lane| async move {
    let mut batches = lane.into_stream();
    let mut rows = 0;
    while let Some(batch) = batches.next_batch().await {
        rows += batch?.num_rows();
    }
    Ok::<_, Error>(rows)
});
let rows: usize = try_join_all(lanes).await?.into_iter().sum();

All streams pull from one shared read. Each stream takes the next batch when its consumer is ready, so a slow consumer does not hold the others back, and a fast one simply reads more. The number of streams does not depend on how the Table is laid out in storage. scan.stream_count() returns how many streams you got.

A Scan with a limit always uses one stream in stored order, so the limit applies to the whole Table. Dropping the last stream cancels the remaining work.

Read splits as units of workCopy Anchor Link

Distributed jobs, such as a training loader with many workers, need each worker to read its own rows and to resume where it stopped. A Scan's splits are the units of work for that: scan.splits() lists them in stored order, each a range of stored rows of one part of the Table version. A work item is a split, or a range of its rows, and scan.read_work reads a list of them in order. Your application decides which worker reads which items; this example stripes whole splits across workers:

let tables = client.project("demo")?.table_session("main").await?;
let table = tables.table("frames").await?;
let scan = table
    .scan(ScanOptions {
        columns: Some(vec!["frame_id".into(), "camera".into()]),
        ..Default::default()
    })
    .await?;
// Stripe whole splits across workers; any other assignment works the same way.
let work = scan
    .splits()
    .iter()
    .enumerate()
    .skip(worker)
    .step_by(workers)
    .map(|(split, planned)| WorkItem {
        split,
        rows: planned.rows.clone(),
    })
    .collect();
let mut batches = scan.read_work(work).await?;
let mut rows = 0;
while let Some(batch) = batches.next_batch().await {
    let batch = batch?;
    rows += batch.batch.num_rows();
    // Save (batch.item, batch.next_row) to resume this share without rereading rows.
}

When every worker computes its list from the same splits, no two workers read the same rows. Any other assignment, such as a shuffle, reads the same way.

Each batch carries the index of its work item and next_row, the stored row after the rows it covered. To resume, read the item again from that row; no row is read twice. scan.count_rows(&work) returns the exact rows each item returns after the Scan's filter, and reads only the filter's columns. Scans with a limit read no work items.

Take rows by positionCopy Anchor Link

A training loader with its own sampler, such as a shuffled epoch that a data loader draws from, needs exactly the rows it drew, not a range of them. scan.take(takes) reads rows of the Scan's splits by position and returns them in the order you asked for. Position p of a split is its p-th row in stored order, counting only the rows the version keeps, so scan.count_rows over a whole split tells your sampler how many positions the split holds:

let tables = client.project("demo")?.table_session("main").await?;
let table = tables.table("frames").await?;
let scan = table
    .scan(ScanOptions {
        columns: Some(vec!["frame_id".into(), "image".into()]),
        ..Default::default()
    })
    .await?;
// A split holds as many positions as it has rows the version keeps.
let whole = scan
    .splits()
    .iter()
    .enumerate()
    .map(|(split, planned)| WorkItem {
        split,
        rows: planned.rows.clone(),
    })
    .collect::<Vec<_>>();
let lengths = scan.count_rows(&whole).await?;
// Your sampler draws the positions; this one takes the last and middle row of each split.
let takes = lengths
    .iter()
    .enumerate()
    .filter(|(_, rows)| **rows > 0)
    .map(|(split, rows)| Take {
        split,
        positions: vec![rows - 1, rows / 2],
    })
    .collect();
let mut batches = scan.take(takes).await?;
let mut rows = 0;
while let Some(batch) = batches.next_batch().await {
    rows += batch?.num_rows();
}

Each Take names a split by its index in scan.splits() and lists positions in any order. Rows come back take by take, each take's rows in the order of its positions, and a repeated position returns its row again. Plan the Scan once and take from it as often as you need: each take reads only the rows it names, with no planning or filtering in between.

Each split's rows are read in one pass, so positions that lie close together in storage share their reads. Group the positions you draw by split, so that each split is read once per call. A take reads whole stretches of storage, not single rows, so its cost follows how many stretches the positions touch: in a Table of 1 KiB rows, one row reads about 1 MiB. That is still far less than a filter that selects the same rows, which reads each stretch of keys separately and plans again on every call.

One request takes at most 65,536 positions from at most 256 splits. The SDK sends a larger take as several requests, and resumes a failed request from the rows it already delivered. A position past the rows of its split fails take before any rows arrive. A Scan with a filter, limit, or emit cannot take rows, because a position would then depend on every row before it.

Collect execution countersCopy Anchor Link

ScanOptions::collect_metrics enables counters for a Scan, embedded or remote. The scan plan reports fragment and file pruning, and each split read reports its storage reads, windows and output rows when it finishes, so counters grow as splits complete. Ordinary Scans do not allocate counters or update them.

Retain the metrics handle before you consume a batch stream:

let tables = client.project("demo")?.table_session("main").await?;
let table = tables.table("driving/frames@main").await?;
let scan = table
    .scan(ScanOptions {
        collect_metrics: true,
        ..Default::default()
    })
    .await?;
let mut batches = scan.into_stream();
let metrics = batches.metrics();
while let Some(batch) = batches.next_batch().await {
    drop(batch?);
}
let counters = metrics.snapshot()?;
if let Some(storage) = counters
    .as_ref()
    .and_then(|counters| counters.storage.clone())
{
    println!("storage bytes: {}", storage.bytes);
    println!("storage requests: {}", storage.requests);
} else {
    println!("storage counters are unavailable");
}

BatchStream::metrics() and ScanStream::metrics() return the same counters for all lanes of one execution. ScanMetrics::snapshot() returns None before the first poll or when the backend supplies no counters.

ScanMetricsSnapshot::storage contains Some(StorageCounters) when the reader supplies counters for the declared scope. A measured zero remains inside Some. Unavailable storage counters use None. The remote backend does not supply storage counters.

The metrics handle survives stream cancellation. A cloned Scan starts a fresh execution with separate counters.

For counter scopes and metric meanings, see DataFusion.

LimitsCopy Anchor Link

• The Scan reads input in chunks of about 32 MiB. A blob payload or an output row larger than 64 MiB fails.
• A Scan's streams cannot be shared across processes; divide its splits into work items instead.
• For SQL, joins, and aggregation, use the DataFusion integration.

Project snapshot and catalog reads require catalog:read; they do not initialize a journal. Creating a Project initializes its journal, so readers can select main as soon as the Project exists.