Delta Kernel Rust User Guide

Delta Kernel is a Rust library for building Delta Lake connectors. It handles the Delta protocol so you don’t have to. Connectors read and write Delta tables through Kernel’s API without needing to understand the protocol internals. When the protocol evolves, updating your Kernel dependency is all that’s needed to pick up new features.

Kernel is query-engine agnostic. It provides a native Rust API and a C/C++ FFI layer, making it usable from virtually any language.

Note

This guide is a work in progress.

Architecture at a glance

     ┌──────────────────────────────────────────┐
     │                Connectors                │
     │  (Query Engines, Analytics Tools, etc.)  │
     └───────────┬──────────────────────┬───────┘
                 │                      │
     ┌───────────▼─────────┐   ┌────────▼───────┐
     │    Rust Bindings    │   │  FFI Bindings  │
     │  (Native Rust API)  │   │  (C/C++ API)   │
     └────────────────┬────┘   └─┬──────────────┘
                      │          │
                  ┌───▼──────────▼─┐
                  │  Delta Kernel  │
                  │  (core logic)  │
                  └───────┬────────┘
                          │ calls into
                  ┌───────▼────────┐
                  │  Engine trait   │
                  │  (abstraction)  │
                  └───────┬────────┘
                          │ implemented by
                  ┌───────▼────────┐
                  │  DefaultEngine │
                  │  (or custom)   │
                  └───────┬────────┘
                          │
                  ┌───────▼────────┐
                  │  Delta Table   │
                  │  (storage)     │
                  └────────────────┘

The Engine trait is the boundary between Kernel and your connector. Kernel defines what needs to happen (read JSON, read Parquet, evaluate expressions); the engine defines how. A batteries-included DefaultEngine is provided for common use cases. See Architecture Overview for details.

Key APIs

Snapshot is a point-in-time view of a Delta table. Every operation starts here: reading the schema, scanning data, or starting a Transaction.

Scan reads data from a table. It supports predicate pushdown for file skipping and column projection. See Building a Scan.

Transaction writes data to a table. It supports creating tables, blind appends, and committing changes atomically. See Creating a Table and Appending Data.

CheckpointWriter compacts the transaction log into a checkpoint for faster reads. See Checkpointing.

Data types and schema

Kernel defines its own protocol-compliant type system, independent of any engine’s type system. This includes primitive types (integers, strings, timestamps, decimals, etc.) and complex types (structs, arrays, maps). The Kernel schema is the source of truth for a table’s structure. Your engine converts to and from it as needed.

FFI layer

The delta_kernel_ffi crate exposes the full Kernel API to C and C++ via a stable FFI boundary. Headers (.h and .hpp) are generated automatically at build time using cbindgen. Rust objects cross the boundary as opaque handles with clear ownership semantics, and every fallible function returns a structured error type.

This means you can build a Delta connector in C, C++, or any language with a C FFI without writing any Rust. See the FFI overview for details.

Design principles

1. Protocol abstraction. Kernel encapsulates the Delta protocol. Connectors pick up new protocol features by updating their Kernel dependency.
2. Engine-agnostic. Kernel defines what to do; engines define how. The Engine trait is the only integration point.
3. Feature flag modularity. Core functionality works without optional dependencies. Pay only for what you use via Cargo feature flags.
4. Clear I/O boundaries. APIs clearly indicate when I/O operations occur, giving connectors control over scheduling and parallelism.

Crate structure

Getting started

For Rust projects, add to Cargo.toml:

delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow"] }

For C/C++ projects, build the FFI crate and link against it. See the FFI overview.

Then follow the quick starts to see Kernel in action.

What’s next

• Quick Start: Reading a Table
• Quick Start: Writing a Table
• Architecture Overview

Installation

delta_kernel is available on crates.io and uses Cargo feature flags to keep the core dependency-light.

Requirements

• Rust edition: 2021
• Minimum Rust version: 1.88

Adding the dependency

Add delta_kernel to your Cargo.toml:

[dependencies]
delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow"] }

This gives you the default engine (which handles all I/O and expression evaluation for you) backed by Arrow, with rustls for TLS. This is the recommended starting point for most users.

Feature flags

You only pay for what you enable.

Engine features

These enable the built-in DefaultEngine, which provides out-of-the-box support for reading and writing Delta tables using Arrow and the object_store crate.

You need exactly one of default-engine-rustls or default-engine-native-tls to use the default engine. If you’re building a custom engine, you may not need either. See Building a Connector for details.

Arrow version pinning

If you need a specific Arrow version (e.g. to match your existing Arrow dependency), you can pin it explicitly:

For more details on managing Arrow version compatibility, see Feature Flags.

Data features

These are pulled in automatically by the default engine features. You typically only need to specify them directly if you’re building a custom engine that still uses Arrow.

Advanced features

Example Cargo.toml

A typical project using delta kernel:

[package]
name = "my-delta-reader"
version = "0.1.0"
edition = "2021"

[dependencies]
delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow"] }

# The kernel re-exports arrow, but you can also depend on it directly
# arrow = "58"

What’s next

With the dependency added, head to Quick Start: Reading a Table to read your first Delta table.

Quick Start: Reading a table

In this tutorial, you will read a Delta table and get the data as Arrow RecordBatches using the default engine.

Create a new project

cargo new delta_read_example
cd delta_read_example

Add the kernel dependency (see Installation for details):

cargo add delta_kernel -F default-engine-rustls -F arrow -F internal-api

Your Cargo.toml should look like:

[dependencies]
delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow", "internal-api"] }

Note

The internal-api feature exposes try_parse_uri, a convenience function used in this tutorial and throughout the guide. This feature flag may be removed in a future release once the API stabilizes.

Write the code

Replace src/main.rs with the following. We’ll walk through each piece below.

Filename: src/main.rs

extern crate delta_kernel;
use std::sync::Arc;

use delta_kernel::arrow::util::pretty::print_batches;
use delta_kernel::engine::arrow_data::EngineDataArrowExt as _;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::{DeltaResult, Snapshot};

fn main() -> DeltaResult<()> {
    // 1. Parse the table location
    let table_path = std::env::args().nth(1).expect("usage: delta_read_example <TABLE_PATH>");
    let url = delta_kernel::try_parse_uri(&table_path)?;

    // 2. Build an object store and engine
    let store = store_from_url(&url)?;
    let engine = DefaultEngine::builder(store).build();

    // 3. Get a snapshot of the table at the latest version
    let snapshot = Snapshot::builder_for(url).build(&engine)?;
    println!("Table version: {}", snapshot.version());
    println!("Schema:\n{}", snapshot.schema());

    // 4. Build and execute a scan
    let scan = snapshot.scan_builder().build()?;
    let mut batches = vec![];
    for data in scan.execute(Arc::new(engine))? {
        let record_batch: delta_kernel::arrow::record_batch::RecordBatch =
            data?.try_into_record_batch()?;
        batches.push(record_batch);
    }

    // 5. Print the results
    print_batches(&batches)?;
    Ok(())
}

Step by step

1. Parse the table location

let url = delta_kernel::try_parse_uri(&table_path)?;

try_parse_uri converts a path string (local path or URI like s3://bucket/path) into a Url.

2. Build an object store and engine

let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();

store_from_url creates an object store from the URL. For cloud storage with custom credentials, use store_from_url_opts instead. See Configuring Storage for S3, Azure, and GCS options.

DefaultEngine::builder(store).build() constructs the default engine, which handles all I/O (Parquet, JSON, file listing) and expression evaluation using Arrow.

3. Get a snapshot

let snapshot = Snapshot::builder_for(url).build(&engine)?;

A Snapshot is an immutable view of the table at a specific version. Without calling .at_version(v) on the builder, this gives you the latest version.

The snapshot gives you access to the table’s schema and properties, and serves as the entry point for scanning and writing.

4. Build and execute a scan

let scan = snapshot.scan_builder().build()?;
for data in scan.execute(Arc::new(engine))? {
    let record_batch = data?.try_into_record_batch()?;
    batches.push(record_batch);
}

scan_builder() returns a ScanBuilder which you can configure with column selection (.with_schema()) or filter predicates (.with_predicate()). Here we use the defaults: all columns, no filter.

execute() returns an iterator of EngineData results. Since we’re using the default engine, each item is backed by an Arrow RecordBatch. The try_into_record_batch() method (from the EngineDataArrowExt extension trait) unwraps it.

Run it

If you have the delta-kernel-rs repo checked out locally, you can test with one of its test tables:

cargo run -- /path/to/delta-kernel-rs/kernel/tests/data/basic_partitioned/

Expected output:

Table version: 1
Schema:
struct:
├─letter: string (is nullable: true, metadata: {})
├─number: long (is nullable: true, metadata: {})
└─a_float: double (is nullable: true, metadata: {})

+--------+--------+---------+
| letter | number | a_float |
+--------+--------+---------+
|        | 6      | 6.6     |
| a      | 4      | 4.4     |
| e      | 5      | 5.5     |
| a      | 1      | 1.1     |
| b      | 2      | 2.2     |
| c      | 3      | 3.3     |
+--------+--------+---------+

What’s next

• Quick Start: Writing a Table covers creating a table and writing data.
• Building a Scan explores column selection, filter pushdown, and more.

Quick Start: Writing a Table

In this tutorial, you will create a new Delta table, write data to it, and read it back. It builds on the concepts from Quick Start: Reading a Table.

Setup

Create a new project and add dependencies:

cargo new delta_write_example
cd delta_write_example

Writing data requires tokio because the default engine’s Parquet writer is async:

cargo add delta_kernel -F default-engine-rustls -F arrow -F internal-api
cargo add tokio -F rt-multi-thread -F macros

Write the code

Replace src/main.rs with the following:

Filename: src/main.rs

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;

use delta_kernel::arrow::array::{Int32Array, RecordBatch, StringArray};
use delta_kernel::arrow::util::pretty::print_batches;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::arrow_conversion::TryIntoArrow;
use delta_kernel::engine::arrow_data::{ArrowEngineData, EngineDataArrowExt as _};
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::CommitResult;
use delta_kernel::{DeltaResult, Snapshot};

#[tokio::main]
async fn main() -> DeltaResult<()> {
    let table_path = std::env::args()
        .nth(1)
        .expect("usage: delta_write_example <TABLE_DIR>");
    let url = delta_kernel::try_parse_uri(&table_path)?;

    // Build the engine
    let engine = DefaultEngine::builder(store_from_url(&url)?).build();

    // 1. Create the table
    let schema = Arc::new(StructType::try_new(vec![
        StructField::not_null("id", DataType::INTEGER),
        StructField::nullable("name", DataType::STRING),
    ])?);

    create_table(url.as_str(), schema.clone(), "quick-start/1.0")
        .build(&engine, Box::new(FileSystemCommitter::new()))?
        .commit(&engine)?;
    println!("Created table at {url}");

    // 2. Write data
    let snapshot = Snapshot::builder_for(url.clone()).build(&engine)?;

    let mut txn = snapshot
        .transaction(Box::new(FileSystemCommitter::new()), &engine)?
        .with_operation("INSERT".to_string())
        .with_engine_info("quick-start/1.0")
        .with_data_change(true);

    // Build an Arrow RecordBatch
    let arrow_schema: delta_kernel::arrow::datatypes::Schema =
        schema.as_ref().try_into_arrow()?;
    let batch = RecordBatch::try_new(
        Arc::new(arrow_schema),
        vec![
            Arc::new(Int32Array::from(vec![1, 2, 3])),
            Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
        ],
    )?;

    // Write Parquet and add file metadata to the transaction
    let write_context = Arc::new(txn.unpartitioned_write_context()?);
    let data = ArrowEngineData::new(batch);
    let file_metadata = engine
        .write_parquet(&data, write_context.as_ref())
        .await?;
    txn.add_files(file_metadata);

    // Commit
    match txn.commit(&engine)? {
        CommitResult::CommittedTransaction(committed) => {
            println!("Committed version {}", committed.commit_version());
        }
        CommitResult::ConflictedTransaction(_) => {
            panic!("unexpected conflict on a brand new table");
        }
        CommitResult::RetryableTransaction(retry) => {
            panic!("commit failed with retryable error: {}", retry.error);
        }
    }

    // 3. Read it back
    let snapshot = Snapshot::builder_for(url).build(&engine)?;
    let scan = snapshot.scan_builder().build()?;
    let batches: Vec<RecordBatch> = scan
        .execute(Arc::new(engine))?
        .map(|data| -> DeltaResult<RecordBatch> {
            Ok(data?.try_into_record_batch()?)
        })
        .collect::<DeltaResult<Vec<_>>>()?;
    print_batches(&batches)?;

    Ok(())
}

Step by step

1. Create the table

let schema = Arc::new(StructType::try_new(vec![
    StructField::not_null("id", DataType::INTEGER),
    StructField::nullable("name", DataType::STRING),
])?);

create_table(url.as_str(), schema.clone(), "quick-start/1.0")
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;

create_table returns a builder. You provide:

• The table path
• A schema (using kernel’s StructType)
• An engine info string (identifies your application)

.build() takes the engine and a Committer. For local filesystem tables, use FileSystemCommitter. For catalog-managed tables, you provide your own committer. Catalog-Managed Tables covers that topic.

.commit() writes version 0 of the table (the initial Protocol and Metadata actions).

2. Write data

The write flow has four parts:

Start a transaction:

let mut txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_data_change(true);

Build your data as an Arrow RecordBatch and wrap it:

let arrow_schema: delta_kernel::arrow::datatypes::Schema =
    schema.as_ref().try_into_arrow()?;
let batch = RecordBatch::try_new(
    Arc::new(arrow_schema),
    vec![
        Arc::new(Int32Array::from(vec![1, 2, 3])),
        Arc::new(StringArray::from(vec!["Alice", "Bob", "Charlie"])),
    ],
)?;
let data = ArrowEngineData::new(batch);

Write the Parquet file and collect file metadata:

let write_context = Arc::new(txn.unpartitioned_write_context()?);
let file_metadata = engine
    .write_parquet(&data, write_context.as_ref())
    .await?;
txn.add_files(file_metadata);

unpartitioned_write_context() creates a WriteContext with the target directory, schema, and stats configuration. write_parquet writes a Parquet file and returns metadata (path, size, stats) that the transaction needs. add_files registers that metadata with the transaction.

Commit:

match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => { /* success */ }
    CommitResult::ConflictedTransaction(_) => { /* another writer won */ }
    CommitResult::RetryableTransaction(retry) => { /* transient error, retry */ }
}

commit() returns a CommitResult with three variants. For blind appends to a table with no concurrent writers, you’ll always get CommittedTransaction.

Run it

mkdir -p /tmp/my_delta_table
cargo run -- /tmp/my_delta_table

Expected output:

Created table at file:///tmp/my_delta_table
Committed version 1
+----+---------+
| id | name    |
+----+---------+
| 1  | Alice   |
| 2  | Bob     |
| 3  | Charlie |
+----+---------+

What’s next

• Creating a Table covers table properties, partition columns, and more.
• Appending Data covers writing to existing tables and retry logic.

Architecture overview

Delta Kernel is a Rust library that encapsulates the Delta Lake protocol so you can build connectors without understanding protocol internals. This matters because the Delta protocol is complex and evolving. Kernel absorbs that complexity and exposes a small, stable API surface for reading and writing Delta tables.

Terminology

Before diving in, here are the key terms used throughout this guide:

• Compute engine is a data processing framework like Apache Spark, Apache Flink, DuckDB, Polars, or DataFusion. Each defines its own DataSource API for reading and writing tables (e.g., Spark’s DataSourceV2, Flink’s Source/Sink APIs).

• Connector is the integration layer between a compute engine and Delta tables. It implements the compute engine’s DataSource API (Table, Scan, Writer, etc.) and uses Delta Kernel to fulfill those interfaces. For example, the Delta Spark connector implements Spark’s Table, ScanBuilder, Scan, and PartitionReader interfaces, delegating the actual Delta protocol work to Kernel.

• Delta Kernel is the library (this project). It implements the Delta protocol and exposes APIs (Snapshot, Scan, Transaction) that connectors use. It never does I/O directly.

• Engine trait is the I/O and compute abstraction that Kernel calls into. It has four required handlers (StorageHandler, ParquetHandler, JsonHandler, EvaluationHandler) and one optional handler (MetricsReporter). A DefaultEngine is provided. Connectors can implement their own for better performance with their native data formats and I/O.

Layered design

Delta Kernel is organized into layers. Each layer has a clear responsibility and a well-defined interface to the layer above and below it.

 ┌────────────────────────────────────────────┐
 │           Compute Engine                   │
 │   (Spark, Flink, DuckDB, Polars, ...)      │
 └──────────────────┬─────────────────────────┘
                    │  calls DataSource API
 ┌──────────────────▼──────────────────────────┐
 │         Your Delta Connector                │
 │                                             │
 │  Implements the compute engine's DataSource │
 │  API and uses Kernel to fulfill it          │
 └──────────────────┬──────────────────────────┘
                    │  calls Kernel APIs
 ┌──────────────────▼──────────────────────────┐
 │            Delta Kernel (core logic)        │
 │                                             │
 │  Snapshot · Scan · Transaction · Log Replay │
 │  Data Skipping · Predicate Pushdown         │
 │  Protocol Compliance · Table Features       │
 └──────────────────┬──────────────────────────┘
                    │  calls into
 ┌──────────────────▼──────────────────────────┐
 │           Engine trait (abstraction)        │
 │                                             │
 │  EvaluationHandler · StorageHandler         │
 │  JsonHandler · ParquetHandler               │
 │  MetricsReporter (optional)                 │
 └──────────────────┬─────────────────────────┘
                    │  implemented by
 ┌──────────────────▼──────────────────────────┐
 │     DefaultEngine  (or your custom engine)  │
 │                                             │
 │  Arrow-based evaluation · object_store I/O  │
 └──────────────────┬──────────────────────────┘
                    │
 ┌──────────────────▼──────────────────────────┐
 │              Storage                        │
 │   (Local FS, S3, GCS, Azure, HDFS, ...)     │
 └─────────────────────────────────────────────┘

Kernel contains all Delta protocol logic: log replay, data skipping, schema enforcement, table features, and transaction coordination. It never does I/O directly.

Engine is a trait that the kernel calls whenever it needs I/O or expression evaluation. You can use the built-in DefaultEngine (Arrow + object_store) or implement your own. See The Engine Trait.

Your connector implements your compute engine’s DataSource API and calls kernel APIs (Snapshot, Scan, Transaction) to do the Delta work. The kernel handles the protocol; your connector controls execution, distribution, and data flow. See Building a Connector for details.

Key types

Snapshot

A Snapshot is an immutable view of a Delta table at a specific version. It is the entry point for everything: reading, writing, and inspecting table metadata.

let snapshot = Snapshot::builder_for("/path/to/table")
    .build(&engine)?;                // returns Arc<Snapshot>

let snapshot_v5 = Snapshot::builder_for("/path/to/table")
    .at_version(5)
    .build(&engine)?;

println!("Version: {}", snapshot.version());
println!("Schema: {:?}", snapshot.schema());

From a snapshot you can:

• Read the table schema and properties
• Build a Scan to read data
• Start a Transaction to write data
• Create a checkpoint

Scan

A Scan reads data from a table. It is built from a snapshot via ScanBuilder:

let scan = snapshot
    .scan_builder()
    .with_schema(my_schema)           // column selection (optional)
    .with_predicate(my_predicate)     // filter pushdown (optional)
    .build()?;

There are two ways to execute a scan:

Simple path. execute() does everything for you:

for data in scan.execute(engine)? {
    let batch = data?.try_into_record_batch()?;
    // process batch
}

Advanced path. scan_metadata() gives you control over parallelism:

for metadata in scan.scan_metadata(engine)? {
    let metadata = metadata?;
    // Each ScanMetadata contains the files to read
    // and per-file transforms to apply.
    // You can distribute these across threads or workers.
}

The advanced path is how you build a distributed connector. See Building a Connector for details.

Transaction

A Transaction writes data to a table. It is built from a snapshot:

let mut txn = snapshot                              // Arc<Snapshot>
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_data_change(true);

// Write Parquet files, then register their metadata.
txn.add_files(file_metadata);

// Commit atomically
match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => println!("v{}", c.commit_version()),
    CommitResult::ConflictedTransaction(_) => { /* handle conflict */ }
    CommitResult::RetryableTransaction(_) => { /* retry */ }
}

The file_metadata argument is an EngineData batch that matches txn.add_files_schema(), not raw file paths. See Appending data for the full pattern, including how to build that batch from Parquet write results.

Note

with_operation applies to update transactions (append, delete, etc.). For create-table transactions the operation is fixed to "CREATE TABLE" and cannot be overridden.

For schema evolution, start from the snapshot’s alter_table() builder instead of transaction(). See Altering a Table.

See Quick Start: Writing a Table for a complete example.

How a read works

When you call scan.execute(engine), here is what happens internally:

1. LOG REPLAY (kernel)
   Read delta log commits via engine.json_handler() and checkpoint
   parquet files (including parquet sidecars) via engine.parquet_handler().
   Determine the set of active files for this table version.

2. DATA SKIPPING (kernel)
   If a predicate was provided, evaluate file-level statistics
   (min/max values, null counts) to skip files that cannot match.
   This happens via engine.evaluation_handler().

3. FILE READING (engine)
   For each remaining file, read the Parquet data via
   engine.parquet_handler().read_parquet_files().

4. TRANSFORM (kernel + engine)
   Apply per-file transformations: partition value injection,
   column mapping, deletion vectors. Produces the final logical
   data that your connector consumes.

The kernel handles steps 1, 2, and 4. The engine handles step 3. This separation means the kernel never touches raw bytes. It works purely with metadata and delegates all I/O.

How a write works

1. START TRANSACTION (kernel)
   Create a Transaction from a snapshot. The snapshot pins the
   table version you're writing against.

2. WRITE DATA (engine / your code)
   Write Parquet files using the engine. Collect file metadata
   (path, size, statistics) and register it with the transaction
   via add_files().

3. COMMIT (kernel + committer)
   The kernel assembles the commit actions (CommitInfo, Add files,
   etc.) and hands them to the Committer. For filesystem tables,
   the Committer writes a JSON delta file atomically. For
   catalog-managed tables, it goes through the catalog.

4. HANDLE RESULT
   CommittedTransaction: success.
   ConflictedTransaction: another writer committed first.
   RetryableTransaction: transient I/O error, safe to retry.

EngineData: staying engine-agnostic

The kernel never assumes your data is Arrow. Instead, it uses the EngineData trait, an opaque interface that any engine can implement. The kernel accesses data through visitor callbacks, not by inspecting columns directly.

 Kernel                          Engine
 ──────                          ──────
 "I need columns [path, size]"
       ──────────────────>
                                 "Here are GetData accessors
                                  for those columns"
       <──────────────────
 Visits rows via GetData

The DefaultEngine implements EngineData with ArrowEngineData (wrapping Arrow RecordBatch). If you use the default engine, you can convert back to RecordBatch with the EngineDataArrowExt trait. If you build a custom engine, you implement EngineData for your own columnar format.

See The Engine Trait for more on how this works.

Crate structure

The project is organized into several crates:

What’s next

• The Engine Trait explains the Engine abstraction and its handlers in detail.
• Building a Scan walks through reading data from a Delta table.
• Building a Connector describes how to integrate Kernel with a compute engine.

See also

• Schema and Types for Kernel’s type system.
• Appending Data for a complete write example.

The Engine trait

The Engine trait is the integration point between Delta Kernel and your connector. Kernel implements the Delta protocol and gives you scan and write APIs, but it needs help with the mechanics: listing files, reading and writing JSON and Parquet, and evaluating expressions for data skipping and logical-to-physical transformations. Kernel never does any of this directly. Instead, it calls into the Engine trait, which your connector implements.

This matters because it lets Kernel stay format-agnostic and runtime-agnostic. You control how I/O happens, what columnar format you use, and how expressions are evaluated.

A DefaultEngine is provided that you can use out of the box. If you need better performance or want to use your own data formats, you can build a custom engine. See Building a Connector for details.

The trait

trait Engine: AsAny {
    fn evaluation_handler(&self) -> Arc<dyn EvaluationHandler>;
    fn storage_handler(&self) -> Arc<dyn StorageHandler>;
    fn json_handler(&self) -> Arc<dyn JsonHandler>;
    fn parquet_handler(&self) -> Arc<dyn ParquetHandler>;
}

Kernel calls these methods whenever it needs to interact with the outside world. Each returns a handler trait object that Kernel uses for a specific category of work. The four handlers cover storage, JSON, Parquet, and expression evaluation. For observability, see Observability.

The four handlers

StorageHandler

File system operations. Kernel calls this to list and read files from the Delta log, and to write commit and checkpoint files.

JsonHandler

