DataFusion

Add SQL to your application with Apache DataFusion: register Spiral's catalog once, then query any Table by its address, for example SELECT frame_id, camera FROM "frames@main". Everything runs in your process. DataFusion pushes column selection and simple filters into the Spiral Scan, so only the rows and columns your query needs leave storage, and it runs joins, sorts, and aggregation itself.

The integration runs above the Rust SDK, using libspiral for bounded scans and SPQL evaluation. SQL remains in DataFusion. The local Connect API uses this integration to serve SQL to Studio browsers using their API access token; the embedded runtime has no SQL engine.

For why Spiral leaves SQL to your engine, see Query engines.

Run SQL from the terminalCopy Anchor Link

The spiral sql command dispatches to spiral-sql, which opens a DataFusion session with SPQL functions. Install spiral-cli[sql] for both commands, or install spiral-sql and run it directly. The base Python SDK includes neither command; its cli and sql extras install the tools when needed.

spiral sql --project demo --format json \
  'SELECT frame_id, camera FROM frames LIMIT 10'
spiral sql --project demo --branch main --file queries.sql

Run spiral sql without SQL input in a terminal to open an interactive session. Add --project demo and --branch main to work with that catalog. End a statement with ;; unfinished statements continue on the next line. Queries share one session, including temporary DataFusion tables and settings. Use Up/Down or Ctrl-R for history, .help for help, and .quit, .exit or Ctrl-D to exit. Ctrl-C clears unfinished input or cancels the current query; query errors return to the prompt. History uses ~/.spiral_history, or the path in SPIRAL_HISTORY when set.

Pass SQL as an argument, with --command, with --file, or through standard input. --file - also reads standard input. Without --project, the session can query local files, DataFusion tables and SPQL functions without signing in. --project opens the Project catalog on --branch, which defaults to main. Table addresses such as "frames@main" select a different read reference; write targets omit @ref.

Results stream without a preview row limit. --format json emits one JSON array per statement, including explicit nulls. --format csv emits one header per statement. The default table format prints each result batch separately. Errors go to standard error. In batch mode, errors return a nonzero exit status. Ctrl-C cancels the current execution.

Statements in a file share one session and run sequentially. Execution stops at the first error. Each successful Table write publishes independently; a later error does not roll back earlier writes. Use an SDK commit for writes that must publish together.

Set upCopy Anchor Link

Add the spiral-datafusion crate with its sql and embedded features. The crate is in development and not yet on crates.io; for early access, contact hello@spiraldb.com. The API reference is at docs.rs/spiral-datafusion.

Build your SDK client with the embedded backend, as for any other read.

Query TablesCopy Anchor Link

Create a SpiralCatalogProvider from a Table session and open a DataFusion session from it. You can then query Tables by address without registering each one:

use std::error::Error;

use datafusion::arrow::util::pretty::print_batches;
use datafusion::execution::SessionStateBuilder;
use datafusion::prelude::{SessionConfig, SessionContext};
use futures::TryStreamExt as _;
use spiral::Client;
use spiral_datafusion::{SpiralCatalogProvider, with_spiral};

/// Query the frames table resolved from a pinned snapshot of the demo Project.
pub async fn query_frames(client: &Client) -> Result<(), Box<dyn Error>> {
    let tables = client.project("demo")?.table_session("main").await?;
    let catalog = SpiralCatalogProvider::new(tables);
    let context = catalog.session(SessionConfig::new());
    let mut batches = context
        .sql("SELECT frame_id, camera = 'front' AS is_front, UPPER(camera) AS camera FROM \"frames@main\" LIMIT 100")
        .await?
        .execute_stream()
        .await?;
    while let Some(batch) = batches.try_next().await? {
        println!("{} frames with is_front flags", batch.num_rows());
    }
    Ok(())
}

The catalog resolves Tables through the Table session, so every Table in a query reads one consistent version: the session pins each branch the first time the query names it. In this query, DataFusion sends the column selection, the camera = 'front' comparison, and the limit into the Scan, which reads only frame_id and camera and stops after 100 rows. DataFusion runs UPPER itself, because that is one of its own functions.

To see newer commits, call SpiralCatalogProvider::refresh() and open a new DataFusion session from the catalog it returns. Sessions you already opened keep reading the versions they pinned, unless they publish their own writes.

