DataFusion Session

Use this integration when your Rust service already owns request handling, authentication, catalogs, and SessionContext lifecycle. Algeon adds native planning and GPU execution without taking over those application concerns.

Dependencies

Use the DataFusion version pinned by Algeon:

[dependencies]
datafusion = "55.0"
algeon-datafusion = { path = "../algeon/crates/algeon-datafusion", features = ["cugraph"] }
futures = "0.3"

Algeon is currently consumed from the algeon-datafusion crate in the repository. Clone the repository recursively as described in Building from Source, then point the application at crates/algeon-datafusion. Its component crates are resolved through the root workspace.

The algeon-datafusion crate has no default features. Add cugraph for graph SQL, cuvs for vector SQL, iceberg for Iceberg sources, and nvml for optional device diagnostics. Omit all of them for relational cuDF execution only.

Build one backend

AlgeonGpuBackend is the process-scoped owner of CUDA devices, admission, per-device ledgers, and execution resources:

use datafusion::execution::SessionStateBuilder;
use datafusion::prelude::SessionContext;
use algeon_datafusion::backend::{
    GpuMemoryOwnership, AlgeonGpuBackend, AlgeonGpuDeviceConfig, AlgeonGpuDeviceProfile,
};

let device = AlgeonGpuDeviceConfig::new(AlgeonGpuDeviceProfile::new(0), GpuMemoryOwnership::WholeDeviceExclusive);
let backend = AlgeonGpuBackend::builder()
    .devices([device])
    .build()?;

let state = backend
    .install_on(SessionStateBuilder::new_with_default_features())?
    .build();

let ctx = SessionContext::new_with_state(state);

Every selected ordinal needs one profile and one matching ownership declaration. AlgeonGpuDeviceProfile::new(0) uses automatic capacity and one active attempt; add explicit limits only when your deployment requires them.

Only one live backend may exist in a process. Clone and share this backend across every session and frontend instead of building another one.

Add cuGraph and cuVS SQL

install_on installs backend-bound native rules and coverage. It does not install cuGraph SQL or cuVS SQL. To enable the cugraph_* table functions on the session, chain try_with_cugraph_sql(config):

use algeon_datafusion::{
    cugraph_sql::CugraphSqlConfig,
    session::AlgeonSessionStateBuilderExt,
};

let state = backend
    .install_on(SessionStateBuilder::new_with_default_features())?
    .try_with_cugraph_sql(CugraphSqlConfig::default())?
    .build();

let ctx = SessionContext::new_with_state(state);

cuGraph SQL functions resolve the backend's admission and native execution configuration from session state when executed. To add the cuvs_* functions, chain try_with_cuvs_sql() in the same manner; it takes no configuration. The functions execute only when the crate was built with the cuvs feature. See cuVS execution.

Execute one attempt per query

For application-owned endpoints, create an attempt before planning, take its completion receiver before running the query, and read the report whether or not the query succeeded. The snippet assumes an orders table is already registered in ctx, for example with CREATE EXTERNAL TABLE ... STORED AS PARQUET or register_parquet:

use algeon_datafusion::attempt::{AttemptOptions, QueryAttemptReport};
use futures::TryStreamExt;
use std::sync::Arc;

let attempt = backend.begin_attempt(AttemptOptions {
    query_id: Some("request-42".to_owned()),
    ..Default::default()
})?;
let completion = attempt.completion();

let result = async {
    let stream = attempt.sql(&ctx, "SELECT SUM(amount) FROM orders").await?;
    stream.try_collect::<Vec<_>>().await
}
.await;

let report: Arc<QueryAttemptReport> = completion.await;
let batches = result?;

The attempt clones the session into invocation-local planning state. Its report records the final disposition, admission grant, timing, and terminal outcome. Completion seals on stream EOF, error, cancellation, or drop. Taking the receiver before planning and awaiting it after the result is what keeps the report available on the error path; returning early with ? on the stream would drop the attempt and route the report to the abandoned-report queue.

Frameworks may own this boundary for you. The Algeon Flight SQL server, for example, begins the attempt while issuing a statement ticket and carries it through DoGet. Call begin_attempt directly when your application owns the query invocation.

Admission configuration

Device capacity belongs to profiles; shared queue behavior belongs to AlgeonGpuQueuePolicy. This fragment replaces the backend construction shown above:

use algeon_datafusion::backend::{
    GpuDeviceMemoryLimit, GpuMemoryOwnership, AlgeonGpuBackend, AlgeonGpuDeviceConfig,
    AlgeonGpuDeviceProfile, AlgeonGpuQueuePolicy,
};