Reads and writes JSON. Kernel uses this for Delta log commits (the _delta_log/*.json files) and the JSON checkpoint manifest that references parquet sidecars. The sidecar files themselves are parquet and are read by the ParquetHandler below.

ParquetHandler

Reads and writes Parquet. Kernel uses this for checkpoint files (including parquet checkpoint sidecars) and for reading data files during scans.

The Parquet handler also supports metadata columns (row index, file path) and field-ID-based column matching for column mapping.

EvaluationHandler

Expression evaluation. Kernel uses this for data skipping (evaluating predicates against file statistics) and for applying per-file transformations (partition value injection, column mapping).

The expression and predicate evaluators are reusable objects that you can call repeatedly on different batches of EngineData.

The Default Engine

The DefaultEngine is a batteries-included implementation that works out of the box:

• Uses Apache Arrow as the in-memory data format
• Uses object_store for I/O (supports local FS, S3, GCS, Azure)
• Runs async I/O on a Tokio thread pool
• Supports multiple Arrow versions (see Feature Flags)

To construct one, create an object store and pass it to the builder:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = url::Url::parse("file:///path/to/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
Ok(())
}
}

DefaultEngine also provides a convenience method for writing Parquet files:

let engine_data = engine
    .write_parquet(&data, &write_context)
    .await?;

This is not part of the Engine trait itself. It’s a helper on DefaultEngine that orchestrates the lower-level ParquetHandler methods with a WriteContext (write directory, schema, stats columns, partition values).

Configuring the Default Engine

DefaultEngine uses a builder pattern that lets you customize the task executor and plug in a metrics reporter. The builder starts from DefaultEngine::builder(store) and chains optional configuration before calling build():

let engine = DefaultEngine::builder(store)
    .with_metrics_reporter(reporter)
    .with_task_executor(executor)
    .build();

All builder methods are optional. Calling DefaultEngine::builder(store).build() gives you a fully functional engine with sensible defaults.

The TaskExecutor trait

DefaultEngine uses asynchronous I/O internally, but Kernel’s public APIs are synchronous. The TaskExecutor trait bridges this gap by defining how async work gets scheduled and awaited. It has four methods:

You don’t need to implement TaskExecutor yourself unless you have a non-Tokio async runtime. Kernel ships two Tokio-based implementations behind the tokio feature flag.

TokioBackgroundExecutor

TokioBackgroundExecutor is the default executor. It spawns a dedicated background thread running a single-threaded Tokio runtime. All async work is dispatched to that thread over a channel.

use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;

let executor = TokioBackgroundExecutor::new();

This is the right choice when:

• You don’t already have a Tokio runtime in your application.
• You want Kernel’s I/O isolated from the rest of your process.
• You’re building a standalone connector or CLI tool.

Because it owns its runtime, TokioBackgroundExecutor works even when no external Tokio runtime is active. On drop, it shuts down the background thread cleanly.

TokioMultiThreadExecutor

TokioMultiThreadExecutor runs async work on a multi-threaded Tokio runtime. It comes in two flavors.

Share an existing runtime. If your application already has a Tokio runtime (for example, a web server or a query engine), pass its handle so Kernel’s I/O shares the same thread pool:

use delta_kernel::engine::default::executor::tokio::TokioMultiThreadExecutor;

let handle = tokio::runtime::Handle::current();
let executor = TokioMultiThreadExecutor::new(handle);

The handle must come from a multi-threaded runtime. Passing a current-thread handle causes a panic.

Own a dedicated runtime. If you want a multi-threaded runtime that Kernel manages, use new_owned_runtime. You can optionally set the number of worker threads and the maximum number of blocking threads. Pass None for either to use Tokio’s defaults:

let executor = TokioMultiThreadExecutor::new_owned_runtime(
    Some(4),   // 4 worker threads
    Some(64),  // up to 64 blocking threads
)?;

Warning

Deeply nested block_on calls can exhaust Tokio’s blocking thread pool and deadlock. If you set a small max_blocking_threads, keep nesting depth low.

Choosing an executor

To use a custom executor, pass it to the builder with with_task_executor:

let executor = Arc::new(TokioMultiThreadExecutor::new(handle));
let engine = DefaultEngine::builder(store)
    .with_task_executor(executor)
    .build();

Entering the runtime context

Some libraries (for example, object_store or reqwest) require an active Tokio runtime context to construct clients or resolve configuration. If you call such code outside of an async function, tokio::runtime::Handle::current() will panic because no runtime is active.

DefaultEngine::enter() solves this by entering the executor’s runtime context:

let guard = engine.enter();
// Code here can call Handle::current() safely.
// The guard must be dropped before acquiring another.
drop(guard);

The returned guard keeps the runtime context active until it is dropped. If you acquire multiple guards, you must drop them in reverse order. Dropping out of order causes a panic.

When to implement your own Engine

You should implement Engine if:

• You have your own columnar data format (not Arrow)
• You need custom I/O (e.g. your own distributed file system client)
• You want to use your own expression evaluation engine
• You need to control parallelism or resource usage beyond what DefaultEngine offers

You do not need a custom engine to use different storage (S3, Azure, etc.). The DefaultEngine supports all object_store backends. See Configuring Storage.

For a guide on implementing Engine, see Implementing the Engine Trait.

What’s next

• Building a Connector explains the role of a connector and how the Engine fits into the bigger picture.
• Implementing the Engine Trait walks through building a custom Engine from scratch.
• Configuring Storage shows how to point DefaultEngine at S3, GCS, or Azure storage.

Schemas and Data Types

Kernel defines its own type system that mirrors the Delta protocol specification. This type system is independent of any engine’s type system (such as Arrow). Your engine converts to and from Kernel types as needed.

Data types

The DataType enum represents all types supported by the Delta protocol:

Primitive types

Decimal

Decimals have a precision (1 to 38 inclusive) and a scale (0 to precision inclusive):

extern crate delta_kernel;
use delta_kernel::DeltaResult;
use delta_kernel::schema::DataType;
fn main() -> DeltaResult<()> {
let price_type = DataType::decimal(18, 2)?;
Ok(())
}

Complex types

Array

An ordered sequence of elements, all of the same type:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::schema::{ArrayType, DataType};
// Array of nullable strings
let array_type = DataType::from(ArrayType::new(DataType::STRING, true));
}

The contains_null parameter indicates whether elements can be null.

Map

A collection of key-value pairs:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::schema::{DataType, MapType};
// Map from string keys to nullable integer values
let map_type = DataType::from(MapType::new(DataType::STRING, DataType::INTEGER, true));
}

Map keys are never null. The value_contains_null parameter controls whether values can be null.

Struct

A named collection of fields (see Schemas below). Structs can be nested:

extern crate delta_kernel;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::DeltaResult;
fn main() -> DeltaResult<()> {
let address_type = StructType::try_new([
    StructField::nullable("street", DataType::STRING),
    StructField::nullable("city", DataType::STRING),
    StructField::nullable("zip", DataType::STRING),
])?;

let person_type = StructType::try_new([
    StructField::not_null("name", DataType::STRING),
    StructField::nullable("address", address_type),
])?;
Ok(())
}

Variant

A semi-structured type that can hold any value. The physical representation uses a struct with metadata and value fields (both binary). To create an unshredded variant column:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::schema::DataType;
let variant_type = DataType::unshredded_variant();
}

Schemas

A schema is a StructType, an ordered collection of named, typed fields. The type aliases Schema and SchemaRef (Arc<StructType>) are used throughout the API.

Creating a schema

extern crate delta_kernel;
use delta_kernel::DeltaResult;
use delta_kernel::schema::{DataType, StructField, StructType};
fn main() -> DeltaResult<()> {
let schema = StructType::try_new([
    StructField::not_null("id", DataType::LONG),
    StructField::nullable("name", DataType::STRING),
    StructField::nullable("score", DataType::DOUBLE),
])?;
Ok(())
}

try_new returns an error if the schema contains duplicate field names (case-insensitive, since Delta column names are case-insensitive).

StructType::builder() provides a builder for incremental construction:

extern crate delta_kernel;
use delta_kernel::DeltaResult;
use delta_kernel::schema::{DataType, StructField, StructType};
fn main() -> DeltaResult<()> {
let schema = StructType::builder()
    .add_field(StructField::not_null("id", DataType::LONG))
    .add_field(StructField::nullable("name", DataType::STRING))
    .build()?;
Ok(())
}

Querying a schema

// Look up a field by name
if let Some(field) = schema.field("name") {
    println!("{}: {:?}, nullable={}", field.name(), field.data_type(), field.is_nullable());
}

// Check if a field exists
assert!(schema.contains("id"));

// Get the positional index of a field
let idx = schema.index_of("name"); // Some(1)

// Iterate over all fields
for field in schema.fields() {
    println!("{}", field.name());
}

// Number of fields
let n = schema.num_fields();

Projecting a schema

project() creates a new schema with a subset of fields. The output preserves the order you specify:

// Table schema: [id, name, email, created_at]
// Select only [email, id] in that order
let projected = schema.project(&["email", "id"])?;

See Column Selection for how this is used in scans.

Fields

A StructField has a name, data type, nullability flag, and optional metadata:

// Non-nullable field
let id = StructField::not_null("id", DataType::LONG);

// Nullable field
let name = StructField::nullable("name", DataType::STRING);

// Field with explicit nullability
let score = StructField::new("score", DataType::DOUBLE, true);

Field metadata

Fields can carry arbitrary key-value metadata:

let field = StructField::nullable("price", DataType::decimal(18, 2)?)
    .with_metadata([("description", "Unit price in USD")]);

Metadata is stored as a HashMap<String, MetadataValue>. The MetadataValue enum supports strings, numbers (i64), booleans, and arbitrary JSON.

Reading a table’s schema

Every Snapshot exposes the table’s schema:

extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let schema = snapshot.schema();

for field in schema.fields() {
    println!("{}: {:?}", field.name(), field.data_type());
}
Ok(())
}

What’s next

• Column Selection: projecting columns in a scan
• Creating a Table: defining a schema for a new table

Feature flags

Delta Kernel uses Cargo feature flags to keep the core library lightweight. The core crate has no required runtime dependencies beyond the Rust standard library. Everything else is opt-in.

Recommended starting point

For most connectors that use the built-in engine with Arrow:

[dependencies]
delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow"] }

Complete feature reference

Default engine

These features enable the built-in DefaultEngine, which provides out-of-the-box support for reading and writing Delta tables.

Pick exactly one. Both pull in default-engine-base plus reqwest (for fetching pre-signed URLs), which together enable:

• arrow-conversion and arrow-expression
• tokio async runtime
• futures
• reqwest HTTP client (TLS backend selected by the feature you choose)

Arrow

arrow-conversion and arrow-expression are pulled in automatically by the default engine. You only need to specify them directly if you’re building a custom engine that still uses Arrow.

Tip

Each arrow-* version feature also pulls in the matching parquet and object_store crate versions. If your connector already depends on a specific Arrow version, pin the matching feature to avoid duplicate transitive dependencies.

Experimental features

These features are under active development. Their APIs may change between releases.

Development features

Common combinations

Read and write with the default engine:

delta_kernel = { version = "0.21", features = ["default-engine-rustls", "arrow"] }

Custom engine using Arrow (no default engine):

delta_kernel = { version = "0.21", features = ["arrow-conversion", "arrow-expression"] }

Minimal custom engine with no Arrow dependency at all:

delta_kernel = { version = "0.21" }

This gives you only the core Kernel types and traits. You implement Engine and EngineData entirely in your own data format.

Building a scan

To read data from a Delta table, you build a Scan from a Snapshot, optionally configure column selection and filter predicates, and then execute it.

The basic pattern

Every scan follows the same pattern:

1. Get a Snapshot of the table
2. Call snapshot.scan_builder() to get a ScanBuilder
3. Configure the builder (optional)
4. Call .build() to create the Scan
5. Execute the scan

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;

let scan = snapshot
    .scan_builder()
    .build()?;
Ok(())
}
}

Without any configuration, this scans all columns with no filter. It’s equivalent to SELECT * FROM table.

Configuring a scan

ScanBuilder supports two main configuration options:

Column selection with with_schema

Pass a schema containing only the columns you want to read:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let read_schema = Arc::new(StructType::try_new([
    StructField::nullable("name", DataType::STRING),
    StructField::nullable("age", DataType::INTEGER),
])?);

let scan = snapshot
    .scan_builder()
    .with_schema(read_schema)
    .build()?;
Ok(())
}
}

The schema you provide must be a subset of the table’s schema. Kernel only reads the columns you specify from each Parquet file.

For more details, see Column Selection.

Filter pushdown with with_predicate

Pass a predicate expression to skip files that cannot contain matching rows:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let predicate = Arc::new(
    Predicate::lt(column_expr!("age"), Scalar::from(30))
);

let scan = snapshot
    .scan_builder()
    .with_predicate(predicate)
    .build()?;
Ok(())
}
}

Kernel uses the predicate to evaluate file-level statistics (min/max values) and skip entire files that cannot match. This is called data skipping and can significantly reduce the amount of data read.

Note

Filtering is best-effort. The scan may still include rows that don’t match the predicate. Your connector should apply the filter to the returned data for exact results.

For more details, see Filter Pushdown and File Skipping.

Executing a scan

There are two ways to execute a scan: the simple path for single-process use, and the advanced path for when you need control over parallelism.

Simple path: execute()

execute() handles log replay, data skipping, file reading, and physical-to-logical transformations. It returns an iterator of EngineData results.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::arrow_data::EngineDataArrowExt as _;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let scan = snapshot.scan_builder().build()?;

let mut batches = vec![];
for data in scan.execute(Arc::new(engine))? {
    let record_batch: delta_kernel::arrow::record_batch::RecordBatch =
        data?.try_into_record_batch()?;
    batches.push(record_batch);
}
Ok(())
}
}

execute() takes an Arc<dyn Engine> (not a reference) because it needs to keep the engine alive for the lifetime of the returned iterator.

If you’re using the default engine, each EngineData is backed by an Arrow RecordBatch. The try_into_record_batch() method (from the EngineDataArrowExt trait) unwraps it.

This is the right choice when you’re running in a single process and don’t need to control how files are distributed across threads.

What’s next

• Column Selection: projecting specific columns
• Advanced Reads with scan_metadata(): distributed and multi-threaded reads
• Filter Pushdown and File Skipping: predicate-based optimization

Column Selection

By default a scan reads all columns from a table. You can select a subset of columns (projection pushdown) so the engine only reads the data you need.

Projecting columns

Use Schema::project() to create a schema containing only the columns you want, then pass it to ScanBuilder::with_schema():

extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
// Table has columns [id, name, email, created_at]
// Select only id and name
let projected_schema = snapshot.schema().project(&["id", "name"])?;

let scan = snapshot
    .scan_builder()
    .with_schema(projected_schema)
    .build()?;
Ok(())
}

The returned data will contain only the projected columns, in the order you specified. Requesting a column that does not exist in the table schema returns an error.

Reordering columns

project() returns columns in the order you provide, which can differ from the table schema order:

// Table schema is [id, name, email]
// Return [email, id]
let reordered = snapshot.schema().project(&["email", "id"])?;

Metadata columns

You can request metadata columns that are not part of the table data but provide information about each row’s origin. Add them to your scan schema with Schema::add_metadata_column():

extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::MetadataColumnSpec;
use delta_kernel::{DeltaResult, Snapshot};
fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
// Start with a projection
let schema = snapshot.schema().project_as_struct(&["id", "name"])?;

// Add a row index metadata column
let schema = schema.add_metadata_column("row_idx", MetadataColumnSpec::RowIndex)?;

let scan = snapshot
    .scan_builder()
    .with_schema(Arc::new(schema))
    .build()?;
Ok(())
}

The available metadata columns:

You choose the column name when calling add_metadata_column(). Only one metadata column of each type is allowed per scan.

What’s next

• Building a Scan covers the full scan API.
• Filter Pushdown and File Skipping covers predicate-based optimization.

Filter pushdown and file skipping

To reduce the amount of data your connector reads from storage, you can provide a predicate to a scan. Kernel uses the predicate for data skipping, evaluating file-level statistics to skip entire Parquet files that cannot contain matching rows.

Before reading this page, make sure you understand Building a Scan.

Building predicates

A predicate is a boolean expression that describes which rows you want. Kernel evaluates predicates against file-level statistics to skip files before reading them.

Predicates are built from the Predicate type and Expression values. The simplest way is to use the column_expr! macro for column references and Scalar for literal values.

Comparison operators

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};

// age < 30
let pred = Predicate::lt(column_expr!("age"), Scalar::from(30));

// price >= 100.0
let pred = Predicate::ge(column_expr!("price"), Scalar::from(100.0_f64));

// name == "Alice"
let pred = Predicate::eq(column_expr!("name"), Scalar::from("Alice"));

// status != "deleted"
let pred = Predicate::ne(column_expr!("status"), Scalar::from("deleted"));
}

The full set of comparison constructors:

distinct is a NULL-safe inequality: it returns true when a and b differ, even when one or both are NULL. The other comparisons follow SQL NULL semantics and produce NULL when either input is NULL.

Each constructor takes impl Into<Expression> for both arguments, so you can pass column_expr!() results, Scalar values, or any Expression directly.

Combining predicates

Use and, or, and not to build compound predicates:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};

// age >= 18 AND age < 65
let pred = Predicate::and(
    Predicate::ge(column_expr!("age"), Scalar::from(18)),
    Predicate::lt(column_expr!("age"), Scalar::from(65)),
);

// status == "active" OR status == "pending"
let pred = Predicate::or(
    Predicate::eq(column_expr!("status"), Scalar::from("active")),
    Predicate::eq(column_expr!("status"), Scalar::from("pending")),
);

// NOT (archived)
let pred = Predicate::not(
    Predicate::eq(column_expr!("archived"), Scalar::from(true)),
);
}

For combining more than two predicates, use and_from or or_from:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};

// age >= 18 AND country == "US" AND active == true
let pred = Predicate::and_from([
    Predicate::ge(column_expr!("age"), Scalar::from(18)),
    Predicate::eq(column_expr!("country"), Scalar::from("US")),
    Predicate::eq(column_expr!("active"), Scalar::from(true)),
]);
}

NULL checks

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Predicate};

// email IS NULL
let pred = Predicate::is_null(column_expr!("email"));

// email IS NOT NULL
let pred = Predicate::is_not_null(column_expr!("email"));
}

Nested columns

The column_expr! macro supports dot-separated paths for nested struct fields:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};

// address.city == "Seattle"
let pred = Predicate::eq(
    column_expr!("address.city"),
    Scalar::from("Seattle"),
);
}

Method syntax

You can also build predicates using method syntax on Expression:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::expressions::{column_expr, Scalar};

// age < 30
let pred = column_expr!("age").lt(Scalar::from(30));

// name == "Alice"
let pred = column_expr!("name").eq(Scalar::from("Alice"));

// email IS NOT NULL
let pred = column_expr!("email").is_not_null();
}

Applying a predicate to a scan

Pass the predicate to ScanBuilder::with_predicate:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::expressions::{column_expr, Predicate, Scalar};
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let predicate = Arc::new(
    Predicate::and(
        Predicate::ge(column_expr!("age"), Scalar::from(18)),
        Predicate::lt(column_expr!("age"), Scalar::from(65)),
    )
);

let scan = snapshot
    .scan_builder()
    .with_predicate(predicate)
    .build()?;
Ok(())
}
}

with_predicate takes impl Into<Option<PredicateRef>>, so you can pass an Arc<Predicate> directly.

How data skipping works

When a scan has a predicate, Kernel applies it in two stages to eliminate files before your connector reads them.

File skipping using statistics. Each Parquet file in a Delta table has associated statistics: minimum and maximum values per column, null counts, and row counts. Kernel rewrites your predicate into a data skipping predicate that evaluates against these statistics. For example, given the predicate age < 30, if a file’s minimum value for age is 35, Kernel knows no rows in that file can match and skips it entirely. If a file’s minimum is 10 and maximum is 50, the file might contain matching rows, so Kernel keeps it.

Partition pruning. For partitioned tables, partition column values are stored in the Delta log metadata rather than in the Parquet files. Kernel evaluates predicates on partition columns directly against these metadata values, which is even cheaper than statistics-based skipping because no file I/O is required.

Filtering is best-effort

Warning

Data skipping is an optimization, not a guarantee. The scan may return rows that do not match your predicate. Your connector must apply row-level filtering after reading the data if exact results are required.

This happens for several reasons:

• Statistics are at the file level, not the row level. A file whose min/max range overlaps the predicate may still contain non-matching rows.
• Not all columns have statistics. Delta tables have a configurable limit on how many columns collect statistics (default: 32).
• Kernel may not fully evaluate complex predicates. It skips what it can and passes through the rest.

Controlling statistics

By default, Kernel reads file-level statistics from the transaction log and uses them internally for data skipping, but does not expose those statistics to your connector. ScanBuilder provides three methods that change this behavior.

Disabling data skipping entirely

If your compute engine performs its own data skipping, you can tell Kernel to skip reading statistics altogether. This avoids the cost of parsing statistics from checkpoint files.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let scan = snapshot
    .scan_builder()
    .with_skip_stats(true)
    .build()?;
Ok(())
}
}

When with_skip_stats(true) is set:

• Kernel skips the stats column entirely during checkpoint reads.
• No statistics-based or partition-value-based file pruning occurs (row-level partition filtering still applies).
• The stats field on each ScanFile is None.

Use this when your connector or compute engine already handles file pruning and you want to avoid the overhead of parsing statistics you won’t use.

Including all statistics in scan metadata

To receive pre-parsed statistics (min/max values, null counts, row counts) for every file in your scan metadata, call include_all_stats_columns():

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let scan = snapshot
    .scan_builder()
    .include_all_stats_columns()
    .build()?;
Ok(())
}
}

The statistics appear in a stats_parsed column in the scan metadata. Which columns have statistics depends on the table’s configuration (delta.dataSkippingStatsColumns or delta.dataSkippingNumIndexedCols).

You can combine this with with_predicate. When both are set, Kernel performs its own data skipping internally and exposes the parsed statistics so your connector can apply additional pruning logic.

Including statistics for specific columns

To receive statistics for only a subset of columns, call with_stats_columns:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::expressions::ColumnName;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let scan = snapshot
    .scan_builder()
    .with_stats_columns(vec![
        ColumnName::new(["age"]),
        ColumnName::new(["city"]),
    ])
    .build()?;
Ok(())
}
}

Only the named columns appear in stats_parsed. Passing an empty list is equivalent to the default behavior (no stats output, but internal data skipping still works if a predicate is set).

Choosing the right mode

Note

These methods follow “last call wins” semantics. If you call include_all_stats_columns() and then with_skip_stats(true), stats are skipped entirely.

What’s next

• Column Selection covers projecting specific columns to further reduce the data you read.
• Scan Metadata explains how to access per-file scan information, including partition values and deletion vectors.

Advanced reads with scan_metadata()

To control when and how data files are read, you can use scan.scan_metadata() instead of scan.execute(). This method gives you access to the file list and metadata without reading the actual Parquet files, so you can decide where and how to perform the I/O. This API provides a path for building connectors for distributed compute engines (Spark, Flink, etc.) where data reads may happen in parallel.

Before reading this page, make sure you understand Building a Scan.

Note

scan_metadata() parallelizes data reads but not log replay. To parallelize log replay as well, see Distributed Log Replay.

Example pattern

The execution flow below illustrates how one might implement parallel data reads using scan_metadata():

sequenceDiagram
    participant D as Driver
    participant W as Workers

    D->>D: 1. Create Snapshot + Scan
    D->>D: 2. Call scan_metadata()
    D->>D: 3. Extract ScanFiles
    D->>W: 4. Distribute files to workers
    W->>W: 5. Read Parquet files
    W->>W: 6. Transform physical → logical
    W->>W: 7. Apply deletion vectors
    W->>D: 8. Return logical data

For a complete working example of multi-threaded reads, see the read-table-multi-threaded example in the kernel repository.

Enumerating ScanFiles

scan_metadata() returns an iterator of ScanMetadata. Each ScanMetadata represents a batch of files that need to be read for the scan. You can then use visit_scan_files on each ScanMetadata to extract each file (represented as a ScanFile) in the batch.

For example:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::scan::state::ScanFile;
use delta_kernel::{DeltaResult, Engine};
fn perform_read(_chunk: &[ScanFile]) {}
fn example(scan: &delta_kernel::scan::Scan, engine: &dyn Engine) -> DeltaResult<()> {
fn collect_files(files: &mut Vec<ScanFile>, file: ScanFile) {
    files.push(file);
}

let mut all_files: Vec<ScanFile> = vec![];
for metadata in scan.scan_metadata(engine)? {
    let metadata = metadata?;
    all_files = metadata.visit_scan_files(all_files, collect_files)?;
}

// Now distribute these files across threads, tasks, or remote workers
let num_workers = 4;
let chunk_size = (all_files.len() / num_workers).max(1);
for chunk in all_files.chunks(chunk_size) {
    perform_read(chunk);
}
Ok(())
}
}

It is preferred to use visit_scan_files to iterate through each ScanFile. However if you need raw access to the set of files as an EngineData, each ScanMetadata contains:

• scan_files: a FilteredEngineData with one row per file to read, plus a selection vector indicating which rows are active (rows excluded by data skipping are marked inactive, so that files which can’t match the scan’s predicate will be filtered out).
• scan_file_transforms: a Vec<Option<ExpressionRef>> with a per-file transformation expression. If present, this expression must be applied to the physical data to produce the correct logical output (e.g., injecting partition column values, applying column mapping).

ScanFile

A ScanFile contains everything needed to read one data file:

pub struct ScanFile {
    pub path: String,                              // relative to table root
    pub size: i64,                                 // file size in bytes
    pub modification_time: i64,                    // millis since epoch
    pub stats: Option<Stats>,                      // parsed statistics
    pub dv_info: DvInfo,                           // deletion vector info
    pub transform: Option<ExpressionRef>,          // physical -> logical transform
    pub partition_values: HashMap<String, String>,  // partition column values
}

In a distributed engine, for example, you could serialize the ScanFile data along with the scan’s physical_schema() and logical_schema(), then ship them to workers. Or in a multi-threaded engine, you can send them through in-memory channels.

Reading and transforming data

Once a worker has a ScanFile, it should read the Parquet file and apply transformations to produce logical data.

Reading the Parquet file

Resolve the file path against the table root and read with the physical schema:

let file_url = scan.table_root().join(&scan_file.path)?;
let size: u64 = scan_file.size.try_into().map_err(|_| Error::generic("negative file size"))?;
let file_meta = FileMeta::new(file_url, scan_file.modification_time, size);

let read_results = engine
    .parquet_handler()
    .read_parquet_files(&[file_meta], physical_schema.clone(), None)?;

Transforming physical data to logical

Each file may need a per-file transform to convert the physical Parquet data into the logical schema that the scan requested. The transform handles:

• Partition value injection: partition columns are not stored in the Parquet file. The transform adds them from the file’s metadata.
• Column mapping: if the table uses column mapping, the transform renames physical columns to their logical names.
• Schema evolution: if the file was written with an older schema, the transform fills in missing columns with nulls.

Use transform_to_logical:

use delta_kernel::scan::state::transform_to_logical;

let physical_schema = scan.physical_schema();
let logical_schema = scan.logical_schema();

for batch in read_results {
    let physical_data = batch?;
    let logical_data = transform_to_logical(
        &engine,
        physical_data,
        physical_schema,
        logical_schema,
        scan_file.transform.clone(),
    )?;
    // logical_data now has the columns the scan requested
}

Warning

If ScanFile.transform is present, you must apply it before returning data. Omitting the transform produces incorrect output — missing partition columns, wrong logical names, or nulls where data should appear. If transform is None, the physical data already matches the logical schema and no transformation is needed.

Applying deletion vectors

Delta tables can use deletion vectors to mark rows as logically deleted without rewriting entire data files. If a file has a deletion vector, you must filter out those rows. Call DvInfo::get_selection_vector() on the ScanFile.dv_info to get a boolean mask:

let selection_vector = scan_file
    .dv_info
    .get_selection_vector(&engine, scan.table_root())?;

let filtered = if let Some(sv) = selection_vector {
    logical_data.apply_selection_vector(sv)?
} else {
    // No deletion vector, so all rows are valid
    logical_data
};

Getting deleted row indexes

If your engine works with row indexes rather than boolean masks, DvInfo also provides get_row_indexes(). This method returns a Vec<u64> containing the indexes of rows that should be removed from the result set:

let deleted_rows = scan_file
    .dv_info
    .get_row_indexes(&engine, scan.table_root())?;

if let Some(indexes) = deleted_rows {
    // indexes contains the positions of deleted rows (e.g., [2, 17, 42])
    // Use these to filter rows out of the result set
}

Choose get_selection_vector() when your engine applies boolean masks directly (for example, Arrow’s filter kernel). Choose get_row_indexes() when your engine removes rows by position, or when the deletion vector is sparse and you want to avoid allocating a boolean vector with one entry per row.

Accessing scan schemas

When you read Parquet files yourself (instead of using scan.execute()), you need to know the schemas and predicate to pass to the Parquet reader. Scan exposes three methods for this.

Physical schema

Scan::physical_schema() returns the schema of the underlying data files. This is the schema you pass to the Parquet reader when opening files. It can differ from the logical schema because partition columns are stored in the Delta log metadata, not in the Parquet files themselves.

let physical_schema = scan.physical_schema();
// Pass this to engine.parquet_handler().read_parquet_files(...)

Logical schema

Scan::logical_schema() returns the output schema of the scan — the schema your engine sees after all transforms have been applied. Pass this to transform_to_logical and serialize it alongside physical_schema and each ScanFile when distributing work to remote workers.

let logical_schema = scan.logical_schema();
// Pass this to transform_to_logical(...)

Physical predicate

Scan::physical_predicate() returns the scan’s predicate rewritten in terms of physical column names. If the table uses column mapping, logical column names in the original predicate are translated to the physical names stored in the Parquet files. This is the predicate you can push down into the Parquet reader for row-group filtering.

if let Some(predicate) = scan.physical_predicate() {
    // Push this predicate into the Parquet reader for row-group skipping
}

If the scan has no predicate, this returns None.

Tip

When distributing work to remote workers, serialize the physical schema, logical schema, physical predicate, and the ScanFile data together. Workers need all four to read files correctly.

What’s next

• Column Selection: projecting specific columns
• Distributed Log Replay: distributing log replay for large tables
• Filter Pushdown and File Skipping: predicate-based optimization

Distributed log replay with parallel_scan_metadata()

To distribute Delta log replay across multiple threads or machines, use parallel_scan_metadata(). The standard scan_metadata() method processes the entire log sequentially on a single node. For large tables with V2 checkpoints (which use sidecars or multi-part checkpoint files), that sequential replay can become a bottleneck. parallel_scan_metadata() splits replay into two phases so that the expensive checkpoint processing can run in parallel.

Why two phases?

A Delta table’s transaction log consists of:

• Commit files (JSON), which must be processed sequentially in version order
• Checkpoint files, which can be large but whose constituent parts (sidecars or multi-part checkpoint files) can be processed independently

parallel_scan_metadata() exploits this structure. The sequential phase handles commits and the checkpoint manifest. The parallel phase then distributes the checkpoint leaf files across workers.

Sequential phase

The sequential phase processes commits and the checkpoint manifest. It returns an iterator of ScanMetadata (the same type that scan_metadata() yields):

use delta_kernel::scan::{
    AfterSequentialScanMetadata, ParallelScanMetadata, ParallelState,
    SequentialScanMetadata,
};

let scan = snapshot.scan_builder().build()?;
let mut sequential = scan.parallel_scan_metadata(engine.clone())?;

// Process the sequential phase: commits and checkpoint manifest
for result in sequential.by_ref() {
    let scan_metadata = result?;
    // Process scan metadata (same as scan_metadata())...
}

After exhausting the iterator, call finish() to find out whether a parallel phase is needed:

match sequential.finish()? {
    AfterSequentialScanMetadata::Done => {
        // All log replay completed in the sequential phase.
        // No checkpoint sidecars or multi-part files to process.
    }
    AfterSequentialScanMetadata::Parallel { state, files } => {
        // Parallel phase needed. `files` contains the checkpoint leaf files
        // (sidecars or multi-part checkpoint parts) to process in parallel.
    }
}

Parallel phase

If finish() returns Parallel, partition the files across workers and create a ParallelScanMetadata iterator per partition:

AfterSequentialScanMetadata::Parallel { state, files } => {
    // Unbox and wrap in Arc for sharing across workers
    let state = Arc::new(*state);

    // Distribute files across workers (one file per worker, or batched)
    for file in files {
        let parallel = ParallelScanMetadata::try_new(
            engine.clone(),
            state.clone(),
            vec![file],
        )?;
        for result in parallel {
            let scan_metadata = result?;
            // Process scan metadata (same as the sequential phase)...
        }
    }
}

Each ParallelScanMetadata reads its assigned checkpoint files and processes them through the shared ParallelState, which handles deduplication (filtering out files already seen in the sequential phase).

Serializing state across the network

For distributed engines where workers run on different machines, the ParallelState can be serialized to bytes and shipped over the network:

AfterSequentialScanMetadata::Parallel { state, files } => {
    // On the driver: serialize the state to bytes after the sequential phase
    let serialized_bytes = state.into_bytes()?;

    // Ship `serialized_bytes` and `files` to remote workers...

    // On each worker: reconstruct state from bytes and create a parallel iterator
    let state = Arc::new(ParallelState::from_bytes(engine.as_ref(), &serialized_bytes)?);
    let parallel = ParallelScanMetadata::try_new(engine.clone(), state, my_files)?;
}

Warning

The serialized state may only be deserialized by the same binary version of delta-kernel-rs. Using different versions for serialization and deserialization leads to undefined behavior.

When to use this

Use parallel_scan_metadata() instead of scan_metadata() when:

• The table uses V2 checkpoints with sidecars or multi-part checkpoint files
• Log replay is a bottleneck (large tables with many files)
• Your engine can distribute work across multiple threads or machines

For most use cases, scan_metadata() is sufficient and simpler.

What’s next

• Advanced reads with scan_metadata() for the standard sequential API
• Building a Scan for the basics of building and executing a scan

Time travel and snapshot management

To read a Delta table at a specific version or refresh an existing snapshot to pick up new commits, you use Kernel’s SnapshotBuilder API. The builder supports both full construction from a table URL and incremental updates from an existing Snapshot.

Before reading this page, make sure you understand Building a Scan.

Reading a table at a specific version

Every Delta table maintains a versioned transaction log. Each commit creates a new version. By default, Snapshot::builder_for loads the latest version. To read a specific historical version, chain .at_version() onto the builder.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
// Read the table at version 5
let snapshot = Snapshot::builder_for(&url)
    .at_version(5)
    .build(&engine)?;

println!("Loaded version: {}", snapshot.version());
Ok(())
}
}

The at_version method accepts a u64 version number. If the requested version does not exist in the transaction log, build returns an error.

To read the latest version, omit at_version:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(&url)
    .build(&engine)?;
Ok(())
}
}

Refreshing an existing snapshot

If you already hold a Snapshot (which is wrapped in an Arc as SnapshotRef) and want to check for newer commits, use Snapshot::builder_from instead of rebuilding from scratch. This performs an incremental update: Kernel reads only the new commits since the existing snapshot’s version, avoiding a full log replay.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
// Build an initial snapshot
let snapshot = Snapshot::builder_for(&url)
    .build(&engine)?;
println!("Initial version: {}", snapshot.version());

// Later, refresh to pick up new commits
let updated = Snapshot::builder_from(snapshot)
    .build(&engine)?;
println!("Updated version: {}", updated.version());
Ok(())
}
}

The incremental update follows these rules:

1. If you call .at_version() with the same version the existing snapshot already holds, Kernel returns the existing snapshot without doing any work.
2. If the table has not advanced since the existing snapshot, Kernel returns the existing snapshot.
3. If new commits exist, Kernel replays only the commits after the existing snapshot’s version.
4. You cannot refresh backward. Requesting a version older than the existing snapshot produces an error.

You can also refresh to a specific newer version by combining builder_from with at_version:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(&url).at_version(3).build(&engine)?;
// Refresh from version 3 to exactly version 7
let updated = Snapshot::builder_from(snapshot)
    .at_version(7)
    .build(&engine)?;
Ok(())
}
}

Getting the timestamp of a version

Snapshot exposes get_timestamp, which returns the timestamp of the snapshot’s version in milliseconds since the Unix epoch. When the table has In-Commit Timestamps (ICT) enabled, the method returns the ICT value stored inside the commit. Otherwise, it falls back to the filesystem’s last-modified time on the commit file.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let snapshot = Snapshot::builder_for(&url)
    .at_version(5)
    .build(&engine)?;

let timestamp_ms = snapshot.get_timestamp(&engine)?;
println!("Version {} was committed at {} ms", snapshot.version(), timestamp_ms);
Ok(())
}
}

Note

get_timestamp requires an &dyn Engine because it may need to read the commit file from storage when In-Commit Timestamps are enabled.

Resolving timestamps to versions

To time travel by timestamp rather than by version number, use the history_manager module. It exposes three helpers that translate timestamps (in milliseconds since the Unix epoch) into the version numbers you can pass to at_version.

Each helper takes a Snapshot reference, which defines the searchable version range. Pass the latest snapshot if you want to search the entire history.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
use delta_kernel::history_manager::latest_version_as_of;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
// 1. Load the latest snapshot to define the search range.
let latest = Snapshot::builder_for(&url).build(&engine)?;

// 2. Resolve a timestamp (Jan 1, 2024 UTC) to a version number.
let timestamp_ms = 1_704_067_200_000;
let version = latest_version_as_of(&latest, &engine, timestamp_ms)?;

// 3. Time travel to that version.
let snapshot = Snapshot::builder_for(&url)
    .at_version(version)
    .build(&engine)?;
println!("Resolved timestamp {timestamp_ms} to version {version}");
Ok(())
}
}

first_version_after is the symmetric variant. It returns the earliest version whose timestamp is at or after the requested timestamp, which is useful for picking up changes that happened after a known point in time.

To resolve a timestamp range (for example, when reading a change feed between two points in time), call timestamp_range_to_versions. The end timestamp is optional. Pass None to indicate no upper bound.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
use delta_kernel::history_manager::timestamp_range_to_versions;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let latest = Snapshot::builder_for(&url).build(&engine)?;
let start_ms = 1_704_067_200_000; // Jan 1, 2024 UTC
let end_ms = 1_706_745_600_000;   // Feb 1, 2024 UTC

let (start_version, end_version) =
    timestamp_range_to_versions(&latest, &engine, start_ms, Some(end_ms))?;
Ok(())
}
}

These helpers return errors when the timestamp falls outside the range of commits visible from the snapshot, or when the requested range is empty. See the LogHistoryError variants for the specific failure modes.

When to use builder_for vs. builder_from

The key difference is cost. builder_for replays the transaction log from the most recent checkpoint. builder_from replays only the commits after the existing snapshot’s version. For long-lived connectors that periodically check for updates, builder_from avoids redundant log replay.

What’s next

• Column Selection to read specific columns from a historical snapshot.
• Filter Pushdown and File Skipping to apply predicates that reduce I/O.
• Advanced Reads with scan_metadata() for distributed scan execution.

Reading change data feed

To read a row-level changelog of what changed between two versions of a Delta table, you use TableChanges and TableChangesScan. This gives you every insert, update, and delete that occurred in the specified version range, with metadata columns that identify the type of change and the commit it came from.

Before reading this page, make sure you understand Building a Scan.

Prerequisites

Change Data Feed (CDF) is a Delta feature that records row-level changes (inserts, updates, deletes) as part of each commit. The table must have the delta.enableChangeDataFeed table property set to true for every version in the range you want to read. If CDF is disabled for any version in the range, TableChanges::try_new returns an error.

Creating a TableChanges

TableChanges::try_new takes a table URL, an Engine reference, a start version, and an optional end version. It validates that CDF is enabled and that the schema is compatible across the requested range.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::table_changes::TableChanges;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/my-table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
// Read changes from version 0 through version 5 (inclusive)
let table_changes = TableChanges::try_new(url, &engine, 0, Some(5))?;
Ok(())
}
}

If you omit the end version by passing None, Kernel defaults to the latest version of the table at the time of the call.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::table_changes::TableChanges;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/my-table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
// Read all changes from version 0 to the latest version
let table_changes = TableChanges::try_new(url, &engine, 0, None)?;
Ok(())
}
}

try_new performs several validation checks before returning:

• CDF must be enabled at both the start and end versions.
• The table schema at the start and end versions must be identical.
• No unsupported reader features (other than deletion vectors) are enabled.

If any check fails, it returns an Error.

CDF metadata columns

The schema of TableChanges is the table’s schema at the end version plus three additional columns:

You can access the full schema (table columns plus CDF columns) via table_changes.schema(). You can include any of the CDF columns in a projection, the same way you would project regular table columns.

Building and executing a scan

TableChangesScanBuilder works like ScanBuilder for regular scans. You can optionally project columns with with_schema and filter rows with with_predicate.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::expressions::{column_expr, Scalar};
use delta_kernel::table_changes::TableChanges;
use delta_kernel::{DeltaResult, Predicate};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/my-table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let table_changes = TableChanges::try_new(url, &engine, 0, Some(5))?;
// 1. Project only the columns you need (including CDF metadata)
let schema = table_changes
    .schema()
    .project(&["name", "age", "_change_type", "_commit_version"])?;

// 2. Build a predicate on table columns
let predicate = Arc::new(Predicate::gt(column_expr!("age"), Scalar::from(25)));

// 3. Build and execute the scan
let scan = table_changes
    .into_scan_builder()
    .with_schema(schema)
    .with_predicate(predicate)
    .build()?;

for data in scan.execute(Arc::new(engine))? {
    let batch = data?;
    println!("Got {} rows of change data", batch.len());
}
Ok(())
}
}

If you skip with_schema, the scan returns all table columns plus all three CDF columns. If you skip with_predicate, no filtering is applied.

Note

Predicates on the CDF metadata columns (_change_type, _commit_version, _commit_timestamp) are not currently supported. Apply predicates only to regular table columns. Filtering on CDF columns after the scan returns is still possible in your connector code.

Note

Like regular scans, CDF filtering is best-effort. The scan may return rows that do not match your predicate. Apply row-level filtering in your connector if exact results are required.

Using Arc<TableChanges>

If you need to retain ownership of the TableChanges (for example, to inspect its schema after building the scan), use scan_builder on an Arc<TableChanges> instead of into_scan_builder:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::table_changes::TableChanges;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/my-table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
let table_changes = Arc::new(
    TableChanges::try_new(url, &engine, 0, Some(5))?
);

let scan = table_changes.clone().scan_builder().build()?;

// You can still access table_changes after building the scan
println!("CDF range: version {} to {}", table_changes.start_version(), table_changes.end_version());
Ok(())
}
}

What’s next

• Building a Scan: the core scan pattern that CDF scanning builds on
• Filter Pushdown and File Skipping: predicate construction and data skipping behavior
• Column Selection: projecting specific columns from a scan

Creating a Table

To create a new Delta table, you configure a CreateTableTransactionBuilder with your schema and table options, then commit it. For a quick end-to-end example that creates a table, writes data, and reads it back, see Quick Start: Writing a Table.

Basic table creation

The create_table function returns a builder that you configure and then commit:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let schema = Arc::new(StructType::try_new([
    StructField::not_null("id", DataType::INTEGER),
    StructField::nullable("name", DataType::STRING),
    StructField::nullable("age", DataType::INTEGER),
])?);

create_table(url.as_str(), schema, "my-app/1.0")
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;
Ok(())
}
}

The three required arguments are:

• path: Where to create the table (local path or URI like s3://bucket/path)
• schema: The table’s column definitions as a StructType
• engine_info: A string identifying your application (stored in the commit log)

.build() validates the inputs and creates a CreateTableTransaction. .commit() writes version 0 of the table, producing the initial Protocol and Metadata actions.

Defining a schema

Schemas are built from StructFields, each with a name, data type, and nullability:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
use std::sync::Arc;
use delta_kernel::schema::{ArrayType, DataType, MapType, StructField, StructType};

let schema = Arc::new(StructType::try_new([
    // Non-nullable integer
    StructField::not_null("id", DataType::LONG),

    // Nullable string
    StructField::nullable("name", DataType::STRING),

    // Nested struct
    StructField::nullable(
        "address",
        StructType::try_new([
            StructField::nullable("street", DataType::STRING),
            StructField::nullable("city", DataType::STRING),
        ])?,
    ),

    // Array of integers
    StructField::nullable(
        "scores",
        ArrayType::new(DataType::INTEGER, true),  // true = elements are nullable
    ),

    // Map from string to double
    StructField::nullable(
        "metrics",
        MapType::new(DataType::STRING, DataType::DOUBLE, true),  // true = values are nullable
    ),
])?);
Ok(())
}
}

For the full list of supported data types, see Schemas and Data Types.

Table properties

You can set custom application properties on the table:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let schema = Arc::new(StructType::try_new([
    StructField::not_null("id", DataType::INTEGER),
])?);
create_table(url.as_str(), schema, "my-app/1.0")
    .with_table_properties([
        ("myapp.version", "2.0"),
        ("myapp.owner", "data-team"),
    ])
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;
Ok(())
}
}

Custom properties (those not starting with delta.) are always allowed. Delta properties (delta.*) are validated against an allow list. Kernel only permits properties for features it supports.

Clustered tables

You can create a clustered table using with_data_layout. Clustering optimizes data file layout for queries that filter on the clustering columns:

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::data_layout::DataLayout;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let schema = Arc::new(StructType::try_new([
    StructField::not_null("id", DataType::INTEGER),
    StructField::nullable("region", DataType::STRING),
    StructField::nullable("timestamp", DataType::TIMESTAMP),
])?);

create_table(url.as_str(), schema, "my-app/1.0")
    .with_data_layout(DataLayout::clustered(["region", "timestamp"]))
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;
Ok(())
}
}

Clustering columns can be top-level or nested. The DataLayout::clustered([...]) helper treats each string as a single top-level column name. To cluster on a nested column, construct the DataLayout::Clustered variant directly with a multi-segment ColumnName:

use delta_kernel::expressions::ColumnName;
use delta_kernel::transaction::data_layout::DataLayout;

let layout = DataLayout::Clustered {
    columns: vec![
        ColumnName::new(["region"]),
        ColumnName::new(["address", "city"]),
    ],
};

For nested clustering columns, every intermediate path component must be a struct field, and the leaf column must have a stats-eligible primitive type.

The kernel automatically enables the required table features (DomainMetadata and ClusteredTable) when clustering is specified.

Partitioned tables

You can create a partitioned table using DataLayout::partitioned(). Partitioning organizes data files into directories based on partition column values, which allows readers to skip entire directories when filtering on those columns.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField, StructType};
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::data_layout::DataLayout;
use delta_kernel::DeltaResult;
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let schema = Arc::new(StructType::try_new([
    StructField::not_null("id", DataType::INTEGER),
    StructField::nullable("name", DataType::STRING),
    StructField::not_null("year", DataType::INTEGER),
    StructField::not_null("month", DataType::INTEGER),
])?);

create_table(url.as_str(), schema, "my-app/1.0")
    .with_data_layout(DataLayout::partitioned(["year", "month"]))
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;
Ok(())
}
}

Validation rules

build() validates partition columns against these rules:

• Top-level only: partition columns must be top-level fields in the schema. Nested paths like address.city are not supported.
• Primitive types only: each partition column must have a primitive type (STRING, INTEGER, LONG, DATE, TIMESTAMP, etc.). Struct, array, and map types are rejected.
• At least one non-partition column: the schema must contain at least one column that is not a partition column. A table with every column marked as a partition column is invalid.
• No duplicates: the same column cannot appear more than once in the partition column list.
• Schema presence: every partition column must exist in the schema.
• At least one partition column: if DataLayout::partitioned(...) is used, the column list cannot be empty.

Physical vs. logical schema

Partition column values are stored in the directory path rather than inside the data files themselves. When you call WriteContext::physical_schema(), the returned schema excludes partition columns. Your connector writes data files using the physical schema, and Kernel reconstructs the full logical schema (including partition values from the path) at read time.

Note

Partitioning and clustering are mutually exclusive. You can call with_data_layout() with either DataLayout::partitioned() or DataLayout::clustered(), but not both. Only the last with_data_layout() call takes effect.

The Committer

The build() method takes a Box<dyn Committer> that controls how the commit is persisted:

• FileSystemCommitter: For standalone filesystem-based tables. Writes commit files directly to _delta_log/ using atomic put-if-absent. This is the default for most use cases.

• Custom Committer: For catalog-managed tables (e.g. Unity Catalog), you implement the Committer trait to route commits through the catalog. See Catalog-Managed Tables.

Handling the result

commit() returns a CommitResult:

match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        println!("Created table at version {}", committed.commit_version());
    }
    CommitResult::ConflictedTransaction(_) => {
        // Another writer created the table concurrently
    }
    CommitResult::RetryableTransaction(retry) => {
        // Transient I/O error. Safe to retry.
        println!("Retryable error: {}", retry.error);
    }
}

For table creation, CommittedTransaction is the expected result (version 0). ConflictedTransaction means another process created the table between your existence check and commit. RetryableTransaction indicates a transient error.

Validations

build() performs these checks before creating the transaction:

Path and existence:

• The path is a valid URI
• No Delta table already exists at that path

Schema:

• The schema has at least one field
• Column names are non-empty
• When column mapping is disabled, column names cannot contain Parquet special characters (space, comma, semicolon, braces, parentheses, tab, newline, or =). Enable column mapping to use these characters.
• When column mapping is enabled, column names cannot contain newlines
• No duplicate column names, case-insensitive, across all nested paths
• Non-null columns (fields with nullable: false) are accepted. When the schema contains any non-null field anywhere in the tree, Kernel auto-enables the invariants writer feature on the new table. This matches Delta-Spark, which treats nullable: false as an implicit column invariant.

Table properties:

• Custom properties (not starting with delta.) are always allowed
• delta.* properties are checked against an allow list
• Feature signals (delta.feature.*) are checked against an allow list
• Catalog-managed tables cannot set delta.enableInCommitTimestamps=false. Kernel auto-enables ICT for catalog-managed tables.

Data layout:

• Clustering columns (if specified) exist in the schema. Both top-level and nested paths are supported. Each leaf column must have a stats-eligible primitive type.
• Partition columns (if specified) exist in the schema, are top-level, have primitive types, and leave at least one non-partition column. The partition column list cannot be empty.

Auto-enabled features

build() auto-enables certain table features based on the schema, properties, and data layout, so you do not need to set them manually. The triggers fall into four groups. Each enabled feature is either a reader/writer feature (bumps both the reader and writer protocol) or a writer-only feature (bumps the writer protocol only).

Table-property-derived

Features enabled from the values of delta.* properties you set with with_table_properties() or from the feature signals that appear in that property bag.

Data-layout-derived

Features enabled by the data layout you pass to with_data_layout().

Schema-derived

Features enabled when specific column types or annotations appear in the schema.

Feature-signal only

Features enabled only when you explicitly include the feature signal in with_table_properties(). Kernel does not derive these from the schema or from other properties.

Note

Auto-enabled features cannot always be set explicitly via delta.feature.*=supported. Kernel enforces that the protocol and metadata stay consistent with the schema and data layout, so some features are controllable only through their driving condition:

• variantType and timestampNtz are enabled only when the schema uses those types. Explicit feature signals for them are rejected.
• clusteredTable is enabled only by passing clustering columns to with_data_layout(). delta.feature.clustering=supported is rejected.
• For catalogManaged tables, delta.enableInCommitTimestamps=false is rejected because ICT is required.

Other auto-enabled features (such as columnMapping, inCommitTimestamp) can still be enabled explicitly via delta.feature.X=supported if you want to pre-enable them before the schema or property trigger applies.

What’s next

• Appending Data to write rows to an existing table
• Writing to Partitioned Tables to write data with partition values
• Quick Start: Writing a Table for a complete end-to-end example

Appending Data

To append data to an existing Delta table, you create a Transaction from a Snapshot, write Parquet files through the engine, register them, and commit. For a quick end-to-end example that creates a table and writes data, see Quick Start: Writing a Table.

The write flow

Appending data to a Delta table follows these steps:

1. Get a Snapshot of the table
2. Create a Transaction from the snapshot
3. Get the WriteContext from the transaction
4. Write Parquet files using the engine and WriteContext
5. Register the written files with the transaction via add_files
6. Commit the transaction

The following example assumes you already have an engine: DefaultEngine, which provides an async write_parquet helper. If you use a custom Engine, the step 4 may differ.

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::arrow::array::{Int32Array, RecordBatch, StringArray};
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::transaction::CommitResult;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
// 1. Get a snapshot
let snapshot = Snapshot::builder_for(url).build(&engine)?;

// 2. Create a transaction
let mut txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_engine_info("my-app/1.0")
    .with_data_change(true);

// 3. Get write context
let write_context = Arc::new(txn.unpartitioned_write_context()?);

// 4. Write Parquet file(s)
// Assumes the table schema is: name (STRING), age (INTEGER), city (STRING)
let batch = RecordBatch::try_new(
    Arc::new(write_context.logical_schema().as_ref().try_into_arrow()?),
    vec![
        Arc::new(StringArray::from(vec!["Dave", "Eve", "Frank"])),
        Arc::new(Int32Array::from(vec![4, 5, 6])),
        Arc::new(StringArray::from(vec!["Austin", "Boston", "Chicago"])),
    ],
)?;
let data = ArrowEngineData::new(batch);
let file_metadata = engine
    .write_parquet(&data, write_context.as_ref())
    .await?;

// 5. Register the files
txn.add_files(file_metadata);

// 6. Commit
match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        println!("Committed version {}", committed.commit_version());
    }
    _ => eprintln!("commit did not succeed"),
}
Ok(())
}

Creating a transaction

A transaction is created from a snapshot. The snapshot pins the table version you are writing against:

let mut txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_engine_info("my-app/1.0")
    .with_data_change(true);

The builder methods:

The WriteContext

Before writing data, obtain a WriteContext. A WriteContext bundles everything needed to correctly write Parquet files:

// For unpartitioned tables
let write_context = txn.unpartitioned_write_context()?;

// For partitioned tables, pass the partition values for this file
let write_context = txn.partitioned_write_context(partition_values)?;

For partitioned tables, see Writing to Partitioned Tables.

WriteContext provides:

Writing Parquet files

The DefaultEngine provides an async helper for writing Parquet:

let file_metadata = engine
    .write_parquet(&data, write_context.as_ref())
    .await?;

• data: An ArrowEngineData wrapping a RecordBatch matching the logical schema
• write_context: From unpartitioned_write_context() or partitioned_write_context()

DefaultEngine::write_parquet handles the logical-to-physical transformation, generates a unique filename, writes the file, collects statistics, and returns file metadata that you pass to txn.add_files().

You can call write_parquet and add_files multiple times to write multiple files in one transaction.

Note

Methods that produce or register data files (unpartitioned_write_context, partitioned_write_context, add_files, stats_schema) are gated by the SupportsDataFiles trait bound and are available on standard write transactions but not on metadata-only transaction states (such as a future AlterTable).

Committing

commit() consumes the transaction and returns a CommitResult:

match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        println!("Committed version {}", committed.commit_version());
    }
    _ => {
        eprintln!("commit did not succeed");
    }
}

Note

commit() returns a CommitResult with three variants: CommittedTransaction on success, ConflictedTransaction if another writer committed first, and RetryableTransaction for transient IO errors. Automatic conflict resolution is not yet supported. A blind append to a table with no concurrent writers always succeeds.

Blind appends

A blind append is a write that adds new files without reading or depending on existing table state. To mark a transaction as a blind append, call with_blind_append() during construction:

let txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_blind_append();

Kernel records isBlindAppend: true in the commit’s commitInfo action. This flag enables conflict resolution optimizations: two blind appends to the same table can never conflict with each other, because neither depends on the other’s output.

Kernel validates the following rules at commit time. If any rule is violated, commit() returns an error:

Tip

Mark your transaction as a blind append whenever you are inserting new data without reading the table first. This gives the committer the information it needs to resolve conflicts safely in multi-writer scenarios.

Custom commit info

Kernel always writes a commitInfo action for every commit. To include your own fields in that action, call with_commit_info() with your custom data and its schema:

let txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_commit_info(engine_commit_info, commit_info_schema);

The engine_commit_info argument is a Box<dyn EngineData> containing the fields you want to add, and commit_info_schema is the corresponding SchemaRef. Kernel merges your fields into the final commitInfo action.

Kernel reserves certain fields and overrides them regardless of what you provide. Do not set these fields in your custom commit info:

Any field in your custom data that shares a name with a Kernel-reserved field is replaced with Kernel’s value. Fields with names that do not collide are preserved as-is in the final commitInfo.

After committing

A successful commit returns a CommittedTransaction with access to a post-commit snapshot and post-commit statistics:

let committed = match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => c,
    _ => panic!("unexpected result"),
};

// The version that was committed
let version = committed.commit_version();

// Post-commit statistics help you decide when to run maintenance
let stats = committed.post_commit_stats();
println!("Commits since last checkpoint: {}", stats.commits_since_checkpoint);
println!("Commits since last log compaction: {}", stats.commits_since_log_compaction);

// The post-commit snapshot reflects the table state after this commit.
// Use it for maintenance operations like publishing, checkpointing, and checksums.
if let Some(snapshot) = committed.post_commit_snapshot() {
    // Write a checksum file for this version
    let (result, snapshot) = snapshot.write_checksum(&engine)?;

    // For catalog-managed tables: snapshot.publish(engine, committer)?
    // For checkpointing: snapshot.checkpoint(&engine)?
}

Post-commit statistics

CommittedTransaction::post_commit_stats() returns a PostCommitStats struct with two fields:

Use these values to decide when to trigger maintenance. For example, you might checkpoint every 10 commits or compact the log every 50 commits.

Writing a checksum

You can call write_checksum() on the post-commit snapshot to write a checksum file (CRC file), which enables faster snapshot loading and table state validation:

if let Some(snapshot) = committed.post_commit_snapshot() {
    let (checksum_result, updated_snapshot) = snapshot.write_checksum(&engine)?;
    // Use updated_snapshot for subsequent operations (it includes the new CRC file)
}

write_checksum() returns a ChecksumWriteResult and an updated SnapshotRef. ChecksumWriteResult::Written means the CRC file was created successfully. ChecksumWriteResult::AlreadyExists means a CRC file already exists at this version, and the original snapshot is returned unchanged. Per the Delta protocol, writers must not overwrite existing checksum files.

Note

write_checksum() requires in-memory CRC information, which is only available on post-commit snapshots. Calling it on a snapshot loaded from disk (without a pre-computed CRC) returns a ChecksumWriteUnsupported error.

The post-commit snapshot is the entry point for maintenance operations that should happen after a successful write. See Checkpointing and Catalog-Managed Tables for details.

What’s next

• Writing to Partitioned Tables covers writing data with partition values.
• Creating a Table covers creating a new table from scratch.
• Checkpointing explains when and how to write checkpoints.

Writing to partitioned tables

To write data to a partitioned table, you create a WriteContext for each distinct set of partition values, write Parquet files through the engine for each one, and commit. Kernel validates partition values, serializes them per the Delta protocol, and constructs the correct directory paths.

Before reading this page, make sure you understand Appending Data and Creating a Table.

How partitioned writes differ

For unpartitioned tables, you create one WriteContext and write all data through it. For partitioned tables, you create one WriteContext per distinct partition value combination. Partition values are baked into the WriteContext at creation time, not passed at write time.

Unpartitioned:  1 WriteContext  -->  write all data
Partitioned:    1 WriteContext per distinct partition  -->  write that partition's data

The write flow

The pattern for partitioned writes is: group your data by partition values, create a WriteContext per group, and write each group.

extern crate delta_kernel;
extern crate tokio;
use std::collections::HashMap;
use delta_kernel::arrow::array::RecordBatch;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::expressions::Scalar;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/partitioned_table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let mut txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("INSERT".to_string())
    .with_data_change(true);

// Suppose you have data grouped by partition values already.
// For each partition, create a WriteContext and write.
let partitions: Vec<(HashMap<String, Scalar>, RecordBatch)> = todo!("group your data");

for (partition_values, batch) in partitions {
    // 1. Create a WriteContext for this partition
    let wc = txn.partitioned_write_context(partition_values)?;

    // 2. Write the data (physical schema excludes partition columns)
    let data = ArrowEngineData::new(batch);
    let file_metadata = engine.write_parquet(&data, &wc).await?;

    // 3. Register the written file
    txn.add_files(file_metadata);
}

// Commit all partitions in a single transaction
txn.commit(&engine)?;
Ok(())
}

Each partitioned_write_context call takes a HashMap<String, Scalar> mapping logical partition column names to typed values:

let partition_values = HashMap::from([
    ("year".to_string(), Scalar::Integer(2024)),
    ("month".to_string(), Scalar::Integer(3)),
]);
let wc = txn.partitioned_write_context(partition_values)?;

Key points:

• Typed values: partition values are Scalar values, not strings. Pass Scalar::Integer(2024) for an integer column, not Scalar::String("2024".into()). Kernel rejects type mismatches.
• Case-insensitive keys: "YEAR" matches schema column "year". Kernel normalizes to the schema case.
• Physical schema: wc.physical_schema() excludes partition columns. Data files contain only the non-partition columns.

Tip

To get the partition column names at runtime, call txn.logical_partition_columns(). This is useful when your connector handles arbitrary tables and does not know the partition columns in advance.

Grouping data by partition values

How you group data by partition values is up to your connector. Kernel’s contract is that each partitioned_write_context call receives a HashMap<String, Scalar> for one distinct partition, and the corresponding data files contain only that partition’s rows.

Kernel provides serialize_partition_value as a public utility for building hashable group keys from Scalar values. It returns a DeltaResult<Option<String>> per value, which you can collect into a Vec<Option<String>> group key for use in a HashMap.

Partition value validation

partitioned_write_context validates the provided values before creating the WriteContext:

If validation fails, partitioned_write_context returns an error before any data reaches disk.

What Kernel handles

When you call partitioned_write_context, Kernel performs the following steps internally. Your connector does not need to implement any of this:

1. Key validation: all partition columns present, no extra keys
2. Case normalization: keys matched case-insensitively against the schema
3. Type checking: each Scalar’s type must match the partition column’s schema type
4. Value serialization: Scalar values converted to protocol-compliant strings (for example, Scalar::Date(19723) becomes "2024-01-01")
5. Key translation: logical column names translated to physical names when column mapping is enabled

write_dir() on the resulting WriteContext returns the directory where data files should be written. Without column mapping, this is a Hive-style path like <table_root>/year=2024/month=3/. With column mapping enabled, this is a random two-character prefix directory like <table_root>/aB/ that avoids exposing physical column names and distributes files across object store prefixes to prevent S3 hotspots.

Note

Hive partition path segments are URI-encoded on top of Hive escaping, matching the on-disk layout produced by Delta-Spark and Delta-Kernel-Java. For example, a partition value 2025-03-31T15:30:00Z produces the path segment p=2025-03-31T15%253A30%253A00Z/ (the : is Hive-escaped to %3A, and the % is then URI-encoded to %25). The same encoded path is recorded in the add.path field of the commit. Callers writing through the filesystem must URI-decode write_dir() once before using it as an OS path.

What’s next

• Appending Data covers the general write flow, commit handling, and blind appends.
• Creating a Table covers creating partitioned tables.
• Idempotent Writes for at-most-once write guarantees.

Removing data

To remove data files from an existing Delta table, you scan the table’s file metadata, select which files to remove, and commit those removals as a Transaction. Kernel tracks removes at file-level granularity. Each removed file produces a remove action in the Delta transaction log.

Before reading this page, make sure you understand Appending Data and Advanced Reads with scan_metadata().

The remove flow

Removing files follows these steps:

1. Get a Snapshot of the table
2. Create a Transaction from the snapshot
3. Build a Scan and call scan_metadata() to get file-level metadata
4. Modify the selection vector to mark files for removal
5. Pass the modified FilteredEngineData to Transaction::remove_files()
6. Commit the transaction

Snapshot ──> Transaction
                │
Snapshot ──> Scan ──> scan_metadata()
                          │
                    ScanMetadata { scan_files: FilteredEngineData, ... }
                          │
                    Modify selection vector
                          │
                    txn.remove_files(modified_scan_files)
                          │
                    txn.commit(engine)

Getting file metadata with scan_metadata()

Scan::scan_metadata() returns an iterator of ScanMetadata. Each ScanMetadata contains a scan_files field of type FilteredEngineData. This FilteredEngineData holds one row per file in the table, along with a selection vector that indicates which rows (files) are active.

The underlying data conforms to the schema returned by scan_row_schema():

{
   path: string,
   size: long,
   modificationTime: long,
   stats: string,
   deletionVector: {
     storageType: string,
     pathOrInlineDv: string,
     offset: int,
     sizeInBytes: int,
     cardinality: long,
   },
   fileConstantValues: {
     partitionValues: map<string, string>,
     tags: map<string, string>,
     baseRowId: long,
     defaultRowCommitVersion: long,
     clusteringProvider: string,
   }
}

You don’t need to construct this data yourself. The scan produces it for you. For full details on working with scan_metadata(), see Advanced Reads with scan_metadata().

Note

If your scan was built with a partition predicate (Scan::with_predicate), the scan-row schema also includes a partitionValues_parsed field. Kernel drops that field internally when transforming the rows for remove_files(), so you don’t need to handle it.

Selecting files for removal

FilteredEngineData pairs engine data with a boolean selection vector. Each true entry marks a row (file) as selected. When you pass a FilteredEngineData to remove_files(), Kernel removes every file whose corresponding selection vector entry is true.

To control which files are removed, decompose the FilteredEngineData with into_parts(), modify the selection vector, and reassemble with FilteredEngineData::try_new():

let (data, mut selection_vector) = scan_files.into_parts();

// Example: deselect all files, then select only the ones you want to remove
for entry in selection_vector.iter_mut() {
    *entry = false;
}
// Mark specific files for removal (e.g., based on your own logic)
selection_vector[0] = true;

let files_to_remove = FilteredEngineData::try_new(data, selection_vector)?;

Calling remove_files()

Once you have a FilteredEngineData with the correct selection vector, pass it to the transaction:

txn.remove_files(files_to_remove);

The signature is:

pub fn remove_files(&mut self, remove_metadata: FilteredEngineData)

You can call remove_files() multiple times to remove files from different scan_metadata() batches. Each call appends to the transaction’s list of pending removals.

Note

remove_files() is available on transaction states that produce data files (gated by the SupportsDataFiles trait bound). Metadata-only transaction states cannot register file removals.

Full example

This example removes the first file from a filesystem-backed table:

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::engine_data::FilteredEngineData;
use delta_kernel::transaction::CommitResult;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/my_table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
// 1. Get a snapshot
let snapshot = Snapshot::builder_for(url).build(&engine)?;

// 2. Create a transaction
let mut txn = snapshot
    .clone()
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_operation("DELETE".to_string());

// 3. Build a scan and get file metadata
let scan = snapshot.scan_builder().build()?;

for metadata in scan.scan_metadata(&engine)? {
    let metadata = metadata?;

    // 4. Modify the selection vector to pick files for removal
    let (data, mut selection_vector) = metadata.scan_files.into_parts();

    // Deselect everything, then select only the first file
    for entry in selection_vector.iter_mut() {
        *entry = false;
    }
    if !selection_vector.is_empty() {
        selection_vector[0] = true;
    }

    let files_to_remove = FilteredEngineData::try_new(data, selection_vector)?;

    // 5. Register the files for removal
    txn.remove_files(files_to_remove);
}

// 6. Commit the transaction
match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        println!("Committed version {}", committed.commit_version());
    }
    _ => eprintln!("commit did not succeed"),
}
Ok(())
}

Tip

In practice, you would inspect file statistics or partition values to decide which files to remove rather than selecting by index. Use a visitor on the FilteredEngineData to read per-file metadata before modifying the selection vector.

Change Data Feed restriction

Warning

If the table has Change Data Feed enabled (delta.enableChangeDataFeed = true), you cannot add and remove files in the same transaction. Kernel does not yet support writing the CDC files that Delta requires for DML operations that both add and remove data. If you attempt this, commit() returns an error. Use separate transactions: one to add files, another to remove files.

Blind appends and remove_files()

A transaction marked with with_blind_append() cannot remove files. Blind appends are optimized for the append-only case and reject any removals at commit time. If your transaction removes files, do not call with_blind_append().

What’s next

• Idempotent Writes covers how to make writes safely repeatable.
• Domain Metadata explains how to attach custom metadata to a commit.
• Advanced Reads with scan_metadata() covers the full scan_metadata() API used to obtain FilteredEngineData.

Domain metadata

To store application-specific configuration alongside a Delta table, you use domain metadata. Each domain is a named key-value pair persisted in the Delta transaction log. Your connector can write, read, and remove domain metadata through the Kernel API without affecting table data.

Domain metadata is useful for tracking connector-specific state, feature flags, or custom configuration that should travel with the table and survive across sessions.

Before reading this page, make sure you understand Appending Data and how transactions work.

Writing domain metadata

To attach domain metadata to a commit, call with_domain_metadata() on the transaction. This method is available on both create-table and existing-table transactions.

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_domain_metadata(
        "myConnector.settings".to_string(),
        r#"{"version": 1, "compress": true}"#.to_string(),
    )
    .with_operation("UPDATE METADATA".to_string());

txn.commit(&engine)?;
Ok(())
}

The with_domain_metadata signature takes two String arguments:

pub fn with_domain_metadata(self, domain: String, configuration: String) -> Self

The domain identifies the metadata namespace and configuration holds the value. The configuration is an opaque string. You can store JSON, plain text, or any format your connector understands.

You can set metadata for multiple distinct domains in the same transaction by calling with_domain_metadata more than once with different domain names.

Removing domain metadata

To remove a domain from an existing table, call with_domain_metadata_removed() on an existing-table transaction. This method is not available on create-table transactions because there is no metadata to remove from a table that does not exist yet.

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_domain_metadata_removed("myConnector.settings".to_string())
    .with_operation("REMOVE METADATA".to_string());

txn.commit(&engine)?;
Ok(())
}

If the domain does not exist in the log, the removal is a no-op. Kernel handles this gracefully during commit.

Reading domain metadata

To read domain metadata from a table, call get_domain_metadata() on a Snapshot.

Tip

Reading domain data from a snapshot is efficient if a CRC file was present when loading the snapshot. Otherwise determining domain metadata requires performing a log replay. Connectors that heavily rely on get_domain_metadata() should ensure checksum CRC files are written with each commit.

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;

// Returns Ok(Some(config)) if the domain exists, Ok(None) if it does not
let config: Option<String> = snapshot.get_domain_metadata("myConnector.settings", &engine)?;

match config {
    Some(value) => println!("Domain config: {value}"),
    None => println!("No metadata found for this domain"),
}
Ok(())
}

Warning

get_domain_metadata() rejects domain names that start with delta. and returns an error. The delta.* namespace is reserved for Kernel’s internal use (row tracking, clustering). You can only read user-defined domains through this API.

Constraints and validation

Kernel validates domain metadata operations at commit time. The following rules apply:

• One domain per transaction. Each domain name can appear at most once per transaction. You cannot set and remove the same domain in a single commit, and you cannot set the same domain twice. If you include a duplicate domain, the commit fails with an error.

• Reserved prefix. Domain names starting with delta. are reserved for Kernel’s internal use (e.g., clustering metadata). Attempting to read, write, or remove a delta. domain through the public API returns an error.

• Feature requirement. Domain metadata operations require the domainMetadata writer feature to be enabled on the table (writer version 7). If the feature is not enabled, the commit fails.

• No removals on create-table. The with_domain_metadata_removed() method is only available on existing-table transactions. The Rust type system prevents calling it on a create-table transaction at compile time.

• Multiple domains allowed. Although each domain can appear only once, you can set or remove metadata for multiple distinct domains in the same transaction.

Note

Validation is deferred until commit(). The builder methods with_domain_metadata() and with_domain_metadata_removed() do not check for duplicates or reserved prefixes eagerly. Errors surface when you call commit().

What’s next

• Idempotent Writes covers SetTransaction actions for at-most-once write guarantees
• Appending Data explains the full write flow for adding data files to a table
• Creating a Table shows how to set domain metadata during table creation

Idempotent writes

To guarantee at-most-once semantics for retryable write pipelines (e.g., a streaming job or a queue consumer), you attach a transaction identifier to each commit. If the same logical write is attempted twice, the second attempt becomes a no-op. Delta supports this through SetTransaction actions recorded in the transaction log.

How it works

Each transaction can carry a SetTransaction action containing:

• app_id: a string identifying the application (e.g., "my-streaming-job")
• version: an application-defined integer. The Delta protocol leaves its semantics up to the connector. A common pattern is a monotonically increasing counter (batch number, Kafka offset, etc.), but other schemes are valid. The protocol only guarantees that the latest committed version per app_id is retrievable.

Before writing, check the table for the latest committed version for your app_id. If that version indicates the write has already been applied, skip it. The comparison logic is up to you.

Writing with a transaction ID

Call with_transaction_id() on the transaction and check get_app_id_version() before committing:

extern crate delta_kernel;
extern crate tokio;
use std::sync::Arc;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::transaction::CommitResult;
use delta_kernel::{DeltaResult, Snapshot};
#[tokio::main]
async fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let app_id = "my-streaming-job";
let batch_version: i64 = 42;

// Check if this batch was already committed
if let Some(committed_version) = snapshot.get_app_id_version(app_id, &engine)? {
    if committed_version >= batch_version {
        println!("batch {batch_version} already committed (table has {committed_version})");
        return Ok(());
    }
}

// Not yet committed. Proceed with the write.
let txn = snapshot
    .transaction(Box::new(FileSystemCommitter::new()), &engine)?
    .with_transaction_id(app_id.to_string(), batch_version)
    .with_operation("STREAMING UPDATE".to_string());

// ... write data, add files, commit ...
Ok(())
}

The >= check above assumes the application uses monotonically increasing versions. A connector with a different scheme (for example, tracking a set of known batch IDs) would compare differently.

The SetTransaction action is written to the commit log alongside the data actions. On the next run, get_app_id_version() will find it and the duplicate write is skipped.

Reading the latest transaction version

Snapshot::get_app_id_version() scans the log for the latest SetTransaction with the given app_id:

let latest: Option<i64> = snapshot.get_app_id_version("my-app", &engine)?;

Returns None if no transaction with that app_id has been committed.

Transaction retention

By default, SetTransaction actions are retained indefinitely. You can configure a retention duration via the delta.setTransactionRetentionDuration table property (e.g. "interval 30 days"). When set, transactions whose lastUpdated timestamp is older than the retention window are filtered out by get_app_id_version().

Note

Kernel automatically sets lastUpdated to the commit timestamp when you call with_transaction_id(). You don’t need to set it manually.

What’s next

• Appending data covers the basics of writing data with Kernel
• Checkpointing explains how to write checkpoints after commits

Altering a table

To add a column to an existing Delta table, you build an AlterTableTransaction from a Snapshot, queue one or more schema operations, and commit. The result is a metadata-only commit that updates the table’s schema without rewriting any data files.

Before reading this page, make sure you understand Creating a Table and Appending Data.

When to use alter table

Use alter_table() when you need to evolve a table’s schema in place. The common case today is adding a new column to a table that already has data, without rewriting the existing files. Existing rows read back NULL for the new column. Subsequent writes can populate it.

Schema evolution is a metadata-only change. The transaction emits an updated Metadata action and commits with data_change: false. No Add or Remove file actions are produced. This matters because readers can apply the new schema to existing files without scanning them, and writers that are only concerned with data changes can ignore the commit.

Note

The first supported operation is add_column(). Other schema operations (drop column, rename, type changes) are not yet available through the AlterTableTransaction API.

Adding a column

Suppose your table has the canonical schema name STRING, age INTEGER, city STRING with rows for Alice, Bob, and Carol, and you want to add a country column. The flow is:

1. Load a Snapshot of the table.
2. Call snapshot.alter_table() to get an AlterTableTransactionBuilder.
3. Call add_column() with the new field.
4. Call build() to produce an AlterTableTransaction.
5. Call commit() to atomically apply the schema change.

#![allow(unused)]
fn main() {
extern crate delta_kernel;
use delta_kernel::committer::FileSystemCommitter;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::schema::{DataType, StructField};
use delta_kernel::transaction::CommitResult;
use delta_kernel::{DeltaResult, Snapshot};
fn example() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
// 1. Load a snapshot of the existing table.
let snapshot = Snapshot::builder_for(url).build(&engine)?;

// 2. Build and commit an alter-table transaction that adds a new column.
let result = snapshot
    .alter_table()
    .add_column(StructField::nullable("country", DataType::STRING))
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .with_engine_info("my-app/1.0")
    .commit(&engine)?;

match result {
    CommitResult::CommittedTransaction(committed) => {
        println!("Schema evolved at version {}", committed.commit_version());
    }
    _ => eprintln!("alter table did not succeed"),
}
Ok(())
}
}

After this commit, the table schema has four fields. Existing rows for Alice, Bob, and Carol read back NULL for country. New writes can populate the column by including it in the RecordBatch they pass to engine.write_parquet().

Validation rules

add_column() checks the new field at build() time. If any rule is violated, build() returns an error and no commit is attempted.

Note

The column-mapping limitation applies to add-column only. If your table uses column mapping (delta.columnMapping.mode = "name" or "id"), you cannot currently add a column through alter_table(). This restriction is expected to be lifted as the alter-table framework grows.

Chaining multiple operations

add_column() can be called more than once to add several columns in a single commit. The operations are applied in order, and the resulting schema is validated as a whole before the commit is constructed:

let result = snapshot
    .alter_table()
    .add_column(StructField::nullable("country", DataType::STRING))
    .add_column(StructField::nullable("postal_code", DataType::STRING))
    .build(&engine, Box::new(FileSystemCommitter::new()))?
    .commit(&engine)?;

The builder uses a type-state pattern to enforce that at least one operation is queued before build() is callable. Calling .build() directly on snapshot.alter_table() without first calling add_column() is a compile error.

What you cannot do on an alter-table transaction

AlterTableTransaction is a metadata-only transaction. It does not implement SupportsDataFiles, so the data-file methods are not available at compile time. In particular, the following are not callable on an AlterTableTransaction:

If you need to add data and evolve the schema, run two transactions: an alter-table transaction first, then a write transaction against the post-commit snapshot. See Appending Data for the write flow.

What’s next

• Appending Data walks through writing data to the evolved table.
• Creating a Table covers creating a new table with the schema you want from the start.
• Schemas and Data Types describes Kernel’s type system in detail.

Checkpointing

To compact a Delta table’s transaction log into a single Parquet file, you write a checkpoint. A checkpoint allows readers to skip old JSON commit files and start from the checkpoint instead, which speeds up table discovery for tables with many commits.

When to checkpoint

Kernel does not checkpoint automatically. Your application decides when to trigger one. The table property delta.checkpointInterval controls the recommended frequency. You can read it via snapshot.table_properties().checkpoint_interval, which returns an Option<NonZero<u64>> representing the number of commits between checkpoints (e.g., 10 means checkpoint every 10 commits).

The simple path: Snapshot::checkpoint()

The easiest way to write a checkpoint is the convenience method on Snapshot:

extern crate delta_kernel;
use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::snapshot::CheckpointWriteResult;
use delta_kernel::{DeltaResult, Snapshot};
fn main() -> DeltaResult<()> {
let url = delta_kernel::try_parse_uri("/tmp/table")?;
let engine = DefaultEngine::builder(store_from_url(&url)?).build();
let snapshot = Snapshot::builder_for(url).build(&engine)?;
let (result, new_snapshot) = snapshot.checkpoint(&engine)?;
match result {
    CheckpointWriteResult::Written => println!("Checkpoint written"),
    CheckpointWriteResult::AlreadyExists => println!("Checkpoint already exists"),
}
Ok(())
}

checkpoint() takes a &SnapshotRef (i.e., &Arc<Snapshot>) and returns a DeltaResult<(CheckpointWriteResult, SnapshotRef)>. The returned SnapshotRef reflects the new checkpoint when Written, or the original snapshot when AlreadyExists.

This method handles everything in one call: reads the log, reconciles actions, writes the checkpoint Parquet file, and updates the _last_checkpoint hint file.

Note

checkpoint() calls ParquetHandler::write_parquet_file to write the checkpoint and StorageHandler::head to retrieve its metadata. The DefaultEngine implements both. If you use a custom engine, make sure it provides working implementations of both.

After a commit

The most common time to checkpoint is right after a successful commit. The post-commit snapshot gives you a checkpoint-ready SnapshotRef:

let committed = match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => c,
    _ => panic!("unexpected result"),
};

if let Some(snapshot) = committed.post_commit_snapshot() {
    let (_result, _new_snapshot) = snapshot.clone().checkpoint(&engine)?;
}

Warning

For catalog-managed tables, you must publish the commit before checkpointing. create_checkpoint_writer() returns an error if the snapshot is not published.

The custom path: CheckpointWriter

If you need control over how the checkpoint file is written, use the lower-level CheckpointWriter API. create_checkpoint_writer() consumes an Arc<Snapshot>:

// 1. Create a CheckpointWriter from a snapshot (consumes the Arc<Snapshot>)
let writer = snapshot.create_checkpoint_writer()?;

// 2. Get the path where the checkpoint should be written
let checkpoint_path = writer.checkpoint_path()?;

// 3. Get the checkpoint data as an ActionReconciliationIterator
let checkpoint_data = writer.checkpoint_data(&engine)?;
let state = checkpoint_data.state();

// 4. Write data to storage (engine-specific).
//    You must fully consume the iterator and write all data.
let lazy_data = checkpoint_data
    .map(|r| r.and_then(|f| f.apply_selection_vector()));
// ... write lazy_data to checkpoint_path in your own way ...

// 5. Get file metadata (size, modification time, etc.)
let file_meta = engine.storage_handler().head(&checkpoint_path)?;

// 6. Build LastCheckpointHintStats from the now-exhausted iterator state.
//    Use 0 for num_sidecars on V1 checkpoints or V2 checkpoints without sidecars.
let state = Arc::into_inner(state)
    .ok_or_else(|| Error::internal_error("checkpoint state Arc still has other references"))?;
let last_checkpoint_stats = LastCheckpointHintStats::from_reconciliation_state(
    state,
    file_meta.size,
    0, // num_sidecars
)?;

// 7. Finalize. Writes the _last_checkpoint hint file.
writer.finalize(&engine, &last_checkpoint_stats)?;

The key requirement is that the data iterator must be fully consumed before building the LastCheckpointHintStats. LastCheckpointHintStats::from_reconciliation_state returns an error if the iterator has not been exhausted. Because finalize() reads the action counts from the stats struct rather than from the iterator, you build the stats first and then pass them to finalize().

Checkpoint format

Kernel automatically selects the checkpoint format based on table features:

You do not need to manage this yourself. CheckpointWriter handles format selection and includes the appropriate actions.

Deciding when to checkpoint

After a successful commit, CommittedTransaction exposes a PostCommitStats struct through post_commit_stats(). You can use these stats to decide whether to trigger a checkpoint or other maintenance operations.

PostCommitStats has two fields:

The typical pattern is to check commits_since_checkpoint against the table’s delta.checkpointInterval property and checkpoint when the threshold is reached:

let committed = match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => c,
    _ => panic!("unexpected result"),
};

let stats = committed.post_commit_stats();
let interval = committed
    .post_commit_snapshot()
    .map(|s| {
        s.table_properties()
            .checkpoint_interval
            .map(|n| n.get())
            .unwrap_or(10)
    })
    .unwrap_or(10);

if stats.commits_since_checkpoint >= interval {
    if let Some(snapshot) = committed.post_commit_snapshot() {
        let (_result, _new_snapshot) = snapshot.clone().checkpoint(&engine)?;
    }
}

See Version Checksums for the recommended full post-commit pattern that includes writing a CRC file before checkpointing.

Log compaction

Log compaction aggregates multiple commit JSON files into a single compacted file, reducing the number of files Kernel must read during log replay. The API follows a similar pattern to the CheckpointWriter API: create a LogCompactionWriter from a Snapshot, retrieve the compaction path and data, then write the data to storage.

Note

Log compaction is currently disabled on both reads and writes due to insufficient integration test coverage. See issue #2337 for tracking. The Snapshot::log_compaction_writer() method exists but returns an error if called. This section will be updated when the feature is re-enabled.

What’s next

• Version Checksums for writing CRC files after every commit
• Appending Data to learn about writing data and committing
• Catalog-Managed Tables to understand publishing commits for catalog-managed tables

Version checksums

To speed up snapshot loading and enable table state validation, you write a version checksum (CRC file) after each commit. A CRC file records a compact summary of the table state at a given version. When a CRC file exists, Kernel can skip expensive log replay on the next snapshot load.

Version checksums are independent of checkpoints. Checkpoints compact the transaction log into Parquet. Checksums are small JSON files that validate and accelerate snapshot construction. You may write both after every commit.

Writing a checksum

Call write_checksum() on the post-commit snapshot after every successful commit. The method writes a small JSON CRC file into the _delta_log/ directory and returns a ChecksumWriteResult indicating whether the file was written or already existed.

use delta_kernel::snapshot::ChecksumWriteResult;

let committed = match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => c,
    _ => panic!("unexpected result"),
};

if let Some(snapshot) = committed.post_commit_snapshot() {
    let (result, new_snapshot) = snapshot.write_checksum(&engine)?;
    match result {
        ChecksumWriteResult::Written => println!("CRC file written"),
        ChecksumWriteResult::AlreadyExists => println!("CRC file already exists"),
    }
}

write_checksum() is safe to call unconditionally. If another writer already wrote the CRC file for this version, it returns ChecksumWriteResult::AlreadyExists without error. Per the Delta protocol, writers must not overwrite existing checksum files.

ChecksumWriteResult

The return type is DeltaResult<(ChecksumWriteResult, SnapshotRef)>:

Use the returned SnapshotRef for subsequent operations (checkpointing, publishing) so that downstream code sees the CRC file.

Note

write_checksum() currently requires a post-commit snapshot with pre-computed CRC information in memory. Calling it on a snapshot loaded from disk (without a pre-computed CRC) returns an Error::ChecksumWriteUnsupported error. In practice, this means you call write_checksum() on the snapshot returned by CommittedTransaction::post_commit_snapshot().

Recommended post-commit pattern

After every commit, write the checksum first, then decide whether to checkpoint:

let committed = match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(c) => c,
    _ => panic!("unexpected result"),
};

if let Some(snapshot) = committed.post_commit_snapshot() {
    // 1. Write the version checksum
    let (_, snapshot) = snapshot.write_checksum(&engine)?;

    // 2. Check whether it is time to checkpoint
    let stats = committed.post_commit_stats();
    let interval = snapshot
        .table_properties()
        .checkpoint_interval
        .map(|n| n.get())
        .unwrap_or(10);

    if stats.commits_since_checkpoint >= interval {
        let (_result, _new_snapshot) = snapshot.checkpoint(&engine)?;
    }
}

This pattern keeps your connector’s maintenance logic in one place: commit, write the CRC file, then conditionally checkpoint based on the table’s configured interval.

Reading file stats from a checksum

When Kernel loads a snapshot whose version has a CRC file, it parses the checksum’s table-level statistics and caches them on the snapshot. Call Snapshot::get_file_stats_if_loaded() to read them back:

if let Some(stats) = snapshot.get_file_stats_if_loaded() {
    println!("num files:        {}", stats.num_files);
    println!("table size bytes: {}", stats.table_size_bytes);
}

get_file_stats_if_loaded() returns None when no CRC file was loaded for this snapshot’s version. It never triggers a CRC read. It only returns stats that were already materialized during snapshot construction, so the call is cheap and always synchronous.

File size histogram

When the writer that produced the CRC file recorded a file size histogram, Kernel exposes it alongside the scalar totals. The histogram groups the snapshot’s live files into size bins:

if let Some(stats) = snapshot.get_file_stats_if_loaded() {
    if let Some(histogram) = &stats.file_size_histogram {
        for ((bin_start, count), bytes) in histogram
            .sorted_bin_boundaries
            .iter()
            .zip(&histogram.file_counts)
            .zip(&histogram.total_bytes)
        {
            println!(">= {bin_start:>12} B: {count} files, {bytes} bytes");
        }
    }
}

sorted_bin_boundaries gives the inclusive lower bound of each bin. The next boundary is the exclusive upper bound. file_counts and total_bytes are parallel arrays with the same length. The histogram is None when the writer did not populate it.

Use these stats for lightweight planning (cost estimation, compaction heuristics, monitoring) without replaying the log. When get_file_stats_if_loaded() returns None, either fall back to aggregating ScanFile.size and ScanFile.stats via visit_scan_files, or write a CRC file on the next commit so subsequent snapshots have stats available.

Note

file_size_histogram is an optional field in the Delta protocol. Kernel surfaces whatever the CRC file contains. A None value means the histogram was omitted, not that the table has no files.

What’s next

• Checkpointing for compacting the transaction log into Parquet
• Appending Data for the full write and commit flow

Building a connector: overview

A connector is the adapter layer between a compute engine (Spark, Flink, DuckDB, Polars, DataFusion, or your own query engine) and Delta tables. This matters because compute engines know nothing about the Delta protocol. They expose their own DataSource APIs, and the connector translates those calls into Delta Kernel operations.

The big picture

Every compute engine defines its own interfaces for reading and writing data. For example:

• Apache Spark has DataSourceV2: Table, ScanBuilder, Scan, Batch, PartitionReader, WriteBuilder, and others.
• Apache Flink has its Source/Sink APIs: Source, SplitEnumerator, SourceReader, Sink, SinkWriter, and others.
• DuckDB has its Extension API: TableFunction, TableFunctionBindInput, and others.

Building a Delta connector means implementing these compute-engine-specific interfaces and using Delta Kernel to fulfill them:

┌─────────────────────────────────────────────────────────┐
│                    Compute Engine                        │
│          (Spark, Flink, DuckDB, Polars, ...)            │
│                                                         │
│  "I need a Table, a Scan, a Writer..."                  │
└───────────────────────┬─────────────────────────────────┘
                        │ calls your DataSource API impl
                        ▼
┌─────────────────────────────────────────────────────────┐
│                 Your Delta Connector                     │
│                                                         │
│  Implements the compute engine's DataSource interfaces   │
│  (e.g. Spark's Table/ScanBuilder/Scan/Batch/Writer)     │
│                                                         │
│  Uses Kernel to fulfill those interfaces:               │
│    Snapshot  ->  table metadata, schema, version        │
│    Scan      ->  which files to read, data skipping     │
│    Transaction -> write files, commit atomically        │
└───────────────────────┬─────────────────────────────────┘
                        │ calls Kernel APIs
                        ▼
┌─────────────────────────────────────────────────────────┐
│                    Delta Kernel                          │
│                                                         │
│  Protocol logic, log replay, data skipping,             │
│  schema enforcement, transaction coordination           │
│                                                         │
│  Calls into Engine trait for I/O and compute            │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────┐
│              Engine trait implementation                  │
│                                                         │
│  DefaultEngine (Arrow + object_store)                   │
│  ...or your custom Engine with native I/O & formats     │
└─────────────────────────────────────────────────────────┘

Example: how the Spark connector works

To make this concrete, here’s how the Delta Spark connector maps Spark’s DataSourceV2 interfaces to Kernel APIs:

The pattern is always the same: the compute engine asks for something through its DataSource API, and the connector translates that into the corresponding Kernel call.

The Snapshot is the center of everything

All connector operations start from a Snapshot, an immutable view of the table at a specific version:

let snapshot = Snapshot::builder_for(table_url).build(&engine)?;

From a snapshot, you can:

What your connector does vs. what Kernel does

The key principle: Kernel handles the Delta protocol, your connector handles execution.

Kernel handles:

• Log replay (figuring out which files are active at a version)
• Data skipping (pruning files using statistics and predicates)
• Schema enforcement and column mapping
• Transaction conflict detection
• Protocol compliance (table features, reader/writer requirements)

Your connector handles:

• Implementing the compute engine’s DataSource API
• Controlling parallelism and distribution (how many threads/tasks, which worker reads which file)
• Data format conversion (Kernel’s EngineData to your engine’s native format)
• Query planning integration (pushing filters and projections down to Kernel)
• Resource management (memory budgets, connection pooling)

The Engine trait: pluggable I/O and compute

Kernel never does I/O directly. When it needs to list files, read Parquet, parse JSON, or evaluate expressions, it calls into the Engine trait. This is where you can plug in your engine’s native implementations for maximum performance.

Kernel provides a DefaultEngine (Arrow + object_store + Tokio). Many connectors start here and only replace specific handlers when they need better performance. See The Engine trait for details on the four required handlers (StorageHandler, JsonHandler, ParquetHandler, EvaluationHandler) and the optional MetricsReporter.

You need a custom Engine when:

• Your engine has its own columnar data format (not Arrow) and you want to avoid conversion overhead
• Your engine has its own I/O layer (e.g. a distributed file system client, encrypted storage, or custom caching)
• You want engine-native expression evaluation (e.g. vectorized execution, JIT compilation)

To use different cloud storage (S3, Azure, GCS), you do not need a custom engine. DefaultEngine supports all object_store backends without modification.

Getting started

Building a connector typically involves these steps:

1. Choose or implement an Engine. Start with DefaultEngine unless you have a reason not to. See Implementing the Engine trait.

2. Implement your DataSource’s read interfaces using Kernel’s Scan API:

   • Create a Snapshot to discover the table schema and version
   • Use ScanBuilder to push down filters and column projections
   • Use Scan to get the list of files to read
   • Distribute those files across your engine’s execution model
   • Read and transform data using the Engine
   • See Building a scan and Filter pushdown

3. Implement your DataSource’s write interfaces using Kernel’s Transaction API:

   • Create a Transaction from a Snapshot
   • Write Parquet files and register them with the transaction
   • Commit atomically, handling conflicts and retries
   • See Creating a table and Appending data

4. Handle distribution (if your engine is distributed):

   • Serialize scan metadata and transaction state to ship to workers
   • Collect results from workers and commit on the driver
   • See Advanced reads with scan_metadata()

What’s next

• Implementing the Engine trait: when and how to customize the Engine
• Building a scan: the Kernel read API
• Appending data: the Kernel write API

Implementing the Engine trait

The Engine trait is the main integration point between your connector and Delta Kernel. For background on what the Engine trait is and when you need a custom one, see the Connector Overview and The Engine Trait.

The Engine trait

The Engine trait has four required methods, each returning a handler:

pub trait Engine {
    fn evaluation_handler(&self) -> Arc<dyn EvaluationHandler>;
    fn storage_handler(&self) -> Arc<dyn StorageHandler>;
    fn json_handler(&self) -> Arc<dyn JsonHandler>;
    fn parquet_handler(&self) -> Arc<dyn ParquetHandler>;
}

You don’t have to implement all four handlers from scratch. A common approach is to start with DefaultEngine and selectively replace handlers. For example, you might provide a custom ParquetHandler that reads into your engine’s native columnar format while reusing the default handlers for everything else.

Many of the Engine handlers take or return EngineData. See EngineData for more information about this type.

StorageHandler

StorageHandler provides file system operations. The kernel calls this to list and read files (as bytes) from storage.

pub trait StorageHandler {
    fn list_from(&self, path: &Url)
        -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<FileMeta>>>>;

    fn read_files(&self, files: Vec<FileSlice>)
        -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<Bytes>>>>;

    fn copy_atomic(&self, src: &Url, dest: &Url) -> DeltaResult<()>;

    fn put(&self, path: &Url, data: Bytes, overwrite: bool) -> DeltaResult<()>;

    fn head(&self, path: &Url) -> DeltaResult<FileMeta>;
}

Key contracts

• list_from: Results must be sorted lexicographically by path. If the path ends with /, list all files in that directory. Otherwise, list files lexicographically greater than the given path in the same directory.

• copy_atomic: Must fail with Error::FileAlreadyExists if the destination exists. This is used for commit publishing in catalog-managed tables.

• put: Writes raw bytes to the given path. If overwrite is false and the file already exists, must fail with Error::FileAlreadyExists.

• head: Must return Error::FileNotFound if the file doesn’t exist.

• read_files: Each FileSlice is a (Url, Option<Range<u64>>). When the range is None, read the entire file.

Default implementation

The DefaultEngine uses object_store for storage, which supports local filesystem, S3, GCS, and Azure out of the box.

JsonHandler

JsonHandler reads and writes JSON. The kernel uses this for Delta log commits (_delta_log/*.json) and checkpoint sidecars.

pub trait JsonHandler {
    fn parse_json(
        &self,
        json_strings: Box<dyn EngineData>,
        output_schema: SchemaRef,
    ) -> DeltaResult<Box<dyn EngineData>>;

    fn read_json_files(
        &self,
        files: &[FileMeta],
        physical_schema: SchemaRef,
        predicate: Option<PredicateRef>,
    ) -> DeltaResult<FileDataReadResultIterator>;

    fn write_json_file(
        &self,
        path: &Url,
        data: Box<dyn Iterator<Item = DeltaResult<FilteredEngineData>> + Send + '_>,
        overwrite: bool,
    ) -> DeltaResult<()>;
}

Key contracts

• parse_json: Input is a single-column batch of strings (JSON objects). Output columns match the output_schema. Missing fields should produce nulls for nullable columns.

• read_json_files: Data must be returned in file order (same order as the files slice argument), and rows within a file must be in source order. The predicate is an optional hint. The engine may ignore it.

• write_json_file: Must write newline-delimited JSON (one JSON object per line). Null columns should be omitted from the output to save space. The write must be atomic. If overwrite is false and the file exists, fail with an error.

Default implementation

The DefaultEngine uses arrow_json for parsing and the object_store crate for I/O.

ParquetHandler

ParquetHandler reads and writes Parquet files. This is typically the most important handler to customize, since it’s on the critical path for data reading performance.

pub trait ParquetHandler {
    fn read_parquet_files(
        &self,
        files: &[FileMeta],
        physical_schema: SchemaRef,
        predicate: Option<PredicateRef>,
    ) -> DeltaResult<FileDataReadResultIterator>;

    fn write_parquet_file(
        &self,
        location: Url,
        data: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send>,
    ) -> DeltaResult<()>;

    fn read_parquet_footer(&self, file: &FileMeta) -> DeltaResult<ParquetFooter>;
}

Key contracts for read_parquet_files

Column resolution: When reading, the handler must resolve columns from the Parquet file to the physical_schema:

1. If a StructField in the schema has a field ID (via ColumnMetadataKey::ParquetFieldId metadata), match by field ID first.
2. Otherwise, fall back to matching by column name.
3. If no match is found: return nulls for nullable columns, or an error for non-nullable columns.

Column Ordering: Columns must be returned in the order specified in the physical_schema argument, which is not necessarily the order they may be specified in the parquet file itself.

Missing Columns: If a column is specified in the schema, and is nullable, the parquet reader must return a column of all nulls.

Ordering: Like JsonHandler, data must be returned in file order, and rows within a file must be in source order.

Metadata columns: The handler must support two virtual metadata columns that are not stored in the Parquet file but generated at read time:

Footer reading: read_parquet_footer reads only the Parquet metadata (no data). If the file has field IDs (column mapping), they must be preserved in the returned schema’s StructField metadata under the ParquetFieldId key.

Default implementation

The DefaultEngine uses the Apache Arrow Parquet reader/writer with support for column projection, predicate pushdown, metadata columns, and field-ID-based column matching.

EvaluationHandler

EvaluationHandler creates reusable evaluators for expressions and predicates. The kernel uses this for data skipping (evaluating predicates against file statistics) and for per-file transformations (partition value injection, row tracking).

pub trait EvaluationHandler {
    fn new_expression_evaluator(
        &self,
        input_schema: SchemaRef,
        expression: ExpressionRef,
        output_type: DataType,
    ) -> DeltaResult<Arc<dyn ExpressionEvaluator>>;

    fn new_predicate_evaluator(
        &self,
        input_schema: SchemaRef,
        predicate: PredicateRef,
    ) -> DeltaResult<Arc<dyn PredicateEvaluator>>;

    fn null_row(&self, output_schema: SchemaRef)
        -> DeltaResult<Box<dyn EngineData>>;

    fn create_many(
        &self,
        schema: SchemaRef,
        rows: &[&[Scalar]],
    ) -> DeltaResult<Box<dyn EngineData>>;
}

The returned evaluators are reusable objects. The kernel creates them once and calls evaluate() on multiple batches:

pub trait ExpressionEvaluator {
    fn evaluate(&self, batch: &dyn EngineData) -> DeltaResult<Box<dyn EngineData>>;
}

pub trait PredicateEvaluator {
    fn evaluate(&self, batch: &dyn EngineData) -> DeltaResult<Box<dyn EngineData>>;
}

Key contracts

• Expression evaluators produce one output row per input row. If output_type is a struct, its fields describe the output columns. Otherwise, the output is a single column.

• Predicate evaluators produce a single nullable boolean column. true means the row matches, false or null means it doesn’t.

• null_row creates a single-row EngineData with all null values. The kernel uses this internally for partition column construction.

• create_many creates a multi-row EngineData by applying the given schema to multiple rows of Scalar values. Each element in rows contains one scalar per top-level field in the schema. Returns an error if any row’s scalar count doesn’t match the schema’s field count, or if a scalar value’s type doesn’t match its corresponding field.

Default implementation

The DefaultEngine uses Arrow compute kernels for expression evaluation.

The EngineData trait

The EngineData trait is the kernel’s interface for columnar data. Any data that flows between the kernel and your connector is represented as EngineData.

If you use the DefaultEngine, you get ArrowEngineData, a wrapper around an arrow RecordBatch that implements the EngineData trait, and don’t need to implement this trait. This page is for connector builders who want to use a different columnar data format.

The trait

pub trait EngineData: AsAny {
    fn visit_rows(
        &self,
        column_names: &[ColumnName],
        visitor: &mut dyn RowVisitor,
    ) -> DeltaResult<()>;

    fn len(&self) -> usize;

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    fn append_columns(
        &self,
        schema: SchemaRef,
        columns: Vec<ArrayData>,
    ) -> DeltaResult<Box<dyn EngineData>>;

    fn apply_selection_vector(
        self: Box<Self>,
        selection_vector: Vec<bool>,
    ) -> DeltaResult<Box<dyn EngineData>>;

    fn has_field(&self, name: &ColumnName) -> bool;
}

The five required methods to implement are visit_rows, len, append_columns, apply_selection_vector, and has_field. The is_empty method has a default implementation that delegates to len.

has_field returns true if a field at the given (possibly nested) path exists in the data’s schema. For a top-level field named "foo", pass ColumnName::new(["foo"]). For nested fields, each non-leaf element of the path must be a struct field at that level.

visit_rows and the visitor pattern

Kernel uses a visitor pattern to access the actual data that’s inside an EngineData. This pattern means the connector can call into kernel with a reference to the data, which makes reasoning about data lifetimes simpler. In particular, engines don’t need to worry about keeping data alive past the invocation of the visitor.

visit_rows is the core data extraction method. The kernel never inspects your columns directly. Instead, it passes a RowVisitor that knows which columns it needs, and your implementation provides typed accessors (GetData) for those columns.

The flow:

Kernel                              Your EngineData
------                              ---------------
"I need columns [path, size]"
    visit_rows(column_names, visitor)
                                    1. Look up the requested columns
                                    2. Create a GetData accessor per column
                                    3. Call visitor.visit(row_count, &getters)

Kernel (inside the visitor)
    for row in 0..row_count:
        path = getters[0].get_str(row, "path")?
        size = getters[1].get_long(row, "size")?

GetData

GetData is the typed accessor trait. Each accessor handles one column. The kernel calls the appropriate typed method based on the column’s data type:

All methods return DeltaResult<Option<T>>. A None value means the field is null. By default, every method returns an “unexpected type” error, so you only need to implement the one that matches your column’s type.

ListItem provides access to a row’s list of strings. Call get(index) to retrieve a single element, or materialize() to collect all elements into a Vec<String>.

MapItem provides access to a row’s string-to-string map. Call get(key) to look up a value by key, or materialize() to collect all entries into a HashMap<String, String>. If a value is null, get returns None and materialize drops that entry.

Not every possible data-type is covered (i.e. no Map<Int, Int>). The trait only covers the data types the kernel needs to fuction, and no more.

TypedGetData

The TypedGetData trait provides a convenience wrapper which is automatically implemented for most types that implement GetData. Instead of calling the specific get_* method for each type, you can write generic code that dispatches based on the Rust type:

// Without TypedGetData: explicit method per type
let path: Option<&str> = getters[0].get_str(row, "path")?;
let size: Option<i64> = getters[1].get_long(row, "size")?;

// With TypedGetData: type-driven dispatch
let path: Option<String> = getters[0].get_opt(row, "path")?;
let size: Option<i64> = getters[1].get_opt(row, "size")?;

TypedGetData also provides a get method that returns DeltaResult<T> (without Option), returning an error if the value is null.

RowVisitor

Engines don’t need to implement RowVisitor. The kernel provides its own visitors. Your visit_rows implementation needs construct the correct getters array and then call visitor.visit(row_count, &getters) once.

The visitor declares which columns and types it expects via selected_column_names_and_types(), which returns (&'static [ColumnName], &'static [DataType]). Your implementation should validate that the requested columns exist and have compatible types before creating getters.

RowVisitor also provides a convenience method visit_rows_of(&mut self, data: &dyn EngineData). This calls data.visit_rows(...) with the column names from selected_column_names_and_types(), saving you from extracting them manually.

append_columns

The kernel calls append_columns to add new columns to your data. This happens when committing to the table, as the kernel needs to build up the commit log entries. For example, row-tracking information is added to the data that needs to be written into the delta log this way.

fn append_columns(
    &self,
    schema: SchemaRef,
    columns: Vec<ArrayData>,
) -> DeltaResult<Box<dyn EngineData>>;

• schema describes only the new columns being appended (not the full result schema)
• columns contains the data as ArrayData, the kernel’s generic columnar representation that you will need to convert to your engine’s format.
• Returns a new EngineData with the original columns plus the appended columns
• The row count of the new columns must match the existing data

apply_selection_vector

The kernel calls apply_selection_vector to filter rows. This is used for deletion vector support and other row-level filtering.

fn apply_selection_vector(
    self: Box<Self>,
    selection_vector: Vec<bool>,
) -> DeltaResult<Box<dyn EngineData>>;

• Within the selection_vector, true means keep the row, false means remove it
• If the selection vector is shorter than the data, remaining rows are kept
• This consumes the EngineData (self: Box<Self>), so you can implement this in-place if your format supports it

FilteredEngineData

The kernel pairs EngineData with a selection vector in FilteredEngineData. This is just a convenience type since the two appear together in many situations. For example, this is used in write paths (e.g., JsonHandler::write_json_file receives an iterator of FilteredEngineData).

Construct a FilteredEngineData with try_new, which validates that the selection vector is not longer than the data:

let filtered = FilteredEngineData::try_new(data, selection_vector)?;

You can also convert a Box<dyn EngineData> directly with .into(), which selects all rows (equivalent to an empty selection vector):

let filtered: FilteredEngineData = engine_data.into();

Key methods:

Warning

If you call into_parts(), you MUST keep the returned parts paired. That is, the Box<dyn EngineData> is not valid for every row, only for those rows as desribed below.

In the selection vector, true means keep the row, false meands remove it. If the selection vector is shorter than the data, uncovered rows are implicitly selected. If the selection vector is longer than the data, try_new returns an error.

FilteredRowVisitor

The FilteredRowVisitor trait processes FilteredEngineData with automatic row filtering. Instead of receiving a raw row count, your visitor gets a RowIndexIterator that yields only the indices of selected rows:

pub trait FilteredRowVisitor {
    fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]);

    fn visit_filtered<'a>(
        &mut self,
        getters: &[&'a dyn GetData<'a>],
        rows: RowIndexIterator<'_>,
    ) -> DeltaResult<()>;

    fn visit_rows_of(&mut self, data: &FilteredEngineData) -> DeltaResult<()>;
}

The default visit_rows_of method handles all the plumbing: extracting the selection vector, building the bridge to RowVisitor, and calling visit_rows. Your implementation only needs to provide selected_column_names_and_types and visit_filtered.

Call rows.num_rows() inside visit_filtered to get the total row count (including deselected rows), which is useful for sizing output vectors.

ArrowEngineData (the default)

The DefaultEngine uses ArrowEngineData, which wraps an Arrow RecordBatch:

use delta_kernel::engine::arrow_data::ArrowEngineData;

// Wrap a RecordBatch
let engine_data = ArrowEngineData::new(record_batch);

// Access the underlying RecordBatch by reference
let batch_ref = engine_data.record_batch();

// Convert back from a Box<dyn EngineData> (requires the arrow-related default features)
use delta_kernel::engine::arrow_data::EngineDataArrowExt;
let batch = engine_data_box.try_into_record_batch()?;

ArrowEngineData also provides try_from_engine_data(engine_data) to downcast a Box<dyn EngineData> to Box<ArrowEngineData>, returning an error if the underlying type is not ArrowEngineData.

If you use the default engine, you work with ArrowEngineData and never need to implement EngineData yourself. EngineDataArrowExt provides try_into_record_batch() for converting the opaque EngineData trait object back to an Arrow RecordBatch for your connector’s use. This trait is implemented for both Box<dyn EngineData> and DeltaResult<Box<dyn EngineData>>, so you can call it directly on scan results.

What’s next

• Implementing the Engine trait covers the four handler traits your engine must provide
• Building a Scan shows how scan results flow through EngineData

Catalog-managed tables

A catalog-managed table is a Delta table whose commits go through a catalog instead of being written directly to the filesystem. This matters because it shifts the source of truth from the filesystem to the catalog, enabling centralized governance, enforceable constraints, and multi-table coordination.

Before reading this page, make sure you understand Architecture overview and The Engine trait.

Filesystem-managed vs. catalog-managed

By default, Delta tables are filesystem-managed: every commit is written directly to _delta_log/ as a JSON file, and readers discover table state by listing that directory. Atomicity comes from the filesystem’s PUT-if-absent semantics.

A catalog-managed table changes this model. The catalogManaged reader-writer table feature makes the catalog the source of truth for commits:

• Writers must commit through the catalog, not directly to the filesystem.
• Readers must contact the catalog to discover recent (possibly unpublished) commits.
• The catalog decides whether a commit attempt succeeds, not the filesystem.
• Path-based access is not supported. Tables must be accessed through the catalog.

Benefits of catalog-managed tables

In filesystem-managed tables, the filesystem is the ultimate authority on table state. Catalog-managed tables change that by putting the catalog on the critical path for all reads and writes:

• Governance across engines. Every read and write goes through the catalog, ensuring all engines enforce the same permissions, constraints, lineage, and auditing.
• Enforceable constraints. The catalog can reject invalid schema or constraint changes (for example, blocking a NOT NULL drop on a column referenced by a foreign key).
• Multi-table atomic updates. The catalog can coordinate commits spanning multiple tables without custom coordination services.
• Faster query planning. The catalog can serve table metadata directly, skipping cloud storage LIST operations that add 100+ ms of latency.

Terminology: commit types

Catalog-managed tables have several kinds of commits:

Writing a staged commit file does not reserve a version. Multiple writers may stage different <version>.<uuid>.json files for the same version; the catalog decides which one wins. Staged commits that lose the race become orphans on disk and are cleaned up out of band (typically by a catalog-driven sweep).

Note

The Delta protocol also defines inline commits, where commit content is sent directly to the catalog rather than written to disk. Kernel does not yet support inline commits.

Protocol requirements

The catalogManaged reader-writer table feature requires inCommitTimestamp to be enabled. A spec-compliant catalog-managed table therefore carries:

• catalogManaged in both readerFeatures and writerFeatures in the Protocol action (it is a reader-writer feature).
• inCommitTimestamp enabled as a writer-only feature.

Kernel constructs this protocol automatically during the create_table call; you do not hand-craft the Protocol or Metadata actions. The list above describes what ends up in the commit, not steps you write by hand.

The Delta spec also requires every CommitInfo action on a catalog-managed table to carry a unique txnId field. Kernel writes a fresh txnId (UUID) on every CommitInfo it generates, so this requirement is met automatically. Custom committers must write the full iterator of actions Kernel supplies to the staged commit file. They do not pick and choose which actions to include, and they do not synthesize replacement CommitInfo actions.

How Kernel fits in

Kernel’s design philosophy for catalog-managed tables is straightforward: Kernel doesn’t know about catalogs. It doesn’t know about table names, table IDs, catalog APIs, or catalog servers. Instead:

• The catalog client (a client-side library you write or import) works together with your connector to resolve table names to paths, fetch credentials, and retrieve recent ratified commits from the catalog.
• The catalog client translates the catalog’s response into a Vec<LogPath> log tail and a max catalog version. Your connector passes these to SnapshotBuilder to load the table.
• For writes, the catalog client provides a Committer that knows how to stage, ratify, and publish commits through the catalog.
• Kernel does its job (protocol compliance, log replay, data skipping, schema enforcement) without knowing or caring whether a catalog exists.

+-----------------------------------------------------------+
|                   Compute Engine                          |
|         (Spark, Flink, DuckDB, Polars, ...)               |
+---------------------------+-------------------------------+
                            | table name
                            v
+-----------------------------------------------------------+
|                   Catalog Client                          |
|                                                           |
|  1. Resolves table name -> path + credentials             |
|  2. Calls catalog API to get ratified commits             |
|  3. Translates commits into LogPath entries (log tail)    |
|  4. Provides a Committer for writing                      |
+---------------------------+-------------------------------+
                            | path, log tail, committer
                            v
+-----------------------------------------------------------+
|               Your Delta Connector                        |
|                                                           |
|  Uses Kernel APIs to read and write:                      |
|    Snapshot::builder_for(path)                            |
|      .with_log_tail(commits)                              |
|      .with_max_catalog_version(version)                   |
|      .build(&engine)                                      |
|    snapshot.transaction(committer, &engine)                |
+---------------------------+-------------------------------+
                            | calls Kernel APIs
                            v
+-----------------------------------------------------------+
|                   Delta Kernel                            |
|                                                           |
|  Knows nothing about catalogs, table names, or IDs.       |
|  Sees: a path, log files, and a Committer trait.          |
|  Handles: log replay, data skipping, protocol compliance  |
+-----------------------------------------------------------+

Key Kernel APIs for catalog-managed tables

Four Kernel APIs form the integration surface between the catalog client and Kernel:

• SnapshotBuilder::with_log_tail(Vec<LogPath>) accepts a contiguous run of commits from version M to version N inclusive (published or staged). The sequence must be ascending, gap-free, and duplicate-free. See Reading catalog-managed tables for the exact rules, including what M and N must equal with and without time travel.

• SnapshotBuilder::with_max_catalog_version(Version) caps the snapshot version at the catalog’s latest ratified version. Kernel requires this to be set for every catalog-managed snapshot, so readers come with an explicit catalog-ratified upper bound rather than silently reading whatever happens to be on the filesystem.

• Committer trait defines how transactions are committed. A catalog committer implements commit() to stage and ratify commits through the catalog API, and publish() to copy ratified commits to the Delta log.

• Snapshot::publish() publishes all unpublished catalog commits at the current snapshot version. Published commits become visible to filesystem-based readers and enable maintenance operations like checkpointing. If the snapshot has no unpublished commits, publish() is a no-op and returns the same snapshot reference without invoking the committer. If unpublished commits exist, both the snapshot and the committer must be catalog-managed; otherwise publish() errors.

Example: the read flow

The following pseudocode shows how a catalog client typically integrates with Kernel for reading. The exact API calls depend on your catalog.

// 1. Resolve table name through catalog API
//    catalog_client.get_table("catalog.schema.table") -> table_id, path, credentials

// 2. Get ratified commits from catalog
//    catalog_client.get_commits(table_id) -> commits, max_version

// 3. Convert catalog response to LogPath entries. If your catalog returns
//    (version, uuid, ...) tuples, format the filename as {version:020}.{uuid}.json
//    first. See "Building a log tail" in reading.md.
//    let log_tail: Vec<LogPath> = commits.into_iter()
//        .map(|c| LogPath::staged_commit(table_root.clone(), &c.filename, c.timestamp, c.size))
//        .collect::<Result<_, _>>()?;

// 4. Build Engine with vended credentials
//    let engine = build_engine_with_credentials(path, credentials);

// 5. Build Snapshot with log tail from catalog
//    let snapshot = Snapshot::builder_for(path)
//        .with_log_tail(log_tail)
//        .with_max_catalog_version(max_version)
//        .build(&engine)?;

// 6. Read the table using standard Kernel APIs
//    let scan = snapshot.scan_builder().build()?;
//    for data in scan.execute(Arc::new(engine))? { ... }

The key insight: the catalog client handles all catalog-specific logic (authentication, API calls, response parsing). By the time Kernel sees the request, it is a standard Snapshot::builder_for() call with an extra log tail and version cap.

Maintenance operations

Catalog-managed tables restrict which maintenance operations a client may run. Two layers enforce the rules:

• Kernel enforces that checkpoints and log compaction run only on published versions. Publish commits before running either. Version checksum (CRC) files, by contrast, can be written for unpublished versions as well.
• The managing catalog enforces which maintenance operations a client may request.

See also

• Unity Catalog integration for a concrete implementation using Unity Catalog.

What’s next

• Implementing a catalog committer describes how to implement the Committer trait for your catalog.
• Reading catalog-managed tables covers how to provide catalog commits to Kernel for snapshot construction.
• Writing: the commit and publish flow walks through the full commit, ratify, and publish lifecycle.

Implementing a catalog committer

To commit transactions on catalog-managed tables, you implement the Committer trait with your catalog’s staging and ratification logic. For filesystem-managed tables, the built-in FileSystemCommitter writes delta files directly via PUT-if-absent. For catalog-managed tables, you provide your own Committer that routes commits through your catalog.

Warning

Kernel rejects FileSystemCommitter on a catalog-managed table at txn.commit() time. You must provide a catalog committer before commit runs.

Before reading this page, make sure you understand Catalog-managed tables.

The Committer trait

pub trait Committer: Send {
    fn commit(
        &self,
        engine: &dyn Engine,
        actions: Box<dyn Iterator<Item = DeltaResult<FilteredEngineData>> + Send + '_>,
        commit_metadata: CommitMetadata,
    ) -> DeltaResult<CommitResponse>;

    fn is_catalog_committer(&self) -> bool;

    fn publish(
        &self,
        engine: &dyn Engine,
        publish_metadata: PublishMetadata,
    ) -> DeltaResult<()>;
}

The trait has three methods. Two (commit() and publish()) carry the real logic; the third (is_catalog_committer()) is a one-line method that returns a constant:

1. commit() atomically commits the given actions at the version specified in CommitMetadata. Returns CommitResponse::Committed on success or CommitResponse::Conflict { version } if another writer already committed this version.

2. is_catalog_committer() returns true for catalog committers. Kernel checks this flag on both commit and publish paths and enforces the pairing in both directions: it rejects a FileSystemCommitter on a catalog-managed table, and it rejects a catalog committer on a filesystem-managed table.

3. publish() copies ratified catalog commits from _staged_commits/ to the main _delta_log/ directory as published delta files. Some catalogs publish server-side, in which case publish() only notifies the catalog; others use PUT-if-absent copies from the client.

CommitMetadata

Kernel constructs CommitMetadata and passes it to your commit() method. Key methods:

impl CommitMetadata {
    pub fn published_commit_path(&self) -> DeltaResult<Url>;
    pub fn staged_commit_path(&self) -> DeltaResult<Url>;  // unique UUID each call
    pub fn version(&self) -> Version;
    pub fn commit_type(&self) -> CommitType;
    pub fn in_commit_timestamp(&self) -> i64;
    pub fn max_published_version(&self) -> Option<Version>;
    pub fn table_root(&self) -> &Url;
}

This block is a subset of CommitMetadata’s public surface. See the rustdoc for the full list.

• published_commit_path() returns the final delta log path (e.g., s3://bucket/table/_delta_log/00000000000000000001.json).
• staged_commit_path() returns a unique staged commit path (e.g., s3://bucket/table/_delta_log/_staged_commits/00000000000000000001.<uuid>.json). Kernel generates a fresh UUID on every call; do not substitute a catalog-supplied UUID.
• commit_type() returns a CommitType variant indicating whether this is a table creation or a write, and whether the table is catalog-managed.
• in_commit_timestamp() returns the timestamp (milliseconds since the Unix epoch, UTC) Kernel will record in the CommitInfo action as inCommitTimestamp. Kernel also writes a fresh txnId (UUID) on every CommitInfo it generates; the Delta spec requires txnId on catalog-managed commits, and Kernel emits it unconditionally.
• max_published_version() returns the highest version already published to the delta log, if any. Your catalog may need this to determine which commits still require publishing.

Warning

Call staged_commit_path() exactly once per commit() invocation. Because each call produces a different UUID, a second call yields a different path: only one path gets written, and if you use the other to tell the catalog, the catalog points at a missing file.

CommitResponse

pub enum CommitResponse {
    Committed { file_meta: FileMeta },
    Conflict { version: Version },
}

Return Committed with the FileMeta of the staged commit file on success. Return Conflict with the attempted version if another writer already committed that version.

Implementing a catalog committer

The typical implementation follows four steps. Steps 1 and 2 form the body of commit(). Step 3 is the is_catalog_committer() flag. Step 4 is publish().

Step 1: Stage the commit

Write the actions to a staged commit file in _staged_commits/:

fn commit(
    &self,
    engine: &dyn Engine,
    actions: Box<dyn Iterator<Item = DeltaResult<FilteredEngineData>> + Send + '_>,
    commit_metadata: CommitMetadata,
) -> DeltaResult<CommitResponse> {
    // Write actions to _staged_commits/<version>.<uuid>.json. `actions` is
    // already a Box<dyn Iterator<...>>, so pass it directly (do not re-box).
    let staged_path = commit_metadata.staged_commit_path()?;
    engine.json_handler().write_json_file(&staged_path, actions, false)?;
    // ...

Step 2: Ratify through the catalog

Call your catalog’s commit API to ratify the staged commit. The exact arguments vary by catalog; Unity Catalog’s CommitRequest, for example, carries the table id, commit version, staged filename, in-commit timestamp, and the maximum published version. Your catalog’s API may look different. Here is the general shape:

    // Tell the catalog about the staged commit.
    // Replace this with your catalog's ratification API. Forward the
    // commit_metadata.in_commit_timestamp() value so the catalog records the
    // same timestamp Kernel writes into the CommitInfo action.
    self.catalog_client.ratify_commit(
        &self.table_id,
        commit_metadata.version(),
        &staged_path,
        commit_metadata.in_commit_timestamp(),
        commit_metadata.max_published_version(),
    )?;

    // Return the staged file metadata on success. HEAD the file via
    // engine.storage_handler().head() to get the real byte size, and use
    // the in-commit timestamp as the logical commit time (not the filesystem
    // mtime, which reflects when the file was written rather than when the
    // commit took effect).
    let staged_file = engine.storage_handler().head(&staged_path)?;
    Ok(CommitResponse::Committed {
        file_meta: FileMeta::new(
            staged_path,
            commit_metadata.in_commit_timestamp(),
            staged_file.size,
        ),
    })
}

Map your catalog’s “another writer won this version” error to CommitResponse::Conflict { version: commit_metadata.version() } rather than propagating it as an Err. Return other errors as Err(...). Kernel classifies only Error::IOError as retryable (surfaced as CommitResult::RetryableTransaction); return IOError for transient storage failures and other variants for everything else. Do not disguise non-I/O errors as IOError to opt into retry semantics.

Step 3: Mark as catalog committer

fn is_catalog_committer(&self) -> bool {
    true
}

Step 4: Implement publish

Publishing copies staged commits from _staged_commits/ to the main _delta_log/. Kernel passes the commits to publish as a contiguous ascending batch via PublishMetadata::commits_to_publish(). Your implementation must:

• Copy the commits in the order Kernel provides (version v-1 before version v). Do not reorder.
• Be idempotent. Treat FileAlreadyExists as success; a previous publish may have already copied some entries.

use delta_kernel::Error;

fn publish(
    &self,
    engine: &dyn Engine,
    publish_metadata: PublishMetadata,
) -> DeltaResult<()> {
    for catalog_commit in publish_metadata.commits_to_publish() {
        let src = catalog_commit.location();            // _staged_commits/<v>.<uuid>.json
        let dest = catalog_commit.published_location(); // _delta_log/<v>.json
        match engine.storage_handler().copy_atomic(src, dest) {
            Ok(()) | Err(Error::FileAlreadyExists(_)) => (), // already published
            Err(e) => return Err(e),
        }
    }
    Ok(())
}

Putting it all together

Here is the full skeleton for a catalog committer. Replace MyCatalogClient with your catalog’s client type and fill in the ratification logic:

// Imports elided for brevity. In addition to the ones below, you will need
// Committer, CommitMetadata, CommitResponse, PublishMetadata, DeltaResult,
// FilteredEngineData, and Engine from delta_kernel.
use delta_kernel::{Error, FileMeta};

pub struct MyCatalogCommitter {
    catalog_client: Arc<MyCatalogClient>,
    table_id: String,
}

impl Committer for MyCatalogCommitter {
    fn commit(
        &self,
        engine: &dyn Engine,
        actions: Box<dyn Iterator<Item = DeltaResult<FilteredEngineData>> + Send + '_>,
        commit_metadata: CommitMetadata,
    ) -> DeltaResult<CommitResponse> {
        // 1. Stage: write actions to _staged_commits/
        let staged_path = commit_metadata.staged_commit_path()?;
        engine.json_handler().write_json_file(&staged_path, actions, false)?;

        // 2. Ratify: register the staged commit with the catalog. ratify_commit
        //    is an imagined example API; your catalog's signature will differ.
        self.catalog_client.ratify_commit(
            &self.table_id,
            commit_metadata.version(),
            &staged_path,
            commit_metadata.in_commit_timestamp(),
            commit_metadata.max_published_version(),
        )?;

        // 3. Return success. HEAD the staged file to get the real byte size,
        //    and use the in-commit timestamp as the logical commit time (not
        //    the filesystem mtime).
        let staged_file = engine.storage_handler().head(&staged_path)?;
        Ok(CommitResponse::Committed {
            file_meta: FileMeta::new(
                staged_path,
                commit_metadata.in_commit_timestamp(),
                staged_file.size,
            ),
        })
    }

    fn is_catalog_committer(&self) -> bool {
        true
    }

    fn publish(
        &self,
        engine: &dyn Engine,
        publish_metadata: PublishMetadata,
    ) -> DeltaResult<()> {
        for catalog_commit in publish_metadata.commits_to_publish() {
            let src = catalog_commit.location();
            let dest = catalog_commit.published_location();
            match engine.storage_handler().copy_atomic(src, dest) {
                Ok(()) | Err(Error::FileAlreadyExists(_)) => (),
                Err(e) => return Err(e),
            }
        }
        Ok(())
    }
}

For a complete Unity Catalog implementation, see Unity Catalog integration.

What’s next

• Reading catalog-managed tables covers how the catalog client provides commits for snapshot construction.
• Writing: the commit and publish flow walks through the complete write lifecycle including publishing.

Reading catalog-managed tables

To read a catalog-managed table, you provide Kernel with a log tail: the set of recent ratified commits your catalog client knows about. Kernel merges these with whatever it finds on the filesystem to build a complete Snapshot.

Before reading this page, make sure you understand Catalog-managed tables.

Table resolution is outside Kernel

Kernel doesn’t know about table names, table IDs, or catalog APIs. Before you can use any Kernel API, the catalog client must:

1. Resolve the table name to a storage path (e.g. unity.my_schema.my_table -> s3://bucket/path/to/table/)
2. Fetch storage credentials (e.g. temporary AWS credentials for the table’s S3 location)
3. Get the latest ratified commits from the catalog (the log tail)

Each catalog has its own APIs for these steps.

Why a log tail?

A catalog-managed table can have a mix of published commits (already in _delta_log/) and unpublished ratified commits (in _staged_commits/ or inline in the catalog). Kernel lists _delta_log/ to find published commits and checkpoints, but it cannot discover unpublished commits on its own. The log tail bridges this gap.

For example, suppose versions 0-7 are published and versions 8-9 are staged:

_delta_log/
  00000000000000000000.json          (published)
  ...
  00000000000000000007.json          (published)
  _staged_commits/
    00000000000000000008.<uuid>.json  (ratified, not published)
    00000000000000000009.<uuid>.json  (ratified, not published)

Without a log tail, Kernel would only see versions 0-7. The log tail tells Kernel about versions 8 and 9.

Building a log tail

The catalog client calls the catalog’s commits API to get the latest ratified commits, then translates each one into a LogPath:

use delta_kernel::LogPath;

let log_path = LogPath::staged_commit(
    table_url.clone(),                              // table root URL (must end with '/')
    "00000000000000000008.<uuid>.json",             // staged filename: <version>.<uuid>.json
    1234567890,                                     // last modified timestamp
    4096,                                           // file size in bytes
)?;

The filename follows the Delta staged-commit convention: a zero-padded 20-digit version, the UUID assigned when the commit was staged, and a .json extension. Kernel mints the UUID inside CommitMetadata::staged_commit_path() at write time; the writer passes the resulting path to the catalog, and the catalog returns that same path on reads. If your catalog returns commits as (version, uuid, ...) tuples, format the filename as {version:020}.{uuid}.json before calling LogPath::staged_commit. Published commit files use {version:020}.json and are loaded via LogPath::try_new, not LogPath::staged_commit. The table_url argument must end with /, otherwise LogPath::staged_commit returns an error.

Then pass the resulting Vec<LogPath> to SnapshotBuilder along with the maximum catalog-ratified version:

use delta_kernel::Snapshot;

let snapshot = Snapshot::builder_for(table_url)
    .with_max_catalog_version(latest_version) // the latest version the catalog ratified
    .with_log_tail(log_paths)                 // the commits the catalog returned
    .build(&engine)?;

The log tail must be a contiguous sequence M..=N: ascending by version, no gaps, no duplicates. The first version M can be any value >= 0 (typical catalog clients return a suffix of recent commits, not the full history). The last version N depends on whether you are time-traveling:

• Without time travel, N must equal max_catalog_version.
• With .at_version(v), N must be >= v, and max_catalog_version must still be set. Both constraints apply simultaneously.

The log tail can overlap with published commits already in _delta_log/. For any version present in both, Kernel uses the log-tail entry, as the Delta spec requires readers to prefer the catalog-supplied commit when the catalog provides one.

Call .with_max_catalog_version(N) to tell Kernel the highest version the catalog has ratified. This prevents Kernel from loading filesystem commits beyond what the catalog knows about.

Note

To time-travel to a specific version within a catalog-managed table, call both .at_version(v) and .with_max_catalog_version(N) on the builder. Call order does not matter. The requested version must not exceed N.

For a complete Unity Catalog example, see Reading Unity Catalog tables.

After building the snapshot, reading works the same as for a filesystem-managed table: build a scan, apply predicates, and read data. See Building a scan.

What’s next

• Writing: the commit and publish flow covers how writes work for catalog-managed tables.
• Implementing a catalog committer describes the Committer trait in detail.

Writing: the commit and publish flow

To write to a catalog-managed table, you follow the same transaction pattern as a filesystem-managed table. The difference is that your catalog Committer stages, ratifies, and publishes commits through the catalog instead of writing directly to the filesystem.

Before reading this page, make sure you understand Catalog-managed tables.

The write lifecycle

A catalog-managed write has four phases:

1. LOAD SNAPSHOT
   Catalog client resolves the table and fetches recent commits.
   Build snapshot with Snapshot::builder_for(path).with_log_tail(commits).build()

2. COMMIT
   Transaction generates actions. Committer stages them to _staged_commits/.
   Committer calls the catalog API to ratify the staged commit.

3. HANDLE RESULT
   Committed: get post-commit snapshot, proceed to publish.
   Conflict: rebase and retry.
   Retryable: transient I/O error, retry.

4. PUBLISH
   Copy staged commits from _staged_commits/ to _delta_log/ as published delta files.
   Returns a new snapshot with all commits marked as published.

Phase 1: Load a snapshot

Your catalog client resolves the table, fetches credentials and recent commits, then builds a snapshot. See Reading catalog-managed tables for the full details.

// Catalog client resolves table name -> path + credentials + commits
let (path, credentials, commits, latest_version) = catalog_client.load_table(table_name)?;

// Build snapshot with log tail and cap at the catalog-ratified version
let snapshot = Snapshot::builder_for(path)
    .with_max_catalog_version(latest_version)
    .with_log_tail(commits)
    .build(&engine)?;

Phase 2: Create a transaction and commit

To begin a write, create a transaction with your catalog’s Committer, add files, and call commit():

// transaction() moves the Box<dyn Committer> into the Transaction, and commit()
// consumes the Transaction, so the boxed committer is gone by the time you need
// to publish. Construct a second committer for publish() in Phase 3 and clone
// any catalog-client state you need to keep in scope across both calls.
let committer = Box::new(MyCatalogCommitter::new(
    catalog_client.clone(),
    table_id.clone(),
));
let mut txn = snapshot
    .transaction(committer, &engine)?
    .with_operation("INSERT".to_string());

// Drive your Parquet writer from the write context, then hand the resulting
// add-file metadata batch to the transaction. See the
// [Appending data](../writing/append.md) how-to for the full Parquet-writing flow.
let write_context = txn.unpartitioned_write_context()?;
// ... use write_context to produce add_metadata: Box<dyn EngineData>
//     matching txn.add_files_schema() ...
txn.add_files(add_metadata);

// Commit the transaction. Kernel invokes committer.commit() internally; Phase 3
// handles the result.
let commit_result = txn.commit(&engine)?;

The ? on txn.commit(&engine)? only propagates non-recoverable errors. Successful commits, conflicts, and retryable I/O errors arrive as the three variants of CommitResult in the match below. Everything else (auth errors, catalog protocol errors, etc.) bubbles out directly.

When txn.commit() runs, Kernel:

1. Assembles the actions: a leading CommitInfo, any Protocol/Metadata updates this transaction makes (absent on steady-state appends), and Add/Remove file actions.
2. Verifies that a catalog committer is being used. Kernel rejects FileSystemCommitter for catalog-managed tables.
3. Calls committer.commit(engine, actions, commit_metadata).

Your committer then:

1. Writes the actions to _staged_commits/<version>.<uuid>.json.
2. Calls the catalog API to ratify the staged commit.
3. Returns CommitResponse::Committed or CommitResponse::Conflict.

Phase 3: Handle the result

On success, CommittedTransaction provides the commit version and a post-commit snapshot that reflects the newly committed state:

use delta_kernel::transaction::CommitResult;

match commit_result {
    CommitResult::CommittedTransaction(committed) => {
        let version = committed.commit_version();
        // post_commit_snapshot() returns an Option. For catalog-managed
        // commits today, Kernel returns Some. The Option exists for
        // incremental-development paths (e.g., table creation). Treat a
        // None as an error rather than silently skipping publish, so the
        // problem surfaces loudly if the invariant ever changes.
        let post_commit = committed
            .post_commit_snapshot()
            .ok_or_else(|| Error::generic("missing post-commit snapshot"))?;

        // commit() consumed the Box<dyn Committer> from Phase 2. publish() only
        // needs &dyn Committer, so construct a fresh instance here. This moves
        // catalog_client and table_id; clone them if you need them for a retry
        // loop around the whole write.
        let publish_committer =
            MyCatalogCommitter::new(catalog_client, table_id);

        // Proceed to publish (Phase 4).
        let published_snapshot = post_commit.publish(&engine, &publish_committer)?;
    }
    CommitResult::ConflictedTransaction(conflicted) => {
        // Another writer already committed at this version.
        // `conflicted.conflict_version()` returns the version this transaction
        // attempted. Rebase onto the new table state and retry.
    }
    CommitResult::RetryableTransaction(retryable) => {
        // Transient I/O error. `retryable.error` gives the underlying cause;
        // `retryable.transaction` is the original transaction you can retry
        // without rebasing. Kernel reaches this arm only for `Error::IOError`
        // variants; return other error kinds as-is rather than disguising
        // them as IOError to force retry.
    }
}

Phase 4: Publish

Call Snapshot::publish() on the post-commit snapshot (shown in Phase 3 above) to make ratified commits visible as normal delta files and to unlock maintenance operations. Snapshot::publish():

1. Finds all unpublished catalog commits in the snapshot’s log segment.
2. Validates that the table is catalog-managed and the committer is a catalog committer.
3. Builds a PublishMetadata containing the list of CatalogCommit entries to publish.
4. Delegates to committer.publish(engine, publish_metadata), which copies each staged commit from _staged_commits/ to _delta_log/<version>.json.
5. Returns a new snapshot where all commits are marked as published.

Publishing matters because:

• It makes commits visible to filesystem-based readers.
• It enables maintenance operations such as checkpointing, which can only operate on published versions.
• It reduces the number of commits the catalog needs to store and serve.

Maintenance operations

Once commits are published, you can perform maintenance operations on the published snapshot:

// Checkpoint the published snapshot. checkpoint() returns
// (CheckpointWriteResult, SnapshotRef): the first element is an enum reporting
// whether a checkpoint was written or one already existed, and the second is
// the snapshot with the updated log segment.
let (_checkpoint_result, _post_checkpoint_snapshot) =
    published_snapshot.checkpoint(&engine)?;

Note

Maintenance operations on catalog-managed tables are subject to extra rules: Kernel requires published versions, and the managing catalog controls which operations a client may run.

For a complete Unity Catalog example, see Writing to Unity Catalog tables.

What’s next

• Implementing a catalog committer: how to build a Committer for your catalog.
• Reading catalog-managed tables: how to provide catalog commits for snapshot construction.
• Appending data: the basics of writing data with Kernel.

Unity Catalog integration

Unity Catalog (UC) integration is a set of crates that connect Delta Kernel to Unity Catalog, a multi-engine governance layer for data and AI assets. This matters because UC-managed Delta tables require all reads and writes to go through the catalog, and these crates handle that coordination so your connector doesn’t have to implement the UC REST protocol from scratch.

Before reading this page, make sure you understand the general concepts in Catalog-Managed Tables: Overview.

What Unity Catalog provides

Unity Catalog acts as the source of truth for catalog-managed tables. When a table has the catalogManaged table feature enabled, your connector can no longer read or write it by accessing the transaction log on disk alone. Instead, the connector must:

1. Resolve the table name to a storage path and table ID via the UC API.
2. Obtain credentials from UC to access the table’s cloud storage.
3. Fetch recent commits from UC that may not yet be published to disk.
4. Commit through UC rather than writing directly to _delta_log/.

The UC integration crates handle steps 1 through 4 while Kernel handles everything else: log replay, data skipping, schema enforcement, and protocol compliance.

The three crates

The UC integration is split across three crates, each with a distinct responsibility.

unity-catalog-delta-client-api: transport-agnostic traits

This crate defines the API contract for communicating with Unity Catalog. It contains no HTTP code or network dependencies. The key types are:

• CommitClient trait: commits a new version to a UC-managed table. Your implementation calls the UC commits API to ratify a staged commit.
• GetCommitsClient trait: retrieves the list of ratified commits for a table. The response includes each commit’s version, file name, size, and the latest ratified table version.
• CommitsRequest / CommitsResponse / Commit: the request and response models for the commits API.
• CommitRequest: the request model for ratifying a single commit.
• TemporaryTableCredentials / AwsTempCredentials / Operation: credential vending models. Operation distinguishes Read, Write, and ReadWrite access.
• InMemoryCommitsClient: a test-only implementation (behind the test-utils feature flag) that stores commits in memory. Useful for unit testing your connector without a live UC server.

Because this crate is transport-agnostic, you can swap in any backend (REST, gRPC, or in-memory) without changing the code that depends on these traits.

unity-catalog-delta-rest-client: HTTP implementation

This crate provides the concrete REST-over-HTTP implementations:

• UCClient: calls the UC tables API (get_table) and the credentials API (get_credentials). You use it to resolve a three-part table name like my_catalog.my_schema.my_table into a table_id and storage_location, then obtain temporary cloud credentials scoped to that location.
• UCCommitsRestClient: implements both CommitClient and GetCommitsClient over HTTP. It talks to the UC commits endpoint to fetch ratified commits and to ratify new ones.
• ClientConfig / ClientConfigBuilder: configuration for the HTTP clients, including the workspace URL and authentication token.

delta-kernel-unity-catalog: the Kernel integration layer

This crate connects the UC client layer to Kernel’s APIs. It depends on both unity-catalog-delta-client-api and delta_kernel. The key types are:

• UCKernelClient<C: GetCommitsClient>: the main entry point. It wraps any GetCommitsClient implementation and provides load_snapshot() and load_snapshot_at() methods. These methods call get_commits, convert the response into a Vec<LogPath> log tail, and pass it to Snapshot::builder_for().with_log_tail() so Kernel can build a Snapshot that includes unpublished commits.
• UCCommitter<C: CommitClient>: implements Kernel’s Committer trait for UC tables. For version 0 (table creation), it writes 000.json directly to the published commit path. For all subsequent versions, it writes a staged commit to _delta_log/_staged_commits/, then calls the UC commit API to ratify it. The publish() method copies ratified staged commits to _delta_log/ as published commits.
• get_required_properties_for_disk(): returns the table properties you must include when creating a UC-managed table (the catalogManaged and vacuumProtocolCheck feature signals, plus the io.unitycatalog.tableId). Kernel’s create_table() consumes these as table properties on the version 0 commit.
• get_final_required_properties_for_uc(): extracts the full set of properties from the post-creation Snapshot (feature signals, protocol versions, in-commit timestamp, optional clustering columns) that you send to your UC server’s table-registration endpoint to finalize the table.

See Creating UC Tables for the end-to-end creation flow and how these two utilities fit together.

Note

UCCommitter requires a multi-threaded tokio runtime. The default Kernel Engine uses tokio, so this is compatible. If you use a custom Engine, ensure your runtime is multi-threaded.

How the crates map to catalog-managed concepts

The Catalog-Managed Tables: Overview describes the generic architecture: a catalog client-side component that resolves tables, fetches commits, and provides a Committer. Here is how the UC crates fill those roles:

Architecture

The following diagram shows how data flows through the three crates when your connector reads or writes a UC-managed table.

 ┌─────────────────────────────────────────────────────────┐
 │                   Your Connector                        │
 │                                                         │
 │  1. UCClient::get_table("catalog.schema.table")         │
 │  2. UCClient::get_credentials(&table_id, Read)          │
 │  3. UCKernelClient::load_snapshot(&table_id, &uri, ..)  │
 │  4. snapshot.scan_builder().build()?.execute(engine)?    │
 └──────────┬──────────────┬───────────────────────────────┘
            │              │
            ▼              ▼
 ┌──────────────────┐  ┌───────────────────────────────────┐
 │  unity-catalog-  │  │  delta-kernel-unity-catalog        │
 │  delta-rest-     │  │                                    │
 │  client          │  │  UCKernelClient                    │
 │                  │  │    calls get_commits()              │
 │  UCClient        │  │    converts to Vec<LogPath>        │
 │  UCCommitsRest   │  │    calls Snapshot::builder_for()   │
 │  Client          │  │      .with_log_tail(commits)       │
 │                  │  │      .build(engine)                 │
 │  Implements:     │  │                                    │
 │  CommitClient    │  │  UCCommitter                       │
 │  GetCommitsClient│  │    implements Committer trait       │
 └──────┬───────────┘  └──────────┬────────────────────────┘
        │                         │
        ▼                         ▼
 ┌──────────────────┐  ┌───────────────────────────────────┐
 │  unity-catalog-  │  │  delta_kernel                      │
 │  delta-client-   │  │                                    │
 │  api             │  │  Snapshot, Scan, Transaction        │
 │                  │  │  Committer trait                    │
 │  CommitClient    │  │  LogPath, SnapshotBuilder           │
 │  GetCommitsClient│  │                                    │
 │  (traits)        │  │  Knows nothing about UC.            │
 └──────────────────┘  └───────────────────────────────────┘

The diagram shows the steady-state commit flow for an existing table. The version 0 commit (table creation) takes a different path: UCCommitter writes _delta_log/00000000000000000000.json directly and skips the UC commits API. See Creating UC Tables for the full creation flow.

Dependencies and feature flags

To use the UC integration, add the following to your Cargo.toml:

[dependencies]
delta-kernel-unity-catalog = { version = "..." }
unity-catalog-delta-rest-client = { version = "..." }

Depend on unity-catalog-delta-client-api whenever you import types from it directly (including Operation, CommitClient, and GetCommitsClient). The REST client crate does not re-export these. You also need the client-api crate when implementing a custom backend, such as a gRPC client.

The delta-kernel-unity-catalog crate has the following feature flags:

The unity-catalog-delta-client-api crate has one feature flag:

Tip

The unity-catalog-delta-rest-client crate also exposes a test-utils feature that enables the test-utils feature on the client API crate transitively. Add it to your [dev-dependencies] to get the in-memory client for tests.

Client configuration and retries

ClientConfigBuilder exposes the following tuning knobs. The defaults are sensible for most workloads, but long-running reads and bursty write paths often benefit from raising timeouts or the retry budget.

use std::time::Duration;
use unity_catalog_delta_rest_client::ClientConfig;

let config = ClientConfig::build(&endpoint, &token)
    .with_timeout(Duration::from_secs(60))
    .with_max_retries(5)
    .with_retry_delays(Duration::from_millis(200), Duration::from_secs(5))
    .build()?;

The REST client automatically retries requests that fail with server errors (HTTP 5xx) or transient network errors, using linear backoff bounded by retry_base_delay and retry_max_delay. Successful 2xx and client errors (HTTP 4xx) are not retried. These retries apply to transport-level failures only. Transaction-level conflicts (another writer won the version) must be handled by the connector through the CommitResult::ConflictedTransaction branch. See Writing to UC Tables for the full retry model.

When not to use this

If your tables are not registered in Unity Catalog, you don’t need these crates. Standard filesystem-managed Delta tables work with Kernel directly. See Building a Scan and Appending Data for the non-catalog path.

If you use a different catalog (Hive Metastore, AWS Glue, Polaris), you need a different catalog client-side component. The generic catalog-managed overview explains the extension points.

What’s next

• Creating UC Tables: how to create a new UC-managed table using get_required_properties_for_disk and get_final_required_properties_for_uc.
• Reading UC Tables: how to load a Snapshot and read data from a UC-managed table.
• Writing to UC Tables: how to commit and publish writes through Unity Catalog.

See also

• Catalog-Managed Tables: Overview: the generic catalog-managed concepts that the UC crates implement.
• Implementing a Catalog Committer: how the Committer trait works under the hood.

Creating Unity Catalog tables

To create a new Unity Catalog-managed Delta table, you register the table with your UC server to obtain a table ID and storage location, write a version 0 Delta log commit with the required catalog-managed properties, and then send a second set of properties back to UC to finalize registration.

Before reading this page, make sure you understand Creating a Table and the Unity Catalog integration overview.

Warning

Steps 1 and 5 below call UC endpoints that are not yet exposed by the Rust unity-catalog-delta-rest-client crate. Route those calls through your connector’s own UC client. They are planned for inclusion in the unity-catalog-delta-client-api crate.

Prerequisites

• A three-part table name, Delta schema, and target storage location.
• A connector-owned UC client that can call UC’s staging and create-table endpoints.

Step 1: Reserve the table in Unity Catalog

// TODO: not yet exposed by `unity-catalog-delta-rest-client`. Call through
// your connector's own UC client.
let staging_info = my_uc_client.get_staging_table(
    "main.default.my_table",
    &schema,
).await?;
let table_id = staging_info.table_id;
let table_uri = staging_info.storage_location;

Step 2: Collect the disk-bound properties

use delta_kernel_unity_catalog::get_required_properties_for_disk;

let disk_props = get_required_properties_for_disk(&table_id);

The returned map has exactly three entries:

Note

The map intentionally omits the inCommitTimestamp feature and delta.enableInCommitTimestamps=true. Kernel’s create_table() auto-enables both when it sees the catalogManaged feature.

Step 3: Build and commit the version 0 transaction

use std::sync::Arc;
use delta_kernel::transaction::create_table::create_table;
use delta_kernel::transaction::CommitResult;
use delta_kernel_unity_catalog::UCCommitter;
use unity_catalog_delta_client_api::Operation;
use unity_catalog_delta_rest_client::{ClientConfig, UCClient, UCCommitsRestClient};

let config = ClientConfig::build(&endpoint, &token).build()?;
let uc_client = UCClient::new(config.clone())?;
let commits_client = Arc::new(UCCommitsRestClient::new(config)?);

// Credentials. Use ReadWrite so the engine can write 000.json into storage.
let creds = uc_client.get_credentials(&table_id, Operation::ReadWrite).await?;
let engine = build_engine_with_credentials(&table_uri, &creds)?;

// Build the create-table transaction with the disk-bound properties.
let committer = Box::new(UCCommitter::new(commits_client.clone(), table_id.clone()));
let create_txn = create_table(table_uri.as_str(), Arc::new(schema), "MyApp/1.0")
    .with_table_properties(disk_props)
    .build(&engine, committer)?;

let post_commit_snapshot = match create_txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => committed
        .post_commit_snapshot()
        .cloned()
        .expect("post-commit snapshot is always populated for create table"),
    CommitResult::ConflictedTransaction(_) => {
        // Another writer created the table first. Delete the UC reservation
        // and fail, or fall through to read the existing table.
        return Err("table already exists".into());
    }
    CommitResult::RetryableTransaction(_) => {
        return Err("version 0 commit failed with a transient error; retry".into());
    }
};

On version 0, UCCommitter writes _delta_log/00000000000000000000.json directly and skips the UC commits API.

See build_engine_with_credentials in Step 4 of Reading UC Tables for the engine construction details.

Step 4: Collect the final UC-bound properties

use delta_kernel_unity_catalog::get_final_required_properties_for_uc;

let uc_props = get_final_required_properties_for_uc(&post_commit_snapshot, &engine)?;

The returned map contains:

• Every entry from the table’s metadata configuration, including io.unitycatalog.tableId, delta.enableInCommitTimestamps=true, and any user-supplied custom properties.
• delta.minReaderVersion and delta.minWriterVersion.
• delta.feature.<name>=supported for every reader and writer feature on the protocol (for a freshly created UC table this is at least catalogManaged, vacuumProtocolCheck, and inCommitTimestamp).
• delta.lastUpdateVersion=0.
• delta.lastCommitTimestamp set to the in-commit timestamp of version 0.
• clusteringColumns as a JSON array of logical column paths, if the table is clustered.

Note

get_final_required_properties_for_uc requires a version 0 snapshot with an in-commit timestamp. The post_commit_snapshot from Step 3 satisfies both.

Step 5: Finalize the table in Unity Catalog

// TODO: not yet exposed by `unity-catalog-delta-rest-client`. Call through
// your connector's own UC client.
my_uc_client.create_table(&table_id, uc_props).await?;

Clustered tables

Chain with_data_layout on the create-table builder:

use delta_kernel::transaction::data_layout::DataLayout;

let create_txn = create_table(table_uri.as_str(), Arc::new(schema), "MyApp/1.0")
    .with_table_properties(disk_props)
    .with_data_layout(DataLayout::clustered(["region"]))
    .build(&engine, committer)?;

get_final_required_properties_for_uc adds a clusteringColumns entry (a JSON array of logical column paths) to its output when clustering is enabled.

What’s next

• Writing to UC Tables for the version >= 1 write flow that follows table creation.
• Reading UC Tables for the read flow.
• Creating a Table for the generic Kernel create-table APIs.

See also

• Catalog-Managed Tables: Overview for staged / ratified / published commit terminology.
• Implementing a Catalog Committer for the Committer trait behind UCCommitter.

Reading Unity Catalog tables

To read a Unity Catalog-managed Delta table, you resolve the table name through the UC REST API, fetch temporary storage credentials, load a Snapshot through the UCKernelClient, and then build a Scan exactly as you would for a filesystem-managed table.

Before reading this page, make sure you understand Catalog-Managed Tables and the Unity Catalog Integration overview.

Note

This example uses the delta-kernel-unity-catalog and unity-catalog-delta-rest-client crates, which are not part of the delta_kernel crate itself. Add them as dependencies alongside delta_kernel.

Dependencies

Add the following to your Cargo.toml:

[dependencies]
delta_kernel = { version = "...", features = ["default-engine-rustls"] }
delta-kernel-unity-catalog = "..."
unity-catalog-delta-rest-client = "..."
unity-catalog-delta-client-api = "..."
url = "2"
tokio = { version = "1", features = ["full"] }

Use default-engine-rustls for a pure-Rust TLS stack or default-engine-native-tls to link against the system’s TLS implementation. You need unity-catalog-delta-client-api directly to import types like Operation, since the REST client crate does not re-export them.

Step 1: Build the UC clients

The UC integration uses two clients that share a ClientConfig:

• UCClient handles table resolution (get_table) and credential vending (get_credentials).
• UCCommitsRestClient implements the GetCommitsClient trait, which UCKernelClient uses to fetch ratified commits from the catalog.

use unity_catalog_delta_rest_client::{ClientConfig, UCClient, UCCommitsRestClient};

let config = ClientConfig::build(&endpoint, &token).build()?;
let uc_client = UCClient::new(config.clone())?;
let commits_client = UCCommitsRestClient::new(config)?;

The endpoint is your UC workspace URL (e.g. "my-workspace.cloud.databricks.com"), and token is a valid authentication token.

Step 2: Resolve the table name

Call get_table with the three-level table name to get the table’s storage location and table ID. The table_id identifies the table in UC’s commits API, and the storage_location points to the table’s root directory in cloud storage.

let table_info = uc_client.get_table("my_catalog.my_schema.my_table").await?;
let table_id = &table_info.table_id;
let table_uri = &table_info.storage_location;

The TablesResponse also includes metadata like catalog_name, schema_name, data_source_format, and table_type. You can verify the table is a Delta table by calling table_info.is_delta_table().

Step 3: Fetch temporary credentials

UC vends short-lived cloud storage credentials scoped to the table’s storage location. For a read operation, request Operation::Read:

use unity_catalog_delta_client_api::Operation;

let creds = uc_client.get_credentials(table_id, Operation::Read).await?;

The returned TemporaryTableCredentials contains cloud-provider-specific credentials. For AWS, extract the temporary credentials:

let aws_creds = creds.aws_temp_credentials
    .ok_or("No AWS temporary credentials in response")?;

Warning

Vended credentials expire. The TemporaryTableCredentials struct provides expiration_time, is_expired(), and time_until_expiry() to check validity. If your scan takes longer than the credential lifetime, you need to refresh credentials and rebuild the Engine before continuing. See Credential refresh below.

Note

Today TemporaryTableCredentials only exposes aws_temp_credentials. Azure and GCP credential vending are tracked in delta-io/delta-kernel-rs#2434.

Step 4: Build an Engine with vended credentials

Pass the temporary credentials as storage options when constructing the object store, then wrap it in a DefaultEngineBuilder:

use std::sync::Arc;
use delta_kernel::engine::default::DefaultEngineBuilder;
use delta_kernel::object_store;

let table_url = url::Url::parse(table_uri)?;
let options = [
    ("region", "us-west-2"),
    ("access_key_id", &aws_creds.access_key_id),
    ("secret_access_key", &aws_creds.secret_access_key),
    ("session_token", &aws_creds.session_token),
];
let (store, _path) = object_store::parse_url_opts(&table_url, options)?;
let engine = DefaultEngineBuilder::new(store.into()).build();

The .into() converts the Box<dyn ObjectStore> returned by parse_url_opts into the Arc<dyn ObjectStore> that DefaultEngineBuilder::new expects. Set the region to match the bucket’s actual AWS region.

Step 5: Load a Snapshot through UCKernelClient

UCKernelClient wraps a GetCommitsClient and handles the full snapshot loading flow: it calls the UC commits API to get the latest ratified commits, translates them into LogPath entries, and builds a Snapshot with the log tail.

use delta_kernel_unity_catalog::UCKernelClient;

let catalog = UCKernelClient::new(&commits_client);
let snapshot = catalog.load_snapshot(table_id, table_uri, &engine).await?;

println!("Table version: {}", snapshot.version());

To read a specific version, use load_snapshot_at:

let snapshot = catalog.load_snapshot_at(table_id, table_uri, 5, &engine).await?;

The requested version must not exceed the latest version the catalog has ratified. If it does, load_snapshot_at returns an error.

Step 6: Build and execute a Scan

From this point on, reading works identically to a filesystem-managed table. Build a Scan from the Snapshot and iterate over the results:

let scan = snapshot.scan_builder().build()?;
for data in scan.execute(Arc::new(engine))? {
    let batch = data?;
    // process each batch of data
}

You can use all the same Scan features: column selection, filter pushdown, and scan metadata for distributed reads.

Complete example

This example puts all the steps together. It connects to Unity Catalog, resolves a table, fetches credentials, loads a Snapshot, and reads the data.

use std::sync::Arc;

use delta_kernel::engine::default::DefaultEngineBuilder;
use delta_kernel::object_store;
use delta_kernel_unity_catalog::UCKernelClient;
use unity_catalog_delta_client_api::Operation;
use unity_catalog_delta_rest_client::{ClientConfig, UCClient, UCCommitsRestClient};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Build UC clients
    let config = ClientConfig::build(&endpoint, &token).build()?;
    let uc_client = UCClient::new(config.clone())?;
    let commits_client = UCCommitsRestClient::new(config)?;

    // 2. Resolve the table name
    let table_info = uc_client.get_table("my_catalog.my_schema.my_table").await?;
    let table_id = &table_info.table_id;
    let table_uri = &table_info.storage_location;

    // 3. Fetch temporary credentials
    let creds = uc_client.get_credentials(table_id, Operation::Read).await?;
    let aws_creds = creds.aws_temp_credentials
        .ok_or("No AWS temporary credentials")?;

    // 4. Build an Engine with the vended credentials
    let table_url = url::Url::parse(table_uri)?;
    let (store, _) = object_store::parse_url_opts(&table_url, [
        ("region", "us-west-2"),
        ("access_key_id", &aws_creds.access_key_id),
        ("secret_access_key", &aws_creds.secret_access_key),
        ("session_token", &aws_creds.session_token),
    ])?;
    let engine = DefaultEngineBuilder::new(store.into()).build();

    // 5. Load the Snapshot via UCKernelClient
    let catalog = UCKernelClient::new(&commits_client);
    let snapshot = catalog.load_snapshot(table_id, table_uri, &engine).await?;
    println!("Loaded table at version {}", snapshot.version());

    // 6. Build and execute the Scan
    let scan = snapshot.scan_builder().build()?;
    for data in scan.execute(Arc::new(engine))? {
        let batch = data?;
        // process each batch of data
    }

    Ok(())
}

Credential refresh for long-running operations

UC vends temporary credentials with a limited lifetime. For short-lived scans, you don’t need to worry about expiration. For long-running operations (large table scans, streaming reads), check creds.is_expired() or creds.time_until_expiry() before starting a new phase of work.

If credentials have expired, call get_credentials again and rebuild the Engine with the fresh credentials before continuing. Kernel does not manage credential lifecycle for you.

What’s next

• Writing to UC Tables: how to append data and commit through Unity Catalog
• Building a Scan: scan configuration options that apply to both filesystem and catalog-managed tables
• Filter Pushdown and File Skipping: reduce the data you read with predicates

Writing to Unity Catalog tables

To write to a Unity Catalog-managed Delta table, you create a UCCommitter, pass it to a Kernel transaction, and then publish the staged commit to make it visible in _delta_log/.

Before reading this page, make sure you understand the generic catalog-managed write lifecycle and the Unity Catalog integration overview.

Note

This page uses the delta-kernel-unity-catalog and unity-catalog-delta-rest-client crates. All code examples use rust,ignore because they require these external crates.

Set up clients and resolve the table

Use UCClient to resolve the table name and fetch read-write credentials, then build a UCCommitsRestClient for the commits API. Both clients share a ClientConfig.

use std::sync::Arc;
use unity_catalog_delta_client_api::Operation;
use unity_catalog_delta_rest_client::{ClientConfig, UCClient, UCCommitsRestClient};

let config = ClientConfig::build("my-workspace.cloud.databricks.com", token).build()?;
let uc_client = UCClient::new(config.clone())?;
let commits_client = Arc::new(UCCommitsRestClient::new(config)?);

// Resolve table name to table ID and storage location
let table_info = uc_client.get_table("my_catalog.my_schema.my_table").await?;
let table_id = &table_info.table_id;
let table_uri = &table_info.storage_location;

// Fetch read-write credentials for the table's cloud storage
let creds = uc_client.get_credentials(table_id, Operation::ReadWrite).await?;

For the full details on resolving tables and building an engine with vended credentials, see Reading UC Tables.

Load a snapshot with the log tail

Use UCKernelClient to load a snapshot. It fetches the catalog’s log tail (staged commits) and passes them to Kernel’s Snapshot::builder_for with with_log_tail and with_max_catalog_version.

use delta_kernel_unity_catalog::UCKernelClient;

let catalog = UCKernelClient::new(commits_client.as_ref());
let snapshot = catalog.load_snapshot(table_id, table_uri, &engine).await?;

The returned Snapshot reflects all ratified commits the catalog knows about, including those that haven’t been published to _delta_log/ yet.

Create a transaction with UCCommitter

UCCommitter implements Kernel’s Committer trait. It stages the commit to _staged_commits/, then calls the UC commits API to ratify it.

use delta_kernel_unity_catalog::UCCommitter;

let committer = Box::new(UCCommitter::new(commits_client.clone(), table_id.clone()));
let mut txn = snapshot.clone().transaction(committer, &engine)?
    .with_operation("INSERT".to_string());

UCCommitter requires a multi-threaded tokio runtime. The default Kernel engine already uses tokio, so this is compatible as long as you use the multi-threaded runtime (the default for #[tokio::main]).

Warning

UCCommitter validates that every commit targets a catalog-managed table with in-commit timestamps enabled. It rejects the commit if the table is missing the catalogManaged, vacuumProtocolCheck, or inCommitTimestamp writer features, if delta.enableInCommitTimestamps is not "true", or if io.unitycatalog.tableId does not match the committer’s table_id. Tables created through Creating UC Tables satisfy all of these automatically.

Write data

From this point, writing data works the same as any Kernel transaction. Get the write context, write Parquet files to the table’s storage location, and add the resulting file metadata to the transaction.

let write_context = txn.unpartitioned_write_context()?;
// ... write Parquet files using write_context ...
txn.add_files(file_metadata);

See Appending Data for the full details on writing Parquet files and registering file metadata.

Commit and handle the result

Call txn.commit() to stage the commit and ratify it through UC.

use delta_kernel::transaction::CommitResult;

match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        let version = committed.commit_version();
        let post_commit_snapshot = committed
            .post_commit_snapshot()
            .expect("post-commit snapshot");
        // Proceed to publish (next step)
    }
    CommitResult::ConflictedTransaction(conflicted) => {
        // Another writer committed this version first. Rebase onto the new
        // snapshot and retry. UCCommitter does not retry at this level.
    }
    CommitResult::RetryableTransaction(_retryable) => {
        // Transient I/O or server error after the UC HTTP client's own retry
        // budget was exhausted. Retry the commit from scratch.
    }
}

The REST client automatically retries transport-level failures (HTTP 5xx, connection errors) according to the retry knobs on ClientConfigBuilder. See Client configuration and retries. Once that budget is exhausted, UCCommitter surfaces the failure as CommitResult::RetryableTransaction. Transaction-level retries (including rebasing after a ConflictedTransaction) are the connector’s responsibility; UCCommitter does not retry commits itself.

Under the hood, UCCommitter::commit does two things for versions >= 1:

1. Writes the transaction’s actions to _delta_log/_staged_commits/<version>.<uuid>.json
2. Calls the UC commits API to ratify the staged commit

For version 0 (table creation), the committer writes directly to _delta_log/00000000000000000000.json instead and skips the UC commits API. See Creating UC Tables for the full creation flow and the catalog-managed write lifecycle for the generic ratification flow.

Warning

UCCommitter does not support ALTER TABLE operations (protocol changes, metadata changes, or clustering column changes). It also rejects attempts to upgrade a path-based table to catalog-managed or downgrade a catalog-managed table to path-based. Attempting any of these returns an error.

Publish staged commits

After a successful commit, publish the staged commit so it becomes visible as a normal delta file in _delta_log/. Without publishing, only catalog-aware readers can see the commit.

use delta_kernel_unity_catalog::UCCommitter;

// Keep a reference to the committer for publishing
let committer: Box<dyn delta_kernel::committer::Committer> =
    Box::new(UCCommitter::new(commits_client.clone(), table_id.clone()));

let published_snapshot = post_commit_snapshot
    .publish(&engine, committer.as_ref())?;

Publishing copies each staged commit from _staged_commits/<version>.<uuid>.json to _delta_log/<version>.json. If a published file already exists (from a previous publish attempt), the copy is silently skipped.

Post-publish maintenance

Once commits are published, you can checkpoint the table:

published_snapshot.checkpoint(&engine)?;

Checkpointing requires published commits. If you skip the publish step, checkpointing fails because it can only operate on published versions.

Complete example

use std::sync::Arc;
use delta_kernel::transaction::CommitResult;
use delta_kernel_unity_catalog::{UCCommitter, UCKernelClient};
use unity_catalog_delta_client_api::Operation;
use unity_catalog_delta_rest_client::{ClientConfig, UCClient, UCCommitsRestClient};

// 1. Set up clients
let config = ClientConfig::build("my-workspace.cloud.databricks.com", token).build()?;
let uc_client = UCClient::new(config.clone())?;
let commits_client = Arc::new(UCCommitsRestClient::new(config)?);

// 2. Resolve table and fetch credentials
let table_info = uc_client.get_table("my_catalog.my_schema.my_table").await?;
let table_id = &table_info.table_id;
let table_uri = &table_info.storage_location;
let creds = uc_client.get_credentials(table_id, Operation::ReadWrite).await?;

// 3. Build engine with vended credentials. `build_engine_with_credentials` is
//    a connector-owned helper, not part of the library. See Step 4 of
//    [Reading UC Tables](./reading.md) for the full expansion.
let engine = build_engine_with_credentials(table_uri, &creds)?;

// 4. Load snapshot via UC log tail
let catalog = UCKernelClient::new(commits_client.as_ref());
let snapshot = catalog.load_snapshot(table_id, table_uri, &engine).await?;

// 5. Create transaction with UCCommitter
let committer = Box::new(UCCommitter::new(commits_client.clone(), table_id.clone()));
let mut txn = snapshot.clone().transaction(committer, &engine)?
    .with_operation("INSERT".to_string());

// 6. Write data
let write_context = txn.unpartitioned_write_context()?;
// ... write Parquet files using write_context ...
txn.add_files(file_metadata);

// 7. Commit, publish, and checkpoint
let committer_for_publish: Box<dyn delta_kernel::committer::Committer> =
    Box::new(UCCommitter::new(commits_client.clone(), table_id.clone()));

match txn.commit(&engine)? {
    CommitResult::CommittedTransaction(committed) => {
        let post_commit_snapshot = committed
            .post_commit_snapshot()
            .expect("post-commit snapshot");

        // Publish staged commits to _delta_log/
        let published_snapshot = post_commit_snapshot
            .publish(&engine, committer_for_publish.as_ref())?;

        // Checkpoint the published snapshot
        published_snapshot.checkpoint(&engine)?;
    }
    CommitResult::ConflictedTransaction(_) => { /* rebase and retry */ }
    CommitResult::RetryableTransaction(_) => { /* retry the commit */ }
}