Name Tables in SQLCopy Anchor Link

A Table's name in SQL is its address, such as frames, frames@main, or frames@v019a…. Qualify another Project as "project-id"."path@ref"; the default Project can be omitted from the name.

Double quotes around an address always work. You can leave them off when the address contains only letters, digits, _, and @, so frames@main works without quotes. Quote any address with /, -, ., or other characters, such as "robot-a/frames" or "frames@feature-x". These are the identifier rules of DataFusion's default SQL dialect; a different dialect can change them.

catalog.session(config) turns off DataFusion's identifier normalization, so frames@Main is not lowercased into a different branch. If you register the catalog in a session of your own, turn off sql_parser.enable_ident_normalization yourself.

To resolve Projects and refs lazily, construct the catalog with SpiralCatalogProvider::from_client(client, default_project, default_ref). The default Project is optional. Qualify another Project as "project-id"."path@ref"; with a default Project, "path@ref" is sufficient. Each Project/branch pair pins on first use, and explicit resource refs override the default. Constant queries need no Project. These rules apply to the shared DataFusion integration, including local Connect SQL.

Write TablesCopy Anchor Link

A catalog opened on a branch publishes DataFusion table creation, INSERT INTO, INSERT OVERWRITE, DELETE FROM, UPDATE, MERGE INTO, TRUNCATE TABLE and table removal through SDK commits, one commit for each statement. New Tables use root names such as frames. Write targets omit @ref; select the branch when opening the catalog. Exact commit catalogs and manually registered immutable Table providers remain read-only.

Each successful operation advances the shared catalog to the branch's new head. Existing query plans keep their original versions. A write planned against an older head fails with a conflict rather than overwriting another writer's work. Refresh the catalog to select an external writer's commit before trying again.

Run synchronous catalog mutations on a multi-threaded Tokio runtime, or call them outside a Tokio runtime. DataFusion's insert execution streams batches into the SDK without collecting the entire input. For writes spanning several Tables in one commit, use an SDK commit.

Deletes return one count column containing the number of rows removed. They select visible rows from the pinned version and record deletions by stable row ID, preserving the IDs of retained rows. A predicate that evaluates to null does not select a row. Deleting no matching rows returns zero without publishing a commit. TRUNCATE TABLE uses the same deletion path without a predicate.

A delete predicate that translates exactly into an SDK filter, such as numeric, boolean and UTF-8 comparisons, boolean combinations and supported IN lists, deletes through that filter. DataFusion evaluates any other predicate, such as DELETE FROM frames WHERE frame_id % 10 = 0: it scans the visible rows with their row IDs, prunes the scan by the parts of the predicate an SDK filter expresses, and deletes the rows the whole predicate selects. Subqueries and joins in delete predicates are not supported.

UPDATE evaluates its WHERE clause and SET expressions in DataFusion, so any DataFusion expression can compute a new value, for example UPDATE frames SET camera = upper(camera) WHERE frame_id % 2 = 0. DataFusion casts each value to its column's stored type. The statement returns the number of updated rows, which keep their row IDs. The new values are recorded as patches beside the data, writing only the assigned columns. UPDATE ... FROM is not supported.

MERGE INTO joins its source with the Table in DataFusion. Rows that meet a WHEN MATCHED clause are updated, as patches, or deleted; source rows that meet a WHEN NOT MATCHED clause are appended; and Table rows that meet a WHEN NOT MATCHED BY SOURCE clause are updated or deleted. A row takes the first clause whose condition it meets. Every change publishes in one commit, and the statement returns the number of rows inserted, updated and deleted. A Table row that matches more than one source row fails the statement. The changed and inserted rows are held in memory until the join completes.

MERGE INTO frames AS f
USING new_frames AS n ON f.frame_id = n.frame_id
WHEN MATCHED THEN UPDATE SET camera = n.camera, captured_at = n.captured_at
WHEN NOT MATCHED THEN INSERT VALUES (n.frame_id, n.camera, n.captured_at, n.image, n.embedding)

Catalog-created sessions install the Spiral planner automatically. When adding a Spiral catalog to an existing session, install with_spiral or with_spql on its session builder. Deletes, updates and merges require this planner to keep their complete selection when DataFusion optimizes an empty one.