let device = AlgeonGpuDeviceConfig::new(
    AlgeonGpuDeviceProfile::new(0)
        .with_memory_limit(GpuDeviceMemoryLimit::Explicit {
            limit_bytes: 8 * 1024 * 1024 * 1024,
        })
        .with_backend_reserve_bytes(512 * 1024 * 1024)
        .with_default_query_limit_bytes(512 * 1024 * 1024)
        .with_max_active_attempts(10),
    GpuMemoryOwnership::WholeDeviceExclusive,
);
let backend = AlgeonGpuBackend::builder()
    .devices([device])
    .queue_policy(
        AlgeonGpuQueuePolicy::default()
            .with_max_queued_attempts(64)
            .with_max_queue_wait(std::time::Duration::from_secs(30))
    )
    .build()?;

Planning does not acquire a GPU grant. Admission begins when the first native stream is polled, and all native nodes in that attempt share the selected device and immutable grant. A full queue or expired wait returns structured service_overloaded backpressure. Eligible waiters are served in FIFO order; there is no overtaking policy.

Native execution policy is owned solely by the backend builder. Configure native execution with with_native_execution_config(...) on AlgeonGpuBackend::builder() before build().

DataFusion sessions carry standard DataFusion configuration only; native execution policy and graph input bounds cannot be altered via SQL SET or session extensions.

Choose an execution mode

Set AlgeonGpuExecutionMode on the shared backend before installing sessions:

Rust mode	Behavior	Use it when
NativePreferred (default)	Select supported relational SQL for native GPU execution; retain DataFusion for unsupported or unselected candidates. Explicit GPU functions still run on GPU.	You want GPU acceleration for ordinary SQL as well as graph and vector functions.
FunctionsOnly	Keep ordinary relational SQL on DataFusion CPU; run explicit cugraph_* and cuvs_* functions on GPU.	You want DataFusion to scan, join, filter, and aggregate around GPU algorithms.
NativeRequired	Attempt native relational selection, then reject any completed plan containing DataFusion CPU execution.	You need to enforce a plan with no DataFusion CPU operators.

This replaces the backend construction above, using the same device:

use algeon_datafusion::planner::AlgeonGpuExecutionMode;

let backend = AlgeonGpuBackend::builder()
    .with_execution_mode(AlgeonGpuExecutionMode::FunctionsOnly)
    .devices([device])
    .build()?;

Install the required function families with the features and session methods shown above. FunctionsOnly changes relational planning; it does not register functions or remove their GPU runtime and backend requirements. The Flight SQL equivalents are documented in Configuration.

Compose CPU and GPU work

In FunctionsOnly, a CPU relation can feed a GPU function whose Arrow results return to CPU SQL. A GPU result can also pass through CPU SQL into another GPU function. cuGraph inputs remain registered table or view names; cuVS inputs remain parenthesized subqueries. Supported compositions share the same backend admission and immutable query grant.

CPU-to-GPU input boundaries stage Arrow data in host memory before upload. The current staging path does not spill: cuGraph host input and output staging each default to a 1 GiB ceiling; cuVS host input staging has a cumulative 1 GiB ceiling across one operation's inputs. DataFusion's memory pool can impose a lower limit. These are host staging limits, separate from the GPU query cap; keep intermediate inputs bounded. See the runnable execution modes example for PageRank with CPU filtering and ordering.

DataFusion fallback and GPU-only plans

An explicit GPU function has no CPU algorithm fallback. Unsupported function inputs, memory exhaustion, cancellation, and runtime failures remain errors; after GPU execution starts, Algeon never replays the query on CPU. See Handling Errors for the structured error contract. NativeRequired checks the completed plan without replanning under a different policy; it does not make unsupported operations GPU-capable.

Use GPU Coverage Validation to inspect the exact final disposition before execution. validate_query(&ctx, sql) is the Rust entry point; algeon_explain_coverage(...) exposes the same planning evidence through SQL.

Observe and shut down

Per-attempt evidence comes from QueryAttemptReport. The backend exposes service-level admission_snapshot, device_ledger_reports, device_memory_failure_reports, and backend_metrics_snapshot methods. At shutdown, stop accepting work, drain streams and sessions, then call close_and_wait on the shared backend.

See the runnable attempt lifecycle example for the complete per-query path, or the Embedded Backend Example for one backend shared by REST and Flight SQL.

Direct optimizer install

The raw session extension is useful for planning and coverage tests:

use datafusion::execution::SessionStateBuilder;
use algeon_datafusion::{
    planner::AlgeonNativeOptimizerConfig,
    session::AlgeonSessionStateBuilderExt,
};

let state = SessionStateBuilder::new_with_default_features()
    .with_algeon_native(AlgeonNativeOptimizerConfig::default())
    .build();

This path has no admission owner. Native execution fails locally with gpu_backend_required, so production GPU execution should always install a AlgeonGpuBackend.