What’s next

• Creating UC Tables for creating a brand-new UC-managed table before you can write to it
• Catalog-managed write lifecycle for the generic commit and publish flow
• Appending Data for the details of writing Parquet files and collecting file metadata
• Reading UC Tables for resolving tables and loading snapshots

Configuring storage

To configure how DefaultEngine accesses your Delta tables, you create an object store from a URL and pass it to the engine builder. The DefaultEngine uses the object_store crate for all storage I/O, supporting local files, S3, GCS, and Azure out of the box.

Before reading this page, make sure you understand The Engine Trait.

Note

The storage APIs on this page require one of the default-engine feature flags (default-engine-rustls or default-engine-native-tls). See Feature Flags for details.

Kernel provides several paths to construct an object store, depending on how much control you need:

flowchart TD
    A[Need a DefaultEngine] --> B{How do you get<br/>your ObjectStore?}
    B -->|"Standard URL"| C["store_from_url(&url)"]
    B -->|"URL + options"| D["store_from_url_opts(&url, opts)"]
    B -->|"Custom URL scheme"| E["insert_url_handler(scheme, handler)\nthen store_from_url"]
    B -->|"Bring your own"| F["Build ObjectStore directly"]
    C --> G["Arc#lt;dyn ObjectStore#gt;"]
    D --> G
    E --> G
    F --> G
    G --> H["DefaultEngine::builder(store).build()"]

    click C href "#standard-url"
    click D href "#url-with-options"
    click E href "#custom-url-schemes"
    click F href "#bringing-your-own-object-store"