Writes keep independent DataFusion streams parallel and publish their fragments in one commit. Each stream can produce several fragments. The writer validates supported column ordering from the input plan and records it per fragment, including descending order and null placement. It computes statistics from the stored rows.

An existing Table's partitioning and clustering guide subsequent writes. DataFusion distributes and sorts supported identity columns before writing; the storage writer applies transformed partition keys and clustering in bounded runs. When the input already follows that layout, the writer skips sorting it again.

On reads, DataFusion can use the common ordering of stored fragments even when there are more fragments than execution streams. Each stream merges a group of fragments in order. Remote merges read resumable pages, so waiting on another fragment does not keep a request slot occupied. Scans without a common ordering share work dynamically.

Column selection, aliases, and appended computed columns preserve the usable ordering prefix. Updates preserve the prefix before the first changed sorting key. Supported floating-point ordering includes signed zeros, NaNs, and null placement. Nested and transformed clustering keys are applied in storage but are not yet exposed as DataFusion ordering expressions.

The integration exposes stored row counts, column bounds, and null counts to DataFusion. It distinguishes exact values from estimates and truncated bounds, and returns unknown for statistics it cannot prove after filters or updates. NaN counts determine whether floating-point bounds are usable; signed-zero bounds are conservative. Fixed-width output size estimates use Arrow types rather than compressed storage sizes. Eligible counts and extrema can use metadata without reading payloads; truncated bounds never replace exact aggregate results. Nested statistics, distinct counts, and histograms are not collected by the default write path.

DataFusion's current CREATE TABLE AS SELECT implementation materializes its input without preserving sort metadata. Those writes compute statistics and apply the destination layout, but cannot recover ordering that DataFusion has already discarded.

Call SPQL functionsCopy Anchor Link

with_spql adds the SPQL functions of the native library to a DataFusion session:

use datafusion::execution::SessionStateBuilder;
use datafusion::prelude::SessionContext;
use spiral_datafusion::with_spql;

/// Create a DataFusion session that runs SPQL functions through scandal.
pub async fn spql_session(scandal: Arc<dyn Scandal>) -> Result<SessionContext, Error> {
    let builder = SessionStateBuilder::new().with_default_features();
    let state = with_spql(builder, Functions::new(scandal).await?).build();
    Ok(SessionContext::new_with_state(state))
}

Quote function names in SQL, because they contain dots: "spql.str.upper"(camera). The functions work on any DataFusion input, not only on Spiral Tables, and you can use them in SELECT, WHERE, GROUP BY, ORDER BY, aggregate and window arguments, CASE, and join conditions. Arguments must already have a type that the function accepts; SPQL does not cast implicitly. Unit, timezone, frequency, and format controls that require literals must be constants in the SQL call; a column cannot supply them. Unsupported types, controls, or policies fail during planning. A valid call whose data contains an invalid local time fails during execution.

Standard SQL aliases also bind unquoted: ST_Contains(a, b) calls spql.st.contains. The spatial reference lists each alias and its source on the canonical entry. Plans and serialized expressions retain the canonical ID. Use spql.st.box for the bounding-box constructor, also available in SQL as ST_MakeEnvelope; the former spql.st.make_envelope ID has been removed.

Plan SPQL queries on a multi-threaded Tokio runtime. A single-threaded runtime cannot drive the SDK callbacks while DataFusion waits for function typing and returns a planning error.

Native Arrow dates and timestamps, including all four timestamp units and named timezones, can enter SPQL calls directly. Duration, Interval, and Period values must retain their SPQL extension metadata. A plain Arrow Duration or SQL INTERVAL literal does not acquire that metadata automatically. Period results retain it across SPQL projection steps, so a later call can read their frequency.

Arrow and Parquet preserve the following representations when the reader keeps the Arrow field metadata and Parquet's embedded Arrow schema:

These round trips preserve nullability and exact integer values, including negative epochs. Keep the embedded Arrow schema when writing Parquet: Parquet's logical types alone cannot describe every unit or SPQL value. A native Arrow interval is unsupported as an SPQL Interval input; export its calendar components in the metadata-bearing struct representation. Months remain calendar months, independently of the days and nanoseconds.

Arrow Duration has no day unit. A day-unit SPQL Duration needs an explicit Int64 field with its SPQL metadata; automatic native Arrow Duration export rejects it. A native Arrow Duration field whose unit conflicts with its SPQL metadata is rejected rather than relabelled or rescaled.