Standard URL

store_from_url creates an object store from a URL. The object_store crate detects the storage backend from the URL scheme:

extern crate delta_kernel;
extern crate url;
use std::sync::Arc;
use url::Url;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url;
use delta_kernel::DeltaResult;
fn main() -> DeltaResult<()> {
let url = Url::parse("file:///path/to/table")?;
let store = store_from_url(&url)?;
let engine = DefaultEngine::builder(store).build();
Ok(())
}

URL with options

To pass provider-specific options (credentials, region, endpoint, etc.), use store_from_url_opts. These options are forwarded directly to the object_store crate:

extern crate delta_kernel;
extern crate url;
use std::collections::HashMap;
use url::Url;
use delta_kernel::engine::default::DefaultEngine;
use delta_kernel::engine::default::storage::store_from_url_opts;
use delta_kernel::DeltaResult;
fn main() -> DeltaResult<()> {
let url = Url::parse("s3://my-bucket/path/to/table")?;
let options = HashMap::from([
    ("region", "us-west-2"),
    ("access_key_id", "AKIA..."),
    ("secret_access_key", "..."),
]);
let store = store_from_url_opts(&url, options)?;
let engine = DefaultEngine::builder(store).build();
Ok(())
}

See the object_store documentation for the full list of supported options per storage provider.

Custom URL schemes

If you need to support a URL scheme that object_store doesn’t handle natively (e.g. hdfs://), register a handler with insert_url_handler:

use std::sync::Arc;
use delta_kernel::engine::default::storage::insert_url_handler;

insert_url_handler("hdfs", Arc::new(|url, options| {
    // Build your custom ObjectStore from the URL and options
    let store = build_hdfs_store(url, &options)?;
    let path = object_store::path::Path::parse(url.path())?;
    Ok((Box::new(store), path))
}))?;

// Now store_from_url uses your handler for hdfs:// URLs
let store = store_from_url(&url)?;

The handler closure receives a &Url and a HashMap<String, String> of options, and returns a Result<(Box<dyn ObjectStore>, Path), Error>.

Bringing your own object store

To bypass URL-based construction entirely, build an ObjectStore instance directly and pass it to the engine builder:

use std::sync::Arc;
use object_store::local::LocalFileSystem;
use delta_kernel::engine::default::DefaultEngine;

let store = Arc::new(LocalFileSystem::new());
let engine = DefaultEngine::builder(store).build();

This is useful when you need full control over the store configuration or want to use a store implementation that isn’t reachable via URL parsing.

Metrics and monitoring

To observe what Kernel is doing at runtime, install a tracing-subscriber layer that converts Kernel’s spans and events into MetricEvent values. Kernel instruments snapshot loading, scanning, and storage I/O with the tracing crate, so you can also receive raw spans and events in any subscriber you already use for logging.

How metrics flow

Kernel emits tracing spans at key milestones (snapshot build, log segment load, scan metadata replay, storage I/O). A ReportGeneratorLayer, attached to your subscriber, watches those spans and forwards MetricEvent values to a MetricsReporter you provide. Your reporter can do whatever you want with each event: print it, push it to Prometheus, increment a counter, or fan it out to several destinations.

Kernel code
    | tracing span / event
    v
tracing-subscriber Registry
    +-- ReportGeneratorLayer  --->  MetricsReporter (your impl)
    +-- fmt::layer (logs)
    +-- EnvFilter, etc.

This means metrics aren’t tied to your Engine. You wire them up once, at process startup, alongside any other tracing layers you want.

Enabling metrics

To enable metrics, build a tracing subscriber, add the metrics layer with WithMetricsReporterLayer::with_metrics_reporter_layer, and call init(). Kernel ships LoggingMetricsReporter as a built-in reporter that logs each event at a tracing level you choose.

Filename: src/main.rs

use std::sync::Arc;
use delta_kernel::metrics::{LoggingMetricsReporter, WithMetricsReporterLayer};
use tracing_subscriber::layer::SubscriberExt as _;
use tracing_subscriber::util::SubscriberInitExt as _;

fn main() {
    tracing_subscriber::registry()
        .with(tracing_subscriber::fmt::layer())
        .with_metrics_reporter_layer(
            Arc::new(LoggingMetricsReporter::new(tracing::Level::INFO)),
        )
        .with(tracing_subscriber::EnvFilter::from_default_env())
        .init();

    // ... build engine and read tables as usual
}

If you don’t install the layer, no MetricEvent values are produced. The underlying spans still exist, so other tracing layers (logging, distributed tracing) keep working.

Note

The MetricsReporter trait, WithMetricsReporterLayer extension, and LoggingMetricsReporter live under delta_kernel::metrics. You also need the tracing and tracing-subscriber crates as direct dependencies of your connector.

Implementing a custom MetricsReporter

The MetricsReporter trait has a single method:

pub trait MetricsReporter: Send + Sync + std::fmt::Debug {
    fn report(&self, event: MetricEvent);
}

Your reported must be Send + Sync because the layer can call report from any thread that produces a span. Keep report cheap. If your destination is slow, push the event onto a channel and drain it from a worker.

Filename: src/reporter.rs

use delta_kernel::metrics::{MetricEvent, MetricsReporter};

#[derive(Debug)]
struct StdoutReporter;

impl MetricsReporter for StdoutReporter {
    fn report(&self, event: MetricEvent) {
        println!("[kernel-metrics] {event}");
    }
}

Wire it in the same way as LoggingMetricsReporter:

use std::sync::Arc;
use delta_kernel::metrics::WithMetricsReporterLayer;
use tracing_subscriber::layer::SubscriberExt as _;
use tracing_subscriber::util::SubscriberInitExt as _;

tracing_subscriber::registry()
    .with(tracing_subscriber::fmt::layer())
    .with_metrics_reporter_layer(Arc::new(StdoutReporter))
    .init();

When a snapshot loads, you’ll see output like:

[kernel-metrics] LogSegmentLoaded(id=a1b2c3d4-..., duration=12.34ms, commits=5, checkpoints=1, compactions=0)
[kernel-metrics] ProtocolMetadataLoaded(id=a1b2c3d4-..., duration=3.21ms)
[kernel-metrics] SnapshotCompleted(id=a1b2c3d4-..., version=5, duration=15.55ms)

Metric events

Every callback receives a MetricEvent enum value. Events fall into three categories: snapshot lifecycle, scan metadata, and storage/file I/O.

Snapshot lifecycle events

These events track the process of loading a Snapshot from the Delta log. Each carries an operation_id (MetricId) that ties all events from the same snapshot load together.

Scan metadata events

ScanMetadataCompleted is emitted when a scan metadata iterator is fully consumed. It provides detailed statistics about the log replay process:

The ScanType enum

The scan_type field tells you which scan execution path produced the event:

If you use parallel_scan_metadata, you’ll receive two ScanMetadataCompleted events per scan: one for each phase.

Storage and file I/O events

These events track low-level I/O operations. The default storage, JSON, and Parquet handlers emit them automatically when the metrics layer is installed. Unlike snapshot events, these don’t carry an operation_id because a single storage call may serve multiple higher-level operations.

Note

If you implement a custom JsonHandler or ParquetHandler, call delta_kernel::metrics::emit_json_read_completed or emit_parquet_read_completed once per read call so connectors that install the metrics layer still see those events from your handler.

Correlating events with MetricId

Several events include an operation_id field of type MetricId. This is a UUID that uniquely identifies an operation instance. All events from the same snapshot load share the same MetricId, so you can group them to reconstruct a timeline:

1. LogSegmentLoaded (how long listing took, how many files)
2. ProtocolMetadataLoaded (how long protocol/metadata parsing took)
3. SnapshotCompleted or SnapshotFailed (final outcome and total duration)

You can store the MetricId in your monitoring system as a trace ID or correlation key. Because everything flows through tracing, you can also correlate Kernel’s events with your own application spans by attaching them to the same parent span.

Example: correlating snapshot events by operation ID

To aggregate timing data per operation, store intermediate events keyed by MetricId and compute totals when the terminal event arrives:

Filename: src/correlating_reporter.rs

use std::collections::HashMap;
use std::sync::Mutex;
use delta_kernel::metrics::{MetricEvent, MetricId, MetricsReporter};

#[derive(Debug)]
struct CorrelatingReporter {
    pending: Mutex<HashMap<MetricId, Vec<MetricEvent>>>,
}

impl CorrelatingReporter {
    fn new() -> Self {
        Self { pending: Mutex::new(HashMap::new()) }
    }
}

impl MetricsReporter for CorrelatingReporter {
    fn report(&self, event: MetricEvent) {
        match &event {
            // 1. Buffer intermediate events, grouped by their shared operation_id.
            MetricEvent::LogSegmentLoaded { operation_id, .. }
            | MetricEvent::ProtocolMetadataLoaded { operation_id, .. } => {
                let mut map = self.pending.lock().unwrap();
                map.entry(*operation_id).or_default().push(event);
            }
            // 2. When the terminal event arrives, drain the group and
            //    compute aggregates (here, a count of sub-events).
            MetricEvent::SnapshotCompleted {
                operation_id,
                version,
                total_duration,
            } => {
                let mut map = self.pending.lock().unwrap();
                if let Some(events) = map.remove(operation_id) {
                    println!(
                        "Snapshot v{version} completed in {total_duration:?} ({} sub-events)",
                        events.len()
                    );
                }
            }
            // On failure, discard the buffered group for this operation.
            MetricEvent::SnapshotFailed { operation_id, .. } => {
                let mut map = self.pending.lock().unwrap();
                map.remove(operation_id);
                println!("Snapshot failed for operation {operation_id}");
            }
            // Storage and scan events don't participate in snapshot correlation.
            _ => {}
        }
    }
}

Sending metrics to multiple destinations

To report to more than one system (for example, a logger and Prometheus), create a composite reporter that fans out to multiple inner reporters:

Filename: src/composite_reporter.rs

use std::sync::Arc;
use delta_kernel::metrics::{MetricEvent, MetricsReporter};

#[derive(Debug)]
struct CompositeReporter {
    reporters: Vec<Arc<dyn MetricsReporter>>,
}

impl MetricsReporter for CompositeReporter {
    fn report(&self, event: MetricEvent) {
        for reporter in &self.reporters {
            reporter.report(event.clone());
        }
    }
}

MetricEvent implements Clone, so each inner reporter receives its own copy. You only install one ReportGeneratorLayer on your subscriber. The composite fans out from there.

See also

• tracing and tracing-subscriber crate documentation, for layer composition, filters, and other subscribers.

FFI (C/C++ integration)

The delta_kernel_ffi crate exposes delta-kernel-rs to C and C++ through a stable FFI boundary. It uses cbindgen to generate header files (.h and .hpp) at build time. This matters because it lets you build a connector in any language that can call C functions, not only Rust.

Building the FFI crate

The crate can be built as a shared library (cdylib) or static library (staticlib):

# Shared library (e.g. libdelta_kernel_ffi.so / .dylib / .dll)
cargo build -p delta_kernel_ffi --release

# Generated headers are written to target/ffi-headers/
# - delta_kernel_ffi.h   (C)
# - delta_kernel_ffi.hpp (C++)

Feature flags

Note: You must enable exactly one of default-engine-rustls or default-engine-native-tls. The default-engine-base feature contains shared implementation details and is not meant to be enabled directly.

The handle system

Objects that cross the FFI boundary are wrapped in handles. These are opaque pointers that carry ownership semantics. There are two kinds:

• Mutable handles (Box-like) represent exclusive ownership. Dropping the handle drops the underlying object. These are neither Copy nor Clone.
• Shared handles (Arc-like) represent shared ownership. Dropping the handle only drops the underlying object if it was the last reference.

Every handle has a corresponding free_* function that you must call to release it. For example, free_engine, free_snapshot, free_scan, free_transaction.

Several FFI functions consume their handle argument and return a new handle. After calling such a function, you must not use the old handle. The function documentation notes this with “CONSUMES the handle.”

Core API surface

The FFI mirrors the Rust API. A typical read flow looks like (no transaction is needed for reads):

get_default_engine()        ->  Handle<SharedExternEngine>
        |
get_snapshot_builder()      ->  Handle<MutableFfiSnapshotBuilder>
        |
snapshot_builder_build()    ->  Handle<SharedSnapshot>
        |
      scan()                ->  Handle<SharedScan>
        |
scan_metadata_iter_init()   ->  Handle<SharedScanMetadataIterator>
        |
  (read parquet, apply transforms, apply selection vectors)

A typical write flow:

get_default_engine()  ->  Handle<SharedExternEngine>
        |
    transaction()     ->  Handle<ExclusiveTransaction>
        |
  with_engine_info()  ->  Handle<ExclusiveTransaction>
        |
    add_files()
        |
    commit()          ->  ExternResult<u64>  (committed version)

For more control over scans, you can use the scan builder API instead of the convenience scan() function:

scan_builder()                ->  Handle<ExclusiveScanBuilder>
        |
scan_builder_with_predicate() ->  Handle<ExclusiveScanBuilder>
        |
scan_builder_with_schema()    ->  Handle<ExclusiveScanBuilder>
        |
scan_builder_build()          ->  Handle<SharedScan>

Public FFI functions

The tables below group the stable FFI functions by purpose. Unless noted, each function is available with the default feature flags. For the full, authoritative list and signatures, consult the generated delta_kernel_ffi.h header.

Engine creation

Snapshots

Snapshot inspection

Use these on a Handle<SharedSnapshot> to read table metadata, protocol, and partition columns without building a scan.

Schema and metadata visitors

Kernel exposes schemas, protocols, and metadata through visitor callbacks so the engine can materialize them into its own types without Kernel allocating engine-owned memory.

See Visitor callbacks below for the pattern.

Schema construction (projection pushdown)

The build-side counterpart to visit_schema: per-field callbacks that let the engine construct a Kernel StructType from its own type system (for example, to pass to scan_builder_with_schema).

Reading (scans)

Engine data and Arrow interop

Rows marked (requires default-engine-base) are only compiled when that feature is enabled; the rest are always available.

Warning: get_raw_engine_data is always exported (regardless of feature flags) but unimplemented. It calls todo!() and will panic. Do not use it.

Writing (transactions)

Write context and file writing

Use a WriteContext to learn where to write parquet files and what schema to write. For unpartitioned writes, one context serves the whole transaction. Partitioned writes (which would use one context per partition) are tracked in #2355.

Engines must append their own <uuid>.parquet filename (and any subdirectory layout) onto the returned table root. The kernel-side WriteContext::write_dir helper – which produces the recommended directory (Hive-style partition paths for partitioned tables when column mapping is off, or a random 2-char prefix when column mapping is on) – is internal and has no FFI binding.

Domain metadata

Table creation

Change data feed (table changes)

Incremental log reads for change data feed. Mirrors the Rust TableChanges API. The entire table_changes module is gated behind the default-engine-base feature.

Warning

scan_table_changes_next returns *mut ArrowFFIData (a heap-allocated Arrow C Data Interface batch). C callers must release each non-null result with free_arrow_ffi_data exactly once. This is an ABI-breaking change from earlier releases that used a different return shape, so connectors upgrading across that boundary need to update both the type and the cleanup path.

Checkpointing

Unity Catalog integration

Requires the delta-kernel-unity-catalog feature. Lets the engine provide a catalog-aware committer without implementing the committer trait from scratch.

Expressions and predicates

Kernel’s expression system is exposed through two parallel surfaces:

• Build (engine AST -> Kernel expression): the engine calls visit_engine_expression / visit_engine_predicate, providing an engine-side iterator. From inside that callback, the engine uses the visit_expression_* / visit_predicate_* builder functions to append Kernel-side nodes (columns, literals, operators) into an internal KernelExpressionVisitorState. The final handle is a SharedExpression or SharedPredicate.
• Walk (Kernel expression -> engine type): the engine calls visit_expression / visit_predicate, supplying an EngineExpressionVisitor struct whose function pointers Kernel invokes as it traverses the tree.

Most engines need only one of the two surfaces.

Tracing

Enable Kernel’s internal tracing instrumentation. Requires the tracing feature flag. Call at most one of these during a process lifetime; later calls return false.

Common utilities

Visitor callbacks

Several FFI entry points take a visitor struct (for example, EngineSchemaVisitor for visit_schema, EngineExpressionVisitor for visit_expression) or individual callbacks (as in visit_metadata). The pattern is the same in every case:

1. You allocate a context (a NullableCvoid you own) that the callbacks can write into. Kernel does not interpret this value.
2. You fill in one callback per node kind. Kernel invokes the callbacks passing the context plus node-specific arguments (names, types, literal values, child handles). For tree-structured inputs (schemas, expressions, predicates), callbacks fire in depth-first order; for flat inputs (e.g. visit_metadata), the callback fires once.
3. When visit_* returns, the context holds your engine-side representation.

Callbacks run synchronously on the same thread that called visit_*. Strings passed to callbacks (KernelStringSlice) are borrowed for the duration of the call; copy them if you need to retain them beyond the callback.

Error handling

Kernel functions that can fail return an ExternResult<T>, which is a tagged union:

// C representation (simplified)
typedef enum { Ok, Err } ExternResultTag;
typedef struct {
    ExternResultTag tag;
    union {
        T ok;
        EngineError* err;
    };
} ExternResult;

You provide an allocate_error callback when creating the engine. Kernel calls this callback to allocate error objects in your memory space whenever an operation fails. Because the engine allocates these errors, the engine is also responsible for freeing them. Kernel returns the error pointer immediately and does not retain it.

The EngineError struct contains a KernelError enum that classifies the error type (e.g., GenericError, FileNotFoundError, InvalidUrlError). The error message string passed to allocate_error is only valid for the duration of the callback, so you must copy it if you need to keep it.

C examples

The repository ships four runnable C examples under ffi/examples/. Each is a complete program that links against delta_kernel_ffi and exercises a different slice of the API.

The high-level flow in the read-table example:

// 1. Create an engine
ExternResultHandleSharedExternEngine engine_res =
    get_default_engine(table_path, allocate_error);

// 2. Build a snapshot
ExternResultHandleMutableFfiSnapshotBuilder builder_res =
    get_snapshot_builder(table_path, engine);
ExternResultHandleSharedSnapshot snap_res =
    snapshot_builder_build(builder);

// 3. Create a scan
ExternResultHandleSharedScan scan_res =
    scan(snap, engine, NULL, NULL);

// 4. Iterate over scan metadata and read data
// ... (see the full example for details)

// 5. Clean up
free_scan(the_scan);
free_snapshot(snap);
free_engine(engine);

What’s next

• Building a Connector: implementing a custom engine in Rust
• The Engine Trait: understanding the engine interface that the FFI wraps