For a camera's local capture times, spql.dt.tz_localize interprets a timezone-free timestamp in a named timezone. It rejects times skipped by a clock change. Repeated times also fail unless you choose ambiguous as earliest or latest; the default is raise. The nonexistent argument currently accepts only raise, and unsupported policies fail before execution. spql.dt.tz_convert takes an already-zoned timestamp and changes its timezone while preserving the instant. Both functions preserve the input timestamp unit and nulls. See the datetime reference.

Calendar fields, predicates, to_date, to_time, to_period, and format use a timestamp's named timezone. A frame captured at 2024-03-01 00:30:45 UTC belongs to February 29 at 19:30:45 in New York:

WITH frames AS (
    SELECT "spql.dt.tz_convert"(
        "spql.dt.tz_localize"(TIMESTAMP '2024-03-01 00:30:45', 'UTC'),
        'America/New_York'
    ) AS captured_at
)
SELECT "spql.dt.to_date"(captured_at) AS capture_date,
       "spql.dt.format"(captured_at, '%H:%M:%S') AS capture_time
FROM frames

The result is capture_date = 2024-02-29 and capture_time = 19:30:45.

trunc, floor, ceil, and round also operate on local clock fields and preserve the timestamp unit and timezone. A 1d frequency describes a local clock day; the elapsed interval between consecutive midnights can be 23 or 25 hours when daylight saving time changes. Durations remain elapsed values. The result's offset is resolved at the new local time, so truncating a summer timestamp to the start of its year can select a winter offset.

If the resulting local time is repeated or skipped by a clock change, these calendar operations fail. They do not take the disambiguation policies of tz_localize. Halfway values in round choose the later local clock boundary. Null values remain null, and unrepresentable results fail instead of losing precision. Native SQL temporal functions run in DataFusion and follow its own semantics.

spql.add and spql.sub accept a date or timestamp and a typed spql.Interval or spql.Duration. Addition also accepts the operands in the opposite order. These are explicit SPQL calls; DataFusion owns the behavior of SQL's +, -, and native interval literals.

An Interval stores months, days, and nanoseconds separately. SPQL applies months first, clamping the day to the last valid day of the target month, then calendar days, then elapsed nanoseconds. For example:

Two successive one-month additions to January 31 produce March 29, because the first addition clamps to February 29. Subtracting a month does not recover a day lost by clamping. Subtraction negates all three components and applies them in the same order.

For a zoned timestamp, months and days modify its local date while keeping its local time. SPQL resolves the combined calendar result once and rejects a repeated or skipped time before applying elapsed nanoseconds. In New York, adding one calendar day to noon on March 9, 2024 produces noon on March 10, 23 elapsed hours later. Adding a Duration of 24 hours produces 13:00 on March 10.

This query combines DataFusion's calendar arithmetic, EXTRACT, and AT TIME ZONE with an explicit SPQL Duration:

WITH frames AS (
    SELECT "spql.dt.tz_localize"(
        TIMESTAMP '2024-03-09 12:00:00', 'America/New_York'
    ) AS captured_at
)
SELECT EXTRACT(HOUR FROM captured_at + INTERVAL '1 day') AS calendar_hour,
       "spql.dt.hour"("spql.add"(
           captured_at, "spql.dt.parse_duration"('24h')
       )) AS elapsed_hour,
       EXTRACT(HOUR FROM captured_at AT TIME ZONE 'UTC') AS utc_hour
FROM frames

It returns calendar_hour = 12, elapsed_hour = 13, and utc_hour = 17.

A zero calendar change preserves either occurrence of a repeated time, and elapsed arithmetic can cross a clock change.

Results retain the date or timestamp type, including timestamp unit and zone. A date result must be an exact whole day; timestamp results must fit their existing unit without losing precision. Overflow fails. Either null operand produces null; a non-null Interval must have non-null months, days and nanoseconds. These rules apply to SPQL-typed inputs; a bare SQL interval literal does not construct spql.Interval.

Use the optional errors control on parse_date, parse_time, and parse_timestamp to keep valid captures when some text is malformed:

SELECT "spql.dt.parse_timestamp"(captured_at, '%Y-%m-%dT%H:%M:%S', 'coerce') AS captured_at
FROM (VALUES ('2024-02-29T12:30:00'), ('invalid'), (NULL)) AS frames(captured_at)

Parsing raises an error by default. With errors = 'coerce', malformed values, precision loss and overflow produce null; null inputs always stay null. The result uses days for dates and nanoseconds for times and zone-free timestamps. Duration and Period parsing always raises errors for invalid values.

Formats and error policies are non-null literals checked during planning. Input formats support numeric %Y, %m, %d, %j, %H, %M, %S, %f (nanoseconds), and %%; other directives and modifiers are rejected. Timezone directives are unsupported: parse civil values, then localize them explicitly. Without a format, dates use year-month-day, times use hour:minute with optional seconds and up to nine fractional digits, and timestamps join the two with T or a space. A timestamp can end in one Z; its result remains zone-free. Use tz_localize to attach UTC when required. format supports the same numeric directives except %f; %% emits one literal percent sign. Formatting a field absent from the input type raises an error.

SPQL's current_date, current_time, current_timestamp, local_time, and local_timestamp sample the system clock once per input batch. They are volatile: values can differ between batches and between executions of a reused plan. DataFusion's native CURRENT_TIMESTAMP is fixed at query planning time.

Named timezone rules come from the native library's timezone database, which is part of the library version rather than of your query. Calendar reads, formatting, localization, rounding, conversion to periods, and calendar arithmetic are immutable: the same inputs on the same library version give the same result, so DataFusion can fold constants, reuse repeated subexpressions, and push comparisons on them into Scans. A stored timezone name does not pin a database version, so a historical result can change when you upgrade the library; if you cache calendar-derived results across deployments, invalidate them on upgrade. Fixed elapsed conversions and period-label boundaries do not consult timezone rules.

Period bounds are half-open dates: period_start is inclusive and period_end is exclusive:

WITH captures AS (
    SELECT "spql.dt.parse_period"('2024-02', 'ME') AS capture_month
)
SELECT "spql.dt.period_start"(capture_month) AS start_inclusive,
       "spql.dt.period_end"(capture_month) AS end_exclusive
FROM captures

Start- and end-labelled months, quarters and years cover the same calendar span. Weeks start on Monday. Use period_bounds to parse a label and obtain both boundaries; these are civil dates, so localize midnight explicitly when comparing them with zoned timestamps.

DataFusion does not call SPQL functions one row at a time. It groups each call, together with the surrounding expressions the native library supports, into one step that sends whole Arrow batches to the library. When that step reads directly from a Spiral Scan, the Scan computes the results next to storage instead. EXPLAIN shows these steps as SpqlProjectExec, with the outputs that a Scan computes.

Both forms include the shared physical projection description: selected implementations, execution dependencies, types and transfers. Planning labels it as a preview. After execution, EXPLAIN ANALYZE also shows the actual plan and execution ID observed in each partition. Compiler decisions distinguish an explicit placement mismatch from a decline for which no reason was supplied. See plan inspection for the structured format, identity scope and omitted literal values.

Blob functions, such as spql.io.fetch, read with the storage access of the Spiral Scans below them, so those Scans must come from one SDK client.

What runs in the ScanCopy Anchor Link

A pushed-down expression gives the same result it would give in DataFusion, including floating-point comparisons with signed zeros and NaN values. Where results would differ, DataFusion keeps the work: its default integer arithmetic wraps on overflow while a Scan fails, so that arithmetic stays in DataFusion. Mixed projections keep their output order, aliases, nullability, and metadata.

Temporal scan comparisons require exactly matching Arrow types, units and timezone names. Half-open ranges use an inclusive lower bound and exclusive upper bound. Table statistics can prune fragments outside those bounds. Timezone conversion, mixed-unit coercion, metadata-bearing semantic types and SPQL predicates remain in DataFusion. A residual predicate still runs after the safe temporal constraint, so pruning preserves the final results.

Observe scan executionCopy Anchor Link

EXPLAIN ANALYZE enables native counters for its execution and reports a readable summary for each operator. The query must contain a Spiral Scan. The Scan summary reports returned rows, direct storage reads, and windows skipped by statistics. The returned row count belongs to the Scan, before DataFusion applies residual operations. The summary uses decimal byte units and separates storage reads from output bytes.

EXPLAIN ANALYZE VERBOSE reports exact native counters in separate groups. It includes resource identities and DataFusion counters for each partition. Both views report unavailable counters explicitly. Zero means that the counter is available and its value is zero.

Catalog sessions and with_spql enable these views automatically. For manual Table registration, install the views on the session builder:

/// Create a session with the Spiral text views for explain results.
pub fn explain_session(config: SessionConfig) -> SessionContext {
    let builder = SessionStateBuilder::new()
        .with_config(config)
        .with_default_features();
    SessionContext::new_with_state(with_spiral(builder).build())
}

Compare both views for a query over the demo frames Table:

/// Report a summary and detailed scan counters for the demo frames Table.
pub async fn explain_frames(client: &Client) -> Result<(), Box<dyn Error>> {
    let tables = client.project("demo")?.table_session("main").await?;
    let context = SpiralCatalogProvider::new(tables).session(SessionConfig::new());
    for sql in [
        "EXPLAIN ANALYZE SELECT COUNT(*) FROM \"driving/frames@main\" WHERE frame_id > 100",
        "EXPLAIN ANALYZE VERBOSE SELECT COUNT(*) FROM \"driving/frames@main\" WHERE frame_id > 100",
    ] {
        let batches = context.sql(sql).await?.collect().await?;
        print_batches(&batches)?;
    }
    Ok(())
}

with_spiral preserves the application query planner set on the builder before the helper call. Explicit metric selections and output formats retain DataFusion's renderer. Queries without a Spiral Scan also retain DataFusion's renderer.

Storage counters describe one shared scan execution. They appear once without a partition identifier. Each DataFusion partition reports its own output_rows and elapsed_compute. Execution lanes take work dynamically, so storage requests do not belong to fixed DataFusion partitions.

The embedded backend reports direct storage and blob reads. StorageCounters::scope is StorageScope::Direct for these counters. They exclude reads through shared file handles, including payload footers and data segments. The summary labels those reads Shared Vortex I/O: unavailable. Direct counters are partial counts, not total query storage counts.

Complete backend counters use StorageScope::Complete and the storage_* metric names. Both scopes exclude metadata that the Table loaded before execution.

The Table retains file handles and completed data in memory. Each measured execution uses fresh counters for direct reads and scan work. Ordinary scans do not allocate native counters or update them.

The first execution that loads a shared entry through direct storage I/O receives its direct read counts. Later executions reuse that entry without extra storage reads. Concurrent executions charge each direct cache miss to the execution that fetches it. Cached segments do not add storage reads.

Size requests, signing requests, failed reads, and transport retry attempts do not enter these counters.

These durations run from request or batch submission to each successful result that the execution observes. The sum includes queue time and delays before the execution receives a completed result. Concurrent reads overlap, so storage_read_time can exceed the scan wall time. It is not the wall time that the scan waits for storage.

DataFusion's elapsed_compute measures stream polling and does not measure native decoding time.

Window counters appear for scans that reconstruct nested columns or column attachments. The direct Vortex scanner does not expose its internal window decisions. For those scans, the SDK returns None in ScanMetricsSnapshot::windows. The Spiral views report that window counters are unavailable. DataFusion's standard renderer omits those metrics. A filter that returns no rows does not necessarily skip a window through statistics.

For SDK counter collection, see Reading.

Differences from plain DataFusionCopy Anchor Link

• Integer arithmetic inside an SPQL step fails on overflow, where DataFusion's wraps.
• A join condition can call an SPQL function over columns from both sides only in an inner join.
• An SPQL step can return the rows of one input batch in more than one output batch.
• SPQL predicates are not yet pushed into Scan filters. The Scan computes the predicate's value and DataFusion applies the filter.

TroubleshootingCopy Anchor Link

Profile SPQL executionCopy Anchor Link

Run EXPLAIN ANALYZE in a session configured with with_spql to collect the executed projection descriptions and aggregate profiles. Reports use the same plan and node IDs as plain explain and include the consuming DataFusion partition. This covers standalone SPQL projections and projections folded into Table scans. Plain EXPLAIN does not enable profiling.

The profiling counter scopes distinguish host elapsed time from device time and buffer estimates from total memory. Inline storage execution reports its scan wait and output, while its internal kernel timing remains unknown.