deltacat-client 0.1.21

Lightweight REST client and thin MCP scaffolding for the DeltaCAT API server.

deltacat-client is the primary Python client package for DeltaCAT. It lets you read and write tables, run jobs, and build data pipelines against a DeltaCAT API server without installing the full storage, compute, or server runtime stack.

The client talks to a DeltaCAT server over HTTP. Metadata operations (schema validation, transaction management, compaction) stay server-side, while large data reads and writes go directly between the client and cloud storage using short-lived credentials the server vends on demand.

Overview

The client is organized around a root Client object with resource-oriented subclients:

Subclient	Purpose
client.catalog	Namespaces, tables, read/write, transactions
client.jobs	Job submission, claiming, lifecycle, progress
client.publications	Incremental producers into the DeltaCAT Lakehouse (pipeline root nodes)
client.subscriptions	Incremental consumers from the DeltaCAT Lakehouse (pipeline leaf nodes)
client.transforms	Incrementally transform data within the DeltaCAT Lakehouse (pipeline intermediate nodes)
client.pipelines	Wire publications, transforms, and subscriptions into a connected DAG

Installation

pip install deltacat-client

Optional extras for local data materialization:

pip install "deltacat-client[all]"    # Full client-side read/write stack
pip install "deltacat-client[pandas]" # Pandas DataFrame support
pip install "deltacat-client[polars]" # Polars DataFrame support
pip install "deltacat-client[daft]"   # Daft DataFrame support
pip install "deltacat-client[lance]"  # Lance dataset support
pip install "deltacat-client[mcp]"    # Typed async MCP HTTP client

Getting Started

Before using the client, you need a running DeltaCAT API server. See Server Setup for instructions.

DeltaCAT lets you manage Tables across one or more Catalogs. A Table is a named collection of data files. A Catalog is a named data lake that contains tables. For the full data model, see the DeltaCAT README.

Quick Start

from deltacat_client import Client
import pyarrow as pa

# Connect to a DeltaCAT server
client = Client("http://localhost:8080")

# Write data to a table. The table is created automatically; by default
# the client uses mode="auto" and lets the server infer the file format.
data = pa.table({
    "id": [1, 2, 3],
    "name": ["Cheshire", "Dinah", "Felix"],
    "age": [3, 7, 5],
})
client.catalog.write(data, table="cool_cats")

# Read the data back
df = client.catalog.read(table="cool_cats", read_as="pandas")
print(df)

Core Concepts

Expand the sections below to see examples of core client operations.

Authentication

When connecting to a production server with auth enabled, provide a bearer token:

from deltacat_client import Client

client = Client(
    "https://deltacat-api.example.com",
    bearer_token="your-api-token",
)

Admins can also onboard a new user, grant their initial access, and receive a one-time token to share over a secure out-of-band channel:

from deltacat_client import Client

admin = Client(
    "https://deltacat-api.example.com",
    bearer_token="admin-api-token",
)

created = admin.auth.create_user(
    user_id="[e-mail]",
    display_name="New Admin",
    email="[e-mail]",
    initial_role="ADMIN",
    resource_type="catalog",
    resource_name="*",
    issue_token_label="bootstrap",
    idempotency_key="create-newadmin-001",
)

bootstrap_token = created.token.token
print("Share this token once via a secure secret channel:", bootstrap_token)

The server validates your token and maps it to a user identity for permission checks (read, write, admin). The client automatically includes the token on every request. For data file access, the server vends short-lived STS credentials that the client uses to read from and write to cloud storage directly.

See Configuration for the detailed auth model, bootstrap tokens, and role-based access control.

Reading Data

# Read a table as a PyArrow table (the default format)
arrow_table = client.catalog.read(namespace="robotics", table="episodes")

# Read as Pandas, Polars, or Daft
df = client.catalog.read(namespace="robotics", table="episodes", read_as="pandas")
polars_result = client.catalog.read(namespace="robotics", table="episodes", read_as="polars")
daft_df = client.catalog.read(namespace="robotics", table="episodes", read_as="daft")

# Filter rows and limit results
df = client.catalog.read(
    namespace="robotics",
    table="episodes",
    read_as="pandas",
    filter_predicate={"eq": ["task", "pick_screwdriver"]},
    limit=5000,
)

# Time travel: read the table as it existed at a prior point in time.
# The as_of value is a nanosecond-precision Unix epoch timestamp.
df = client.catalog.read(
    namespace="robotics",
    table="episodes",
    read_as="pandas",
    as_of=1712697600000000000,
)

See Reading and Writing for scalable reads, lazy vs. eager materialization, and format-specific behavior. For PackDS-backed training tables, see Training Data.

Listing / paginating partitions

# Page through a table's partitions (e.g. to pick a PackDS episode partition).
token = None
while True:
    res = client.catalog.query_partition_stats(namespace="xdof", table="my_steps", limit=100, after_token=token)
    for row in res.rows:               # each row: partition id/values + stats (record_count, ...)
        print(row)
    token = res.next_token
    if not token:
        break

PackDS _steps reads must be scoped to a single episode_id partition (pass partition_filter=[<partition_id>] to read/plan); paginating query_partition_stats is how you discover the valid partition ids to scope to.

Fast metadata-only planning (counts, sizes, existence)

plan(..., include_files=False) returns a metadata-only plan: the server skips manifest hydration and file enumeration entirely, answering from catalog partition statistics instead of resolving per-file URIs. It is the right call whenever you do not intend to read row data — existence checks, row/byte counts, content-type discovery, and (for PackDS) episode counts — and it stays fast on tables with thousands of partitions where full file enumeration would otherwise be expensive.

# Metadata-only: O(partitions) catalog stats, no per-file resolution.
plan = client.catalog.plan(namespace="robotics", table="steps", include_files=False)

has_data   = plan.total_records > 0          # existence check
row_count  = plan.total_records              # summed from partition stats
byte_size  = plan.total_bytes
fmt        = plan.content_type
episodes   = plan.episode_count              # PackDS partition (episode) count
assert plan.total_files == 0 and not plan.scan_tasks   # no files were enumerated

A metadata-only plan carries no scan_tasks and no vended credentials, so it cannot be used to read data. When you actually need to materialize rows, request files — include_files=True resolves the manifest into scan_tasks (file URIs, per-file stats, and short-lived credentials) so a downstream read(...) / pl.scan_parquet(...) has something to scan:

# Data read: resolve files (scope PackDS reads to a partition / episode_id).
plan  = client.catalog.plan(
    namespace="robotics", table="steps",
    filter_predicate={"eq": ["episode_id", "ep_1"]},
    include_files=True,
)
steps = client.catalog.read(namespace="robotics", table="steps", plan=plan, read_as="polars")

The metadata-only fast path fails closed on stale or dirty partition statistics (it raises rather than return a stale count) and falls back to exact planning when stats are incomplete, so the numbers it returns are never silently out of date.

Reading PackDS video blobs

A PackDS _steps table stores each camera — and each resolution — as its own video_blob:<key> column. Frames are packed into 16-frame H.264 groups whose leader row carries the byte-range reference (the other 15 rows are empty); a companion video_blob_frame_index:<key> column records each row's position (0..15) within its group. Read blobs through the client: the server vends short-lived, scoped credentials on demand, so you never need direct S3 (or S3-Express) access.

step_index (a real _steps column) is the canonical selection coordinate — group-leader recovery is value-based on it, so selection stays correct under filtering/reordering. Select by step_index= (canonical) or where= (a deltacat filter_predicate dict applied client-side).

# 1. Scope to one episode partition (see "Listing / paginating partitions" for how
#    to discover a partition_id). read(read_as="pyarrow") attaches the scoped
#    storage-authority lease to the returned table.
plan = client.catalog.plan(namespace="xdof", table="my_steps", partition_filter=[partition_id])
steps = client.catalog.read(namespace="xdof", table="my_steps", plan=plan, read_as="pyarrow")

# 2. Discover which camera/resolution blobs the table carries.
#    Bare key = native resolution; the "_320_240" suffix = the downscaled copy.
keys = client.catalog.packds_video_blob_keys(steps)
# ['video_blob:left_camera-images-rgb', 'video_blob:left_camera-images-rgb_320_240', ...]

# 3. Read a SPECIFIC blob for a step by step_index (canonical) or a predicate.
#    The 16-frame group leader is auto-resolved, so any step works. Returns the
#    raw H.264 group slice bytes (no `av` needed):
native = client.catalog.read_packds_video_blob(
    steps, video_blob_key="left_camera-images-rgb", step_index=42, plan=plan)
by_anno = client.catalog.read_packds_video_blob(
    steps, video_blob_key="left_camera-images-rgb",
    where={"eq": ["annotation", "grasp"]}, plan=plan)

# 4. Batch-resolve several cameras across several steps in ONE credential vend
#    -> dict keyed by (row, column):
blobs = client.catalog.resolve_packds_video_blobs(
    steps, video_blob_keys=keys, step_indices=[10, 26, 42], plan=plan)

# 5. The pyarrow accessor reuses ONE plan + fs cache across calls and recovers
#    the lease from the table automatically:
vb = client.catalog.video_blobs(steps)
data   = vb.blob("left_camera-images-rgb", step_index=42)             # bytes
frames = vb.frames("left_camera-images-rgb", step_index=42)           # list[np.ndarray]
many   = vb.resolve(keys, step_indices=[10, 26])                      # dict

High-level episode readers

These do their own episode-scoped read, so the group leader is always present — you do not pre-read the steps table:

# A single frame by ABSOLUTE frame number (frame is offset via min(step_index);
# fail-closed if outside the episode's step range). Decodes -> np.ndarray:
frame = client.catalog.read_packds_video(
    namespace="xdof", table="my_steps", episode_id=ep_id,
    video_blob_key="top_camera-images-rgb", frame=128, decode=True)

# Or by canonical step_index / predicate (returns the 16-frame group bytes):
group = client.catalog.read_packds_video(
    namespace="xdof", table="my_steps", episode_id=ep_id,
    video_blob_key="top_camera-images-rgb", step_index=128)

# Walk the WHOLE episode's video IN ORDER (group leaders only, sorted by
# step_index) to reassemble it — yields each 16-frame group:
for group_bytes in client.catalog.iter_packds_episode_video(
        namespace="xdof", table="my_steps", episode_id=ep_id,
        video_blob_key="top_camera-images-rgb"):
    ...  # concatenate / write out in order

Polars: resolve a column lazily

With polars installed, a packds namespace resolves the group leader for every current row lazily (only the rows you .collect() are resolved). It force-keeps the video_blob_frame_index:<key> companion and fails closed if it was projected away:

df = client.catalog.read(namespace="xdof", table="my_steps", plan=plan, read_as="polars")
resolved = (
    df.lazy()
      .packds.with_resolved_video("top_camera-images-rgb", out_column="rgb")
      .collect()
)  # adds an `rgb` column of raw group-slice bytes

Pass plan= (the scoped read plan) to reuse the credentials it already vended; or omit it and pass namespace=/table= and the client vends a plan for you. The same where= filter_predicate dict has identical semantics server-side (in plan(...), prunes files) and client-side (against a materialized table, scans rows). Decoding needs pip install "deltacat-client[packds]" (pulls av); the raw-bytes path does not.

Converting a dataset to PackDS

Turn any source robotics dataset into the shared PackDS training corpus with a small format plugin + one command. By default every dataset/run converges into one unified corpus in the auto-created iris namespace:

• iris.packds_steps — the unified steps table, identity-partitioned by episode_id; every conversion appends here.
• iris.packds_episodes — its per-episode companion (source_format, embodiment, length, duration, video-codec metadata, …).
• iris.<dataset>_crawl / _diff / _placement / _facts — per---dataset incremental state: one set per source dataset, reused across that source's runs, so re-running only converts what changed.

mp4pack video chunks land in the chunks S3-Express bucket and are referenced by-ref from packds_steps. To pull one dataset back out of the shared table, filter on the episode_id dataset prefix (it's the partition key → prunes files) or the companion's source_format / embodiment. Pass --out-table to carve a separate, isolated corpus instead.

1 — write a format plugin. Subclass FormatLoader (deltacat_io_core.packds_format):

from deltacat_io_core.packds_format import FormatLoader, register_format

@register_format("my_format")
class MyLoader(FormatLoader):
    @property
    def name(self) -> str: return "my_format"
    @property
    def control_freq(self) -> float: return 30.0            # Hz
    def discover_episodes(self, input_path):                # -> list[dict] episode descriptors
        ...
    def load_episode(self, descriptor):                     # -> EpisodeStream of PackDSStep
        ...   # each step carries decoded_video / state / action / annotations

deltacat_io_core/packds_lerobot.py::LeRobotLoader is a complete in-repo example to copy. (The gr00t XDOF loader lives on a gr00t branch and ships as a staged payload module — it is the default --plugin-source below; pass your own to convert a new format.)

2 — create the pipeline (client-only). One thin-client call declares the whole autonomous cascade crawl → diff → place → convert and triggers it from a single root crawler. dataset is a stable slug that names the per-dataset incremental tables (re-running the same source reuses them); plugin_source is YOUR loader module. The iris namespace is created on demand (idempotent), and every run converges into the unified iris.<out_table>_steps / _episodes corpus:

from deltacat_client import Client, create_packds_converter

client = Client(server_url="https://deltacat-api.example.com", bearer_token="...")
result = create_packds_converter(
    client,
    dataset="my_dataset",
    seed_prefixes=["s3://my-source-bucket/my/episodes/"],
    plugin_source="./my_loader.py",   # your FormatLoader, staged as a flat payload module
    crawl_schedule="0 0 * * *",       # cron: re-crawl nightly (default; pass None for manual-only)
)
print(result.pipeline_id, result.triggered_nodes)  # created + the single root trigger fanned out

create_packds_converter returns a PackDSConverterResult carrying the created node ids (crawler_id / diff_id / placement_id / conversion_id / pipeline_id). crawl_schedule is persisted on the crawler as a cron trigger (see Scheduled Processing below) so the schedule scanner re-crawls on the cadence; the first run still fires immediately (trigger_now=True). Pass dry_run=True to print the full plan with no server calls. In prod, pass conversion_cluster_affinity="xlarge" for multi-TB or million+ episode backfills so they can only be claimed by the xlarge conversion cluster. The convert-only operator CLI (python -m deltacat.compute.training.submit_x_to_packds, for an already-crawled facts table) remains available for that narrower case.

Multiple converters, one unified corpus (multi-writer). Several create_packds_converter pipelines may feed the SAME unified iris.<out_table>_steps / _episodes tables — this is the intended end state for consolidating many sources into one queryable corpus. Each converter flags its shared _steps / _episodes sinks as co-writable, and the server only relaxes its single-writer guard when EVERY writer of a table is a mutually-flagged packds converter (a packds converter never silently collides with an unrelated general transform). The per-dataset _facts table stays single-writer.

Duplicate rejection. Before any server call, create_packds_converter rejects creating a TRUE duplicate — a pipeline that feeds the SAME _steps table from the SAME crawl source (same seed-prefix set + crawl binding) — and raises PackDSDuplicatePipelineError naming the offending pipeline / conversion / crawler ids. DISTINCT sources feeding the same unified table is allowed (that is the multi-writer case above). Pass force=True (a.k.a. allow_duplicate=True) to create it anyway; under dry_run=True the finding is printed instead of raised.

Redrive is disabled by default. Because a redrive does a WHOLE-TABLE rollback per declared sink, redriving one converter would reverse EVERY co-writer's contributions to the shared _steps / _episodes tables. The conversion transform is therefore created with redrive disabled: the server rejects client.transforms.redrive(conversion_id) (409 transform_redrive_disabled) and client.pipelines.redrive(pipeline_id) (409 pipeline_redrive_disabled). To replace episodes, drive the change through the SOURCE instead — see PackDS Episode Rewrites below: update the crawled source so the diff emits the affected episode_ids as REPLACE (per-episode steps_write_mode=REPLACE), or delete/deprecate the offending episode-ID partitions. (Proper shared-table redrive is tracked separately.)

Run an existing converter on demand. Re-crawl + reconvert the latest source files whenever you want by running the converter's root crawler publication. Use the crawler_id from the PackDSConverterResult (the cascade fans out from this single trigger — diff → place → convert drain automatically):

run_result = client.publications.run(result.crawler_id)
print(run_result.triggered_nodes)  # the downstream nodes the crawl completion fanned out to

Update the configured crawl schedule. Change the cron cadence of an existing converter's crawler with a new schedule trigger; the server persists the new schedule_cron and the schedule scanner picks up the next fire on the new cadence. A manual publications.run(...) still works alongside the schedule:

from deltacat_client import PublicationTrigger

# Re-crawl every 6 hours instead of nightly (pass cron=None / a different trigger to clear).
client.publications.update(
    result.crawler_id,
    trigger=PublicationTrigger.schedule(cron="0 */6 * * *", timezone="UTC"),
)

Writing Data

The client supports writing PyArrow tables, Pandas DataFrames, Polars DataFrames, Daft DataFrames, NumPy arrays, Ray Datasets, and local files (Parquet, CSV, TSV, PSV, Feather, JSON, ORC, AVRO, Lance).

import pyarrow as pa

# Write data to a table. The defaults are mode="auto" (creates the
# table on first write, appends thereafter) and the client default Parquet format.
data = pa.table({"episode_id": [1, 2], "score": [0.95, 0.87]})
client.catalog.write(data, namespace="robotics", table="predictions")

# Append more data — same call, same defaults.
data2 = pa.table({"episode_id": [3, 4], "score": [0.91, 0.89]})
client.catalog.write(data2, namespace="robotics", table="predictions")

# Write a Pandas DataFrame
import pandas as pd
df = pd.DataFrame({"episode_id": [5], "score": [0.93]})
client.catalog.write(df, namespace="robotics", table="predictions")

# Override the defaults when you need a specific mode or format. Here we
# create a Lance-formatted table with explicit schema, partitioning, and
# sort order.
from deltacat_io_core.schemes import (
    PartitionKey,
    PartitionScheme,
    PartitionTransform,
    SortKey,
    SortOrder,
    SortScheme,
)

client.catalog.create_table(
    namespace="robotics",
    table="scored_episodes",
    schema=pa.schema([
        pa.field("episode_id", pa.int64()),
        pa.field("score", pa.float64()),
        pa.field("episode_day", pa.string()),
    ]),
    partition_scheme=PartitionScheme.of([
        PartitionKey(key=["episode_day"], transform=PartitionTransform.IDENTITY),
    ]),
    sort_scheme=SortScheme.of([
        SortKey(key=["episode_id"], sort_order=SortOrder.DESCENDING),
    ]),
    auto_create_namespace=True,
)

data = pa.table({
    "episode_id": [1, 2],
    "score": [0.95, 0.87],
    "episode_day": ["2025-01-15", "2025-01-15"],
})
client.catalog.write(data, namespace="robotics", table="scored_episodes")

# Evolve a table's schema after creation
from deltacat_io_core.schema_updates import add_field

client.catalog.alter_table(
    namespace="robotics",
    table="predictions",
    schema_updates=[add_field(pa.field("confidence", pa.float64()))],
)

See Reading and Writing for write modes, staged writes, path-based writes with create_if_missing, and the full list of supported formats and schema inputs. For PackDS training data writes, see Training Data.

Live Feature Enrichment

Declare one or more merge keys on the table schema and subsequent writes update existing records in place.

import pyarrow as pa
import deltacat as dc
from deltacat_client import Client

client = Client("http://localhost:8080")

# Fields tagged with is_merge_key=True identify the key columns used to
# reconcile updates.
schema = dc.Schema.of([
    dc.Field.of(pa.field("user_id", pa.int64()), is_merge_key=True),
    dc.Field.of(pa.field("name", pa.string())),
    dc.Field.of(pa.field("age", pa.int32())),
    dc.Field.of(pa.field("job", pa.string())),
])

# First write creates a Lance table with the merge-key schema.
initial = pa.table({
    "user_id": pa.array([1, 2, 3], type=pa.int64()),
    "name": ["Jim", "Dinah", "Bob"],
    "age":  pa.array([30, 28, 45], type=pa.int32()),
    "job":  ["Teacher", "Painter", "Sailor"],
})
client.catalog.write(
    initial,
    namespace="demo",
    table="users",
    format="lance",
    create_if_missing={"schema": schema, "auto_create_namespace": True},
)

# Upsert: user_id 1 + 3 are updated; 4 and 5 are inserted.
upsert = pa.table({
    "user_id": pa.array([1, 3, 4, 5], type=pa.int64()),
    "name": ["Cheshire", "Felix", "Tom", "Simpkin"],
    "age":  pa.array([3, 2, 5, 12], type=pa.int32()),
    "job":  ["Tour Guide", "Drifter", "Housekeeper", "Mouser"],
})
client.catalog.write(
    upsert,
    namespace="demo",
    table="users",
    mode="merge",
    format="lance",
)

print(client.catalog.read(namespace="demo", table="users", read_as="pyarrow"))

# Delete: only the merge-key columns need to be supplied.
client.catalog.write(
    pa.table({"user_id": pa.array([3, 5], type=pa.int64())}),
    namespace="demo",
    table="users",
    mode="delete",
    format="lance",
)

# user_id 3 and 5 are gone; 1, 2, 4 remain.
print(client.catalog.read(namespace="demo", table="users", read_as="pyarrow"))

PackDS v5 tables do not support DeltaCAT merge keys. For canonical PackDS training data, use episode_id identity partitioning and REPLACE_PARTITION. If you need row-level MERGE/DELETE semantics, use a normal DeltaCAT table instead of layout="packds". See Training Data.

PackDS Episode Rewrites

PackDS v5 tables are identity-partitioned by episode_id and reject merge keys. Live feature enrichment runs episode-at-a-time: discover the episodes you want to enrich from the public <table>__episodes companion alias, then rewrite each target episode with mode="replace_partition". The replaced partition is committed atomically; sibling episodes are never touched.

import pyarrow as pa
from deltacat_client import Client

client = Client("http://localhost:8080")

# 1. Initial PackDS write across multiple episodes. DeltaCAT auto-creates
#    the table with episode_id identity partitioning, no merge keys, and
#    the single managed `<table>_info__system` system companion table. Client
#    reads should use the `episodes_table` field from the read plan.
initial = pa.table({
    "episode_id": [
        "ep_0001", "ep_0001", "ep_0001",
        "ep_0002", "ep_0002",
        "ep_0003", "ep_0003", "ep_0003", "ep_0003",
    ],
    "step_index": [0, 1, 2, 0, 1, 0, 1, 2, 3],
    "task": [
        "pick", "pick", "pick",
        "place", "place",
        "stack", "stack", "stack", "stack",
    ],
    "annotation": [None] * 9,
})
client.catalog.write(
    initial,
    namespace="robotics",
    table="steps",
    format="lance",
    create_if_missing={
        "schema_def": [
            {"name": "episode_id", "type": "string"},
            {"name": "step_index", "type": "int64"},
            {"name": "task", "type": "string"},
            {"name": "annotation", "type": "string"},
        ],
        "layout": "packds",
        "default_content_type": "lance",
        "auto_create_namespace": True,
    },
)

# 2. Discover which episodes exist by reading the companion table.
#    Companion rows are maintained after PackDS commits by default, so
#    recently-written rows may be briefly absent while background maintenance
#    catches up.
#    A full companion enumeration must opt in explicitly. Production hot paths
#    should normally use a partition filter or a bounded episode_id predicate.
companion = client.catalog.read(
    namespace="robotics",
    table="steps__episodes",
    read_as="pyarrow",
    columns=["episode_id", "total_frames"],
    allow_full_companion_scan=True,
).sort_by("episode_id")

# Pick the first episode by id. The companion's read order is not
# guaranteed without a sort, so we sort first and pull both the id and
# the row count from the same row.
target_id = companion["episode_id"][0].as_py()
target_len = companion["total_frames"][0].as_py()

# 3. Enrich the target episode end-to-end via REPLACE_PARTITION. No
#    merge keys, no merge-on-read. The replacement is per-partition;
#    sibling episodes are unaffected. Note that ``layout="packds"`` is
#    only required at create time; subsequent writes pick the layout up
#    from the table's stored ``dataset_layout`` property.
enriched = pa.table({
    "episode_id": [target_id] * target_len,
    "step_index": list(range(target_len)),
    "task": ["pick"] * target_len,
    "annotation": [f"auto-labeled-step-{i}" for i in range(target_len)],
})
client.catalog.write(
    enriched,
    namespace="robotics",
    table="steps",
    mode="replace_partition",
    format="lance",
)

# 4. Scoped read of the enriched episode. PackDS reads must be scoped
#    to a partition filter or a bounded set of episode_ids (equality / IN,
#    capped by ``DELTACAT_PACKDS_EPISODE_ID_FILTER_MAX_FANOUT``, default
#    1024); unscoped full-table PackDS reads are rejected by default and
#    require ``allow_full_packds_scan=True`` (or
#    ``DELTACAT_ALLOW_FULL_PACKDS_SCAN=true`` server-side) to opt in.
print(client.catalog.read(
    namespace="robotics",
    table="steps",
    read_as="pyarrow",
    filter_predicate={"eq": ["episode_id", target_id]},
    columns=["episode_id", "step_index", "task", "annotation"],
))

Larger enrichment loops typically batch over filtered companion reads (e.g. filter_predicate={"in": ["episode_id", batch_ids]} within the server fanout cap, or partition-filter pagination for larger batches) so each replace_partition write touches exactly one episode while the worker pool processes many in parallel. Because each episode is its own DeltaCAT partition, concurrent replace-partition writes to disjoint episodes do not contend.

See Training Data for the companion-table schema and guidance on choosing between PackDS episode rewrites and a normal merge-key table.

Transactions

Transactions provide atomic multi-step operations with automatic heartbeating and rollback on failure.

import pyarrow as pa

with client.transaction(commit_message="Backfill predictions") as tx:
    client.catalog.write(
        pa.table({"episode_id": [10, 11], "score": [0.88, 0.92]}),
        namespace="robotics",
        table="predictions",
        mode="add",
    )

    # Reads within the transaction see uncommitted writes
    df = client.catalog.read(
        namespace="robotics",
        table="predictions",
        read_as="pandas",
    )
    print(f"Rows visible in transaction: {len(df)}")
# Transaction commits automatically on exit; aborts on exception

See Transactions for time-travel reads, manual commit/abort, and transaction rules.

Jobs

DeltaCAT uses a durable job system for background work (compaction, data relay, subscription processing). The client can submit, monitor, and execute jobs.

# List all jobs
jobs = client.jobs.list()
for job in jobs:
    print(f"{job.job_id}: {job.state}")

# Submit a compaction job
result = client.jobs.submit_compaction(table="predictions")
print(f"Submitted: {result.job_id}")

# Wait for it to complete
status = client.jobs.wait(result.job_id, timeout_seconds=120)
print(f"Final state: {status.state}")

Workers claim and execute jobs. Any process can act as a worker:

# Claim the next pending job matching our worker tags
job = client.jobs.claim(worker_tags=["subscriber"])
if job:
    print(f"Claimed {job.job_id} (type: {job.context.get('job_type')})")

    # Report fine-grained progress and declare the deadline for the next chunk
    client.jobs.emit_event(
        job,
        event_name="batch_started",
        completed=1,
        expected=4,
        metadata={"batch": 1},
        heartbeat_timeout_seconds=300,
    )

    # Do the work...

    # Mark complete. Source-consuming jobs must include the advanced
    # watermark so the server can persist progress correctly.
    client.jobs.complete(
        job,
        records_processed=1000,
        watermark={
            "partition_watermarks": {"analytics.events": 42},
            "known_partitions": ["analytics.events"],
        },
    )

See Jobs and Workers for job types, worker routing, heartbeat rules, retry semantics, and dispatch modes.

Managed Ray Example

For subscription, transform, and publication jobs, dispatch_mode="ray" submits the work onto DeltaCAT-managed Ray clusters instead of the shared custom-worker claim pool. For large prebuilt environments, prefer docker.image as the primary runtime artifact and use payload only for small code/config overlays. The server still vends the DeltaCAT runtime from its promoted runtime manifest so the remote cluster gets compatible deltacat, deltacat_client, and deltacat_io_core packages.

from deltacat_client import Client, DispatchMode, SubscriberType
from deltacat_io_core.triggers import SubscriptionTrigger

client = Client("http://localhost:8080", bearer_token="token")

payload = client.jobs.stage_ray_payload(
    py_modules=["./shared_logic"],
)

ray_dispatch_yaml = """
cluster_shutdown_policy: terminate
docker:
  image: "registry.example.com/ml/runtime@sha256:0123456789abcdef"
payload:
  py_modules:
    - {py_module}
head_node_type: ray.head.default
available_node_types:
  ray.head.default:
    node_config:
      InstanceType: m7i.2xlarge
  ray.worker.default:
    min_workers: 1
    max_workers: 4
    node_config:
      InstanceType: m7i.2xlarge
""".format(
    py_module=payload.payload["py_modules"][0],
)

client.subscriptions.create(
    subscriber_id="episode_processor",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    subscriber_type=SubscriberType.CUSTOM,
    dispatch_mode=DispatchMode.RAY,
    dispatch_config=ray_dispatch_yaml,
    trigger=SubscriptionTrigger.schedule(interval_seconds=300),
)

The scheduler now auto-triggers the subscription every five minutes, DeltaCAT auto-provisions the Ray cluster when a run starts, and cluster_shutdown_policy: terminate tears the cluster back down after each run by default. If the deployment advertises a compatible image profile, DeltaCAT will execute inside the requested docker.image while keeping the promoted DeltaCAT runtime bundle as the compatibility layer.

For reference, the Ray launcher YAML above is:

cluster_shutdown_policy: terminate
docker:
  image: registry.example.com/ml/runtime@sha256:0123456789abcdef
payload:
  py_modules:
    - s3://.../shared_logic.tar.gz
head_node_type: ray.head.default
available_node_types:
  ray.head.default:
    node_config:
      InstanceType: m7i.2xlarge
  ray.worker.default:
    min_workers: 1
    max_workers: 4
    node_config:
      InstanceType: m7i.2xlarge

Use payload for small overlays only:

• experiment modules
• config bundles
• supplemental wheels

Do not treat payload as the primary transport for a large application image.

After launch, owner/admin callers can inspect cluster metadata and then use the managed-Ray access helpers against a specific job:

# `access` describes the currently discovered cluster shape for the job:
# region, cluster name, head instance id/private IP, worker ids, and the
# launch template metadata DeltaCAT used for the cluster.
access = client.jobs.get_managed_ray_access("job-id")
print(access.head_instance_id, access.head_private_ip)

# By default the SSM session starts on the Ray head node.
# The AWS CLI / Session Manager plugin will open an interactive session in
# your terminal, and the helper asks SSM to launch `bash -l` there so you land
# directly in a login shell on the instance.
client.jobs.start_managed_ray_ssm_session("job-id")

# For non-interactive inspection across the whole cluster, use send-command.
client.jobs.send_managed_ray_ssm_command("job-id", commands=["hostname"], target="all")

Publications

Publications are incremental producers that write new data into the DeltaCAT Lakehouse. They sit at the root of a pipeline DAG and can be triggered manually or fired by an upstream event.

# Create a publication that writes to a sink table
client.publications.create(
    publication_id="episode_publisher",
    name="Episode Publisher",
    sink_tables=[{"namespace": "robotics", "table": "clean_episodes"}],
    dispatch_mode="local",
)

# Run the publication
result = client.publications.run("episode_publisher")
print(f"Published: {result}")

See Pipelines for publication configuration and DAG construction.

Subscriptions

Subscriptions are incremental consumers of DeltaCAT tables. They sit at the leaves of a pipeline DAG, tracking a per-partition watermark so each run picks up only new data.

# Create a subscription that watches for new data in "raw_episodes"
client.subscriptions.create(
    subscriber_id="episode_processor",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    subscriber_type="custom",
    dispatch_mode="custom",
)

# Trigger processing (dispatches a job to a subscriber worker)
client.subscriptions.trigger("episode_processor")

# Check watermark state
wm = client.subscriptions.get_watermark("episode_processor")
print(f"Watermark: {wm.watermark}")

# Pause / resume / delete
client.subscriptions.pause("episode_processor")
client.subscriptions.resume("episode_processor")
client.subscriptions.delete("episode_processor")

See Pipelines for subscription modes (delta vs. version), triggers, and redrive.

Transforms

Transforms are the intermediate nodes of a pipeline DAG. Each transform reads from one or more source tables, applies processing logic, and writes to one or more sink tables.

# Create a transform: raw_episodes -> clean_episodes
client.transforms.create(
    transform_id="episode_cleaner",
    name="Episode Cleaner",
    source_tables=[{"namespace": "robotics", "table": "raw_episodes"}],
    sink_tables=[{"namespace": "robotics", "table": "clean_episodes"}],
    dispatch_mode="custom",
)

# Trigger transform processing
client.subscriptions.trigger("episode_cleaner")

# Pause / resume
client.transforms.pause("episode_cleaner")
client.transforms.resume("episode_cleaner")

See Pipelines for transform configuration, redrive, and rollback.

Pipelines

Pipelines wire publications, transforms, and subscriptions into a connected DAG. When an upstream node completes, downstream nodes are triggered automatically.

# First, create connected pipeline nodes
client.publications.create(
    publication_id="ingest_pub",
    name="Raw Ingest Publisher",
    sink_tables=[{"namespace": "robotics", "table": "raw_data"}],
    dispatch_mode="local",
)
client.transforms.create(
    transform_id="clean",
    name="Data Cleaner",
    source_tables=[{"namespace": "robotics", "table": "raw_data"}],
    sink_tables=[{"namespace": "robotics", "table": "clean_data"}],
    dispatch_mode="custom",
)
client.subscriptions.create(
    subscriber_id="consume_clean",
    source_tables=[{"namespace": "robotics", "table": "clean_data"}],
    subscriber_type="custom",
    dispatch_mode="custom",
)

# Preview the connected pipeline from an interior seed node
preview = client.pipelines.discover(seed_node_ids=["clean"])
print(preview.execution_order)

# Option A: persist exactly the previewed node_ids
client.pipelines.create(
    pipeline_id="etl_pipeline_pinned",
    name="ETL Pipeline (Pinned)",
    node_ids=preview.node_ids,
)

# Option B: create directly from seed node ids
client.pipelines.create(
    pipeline_id="etl_pipeline_seeded",
    name="ETL Pipeline (Seeded)",
    seed_node_ids=["clean"],
)

# Check pipeline status
status = client.pipelines.status("etl_pipeline_seeded")

# Pause / resume all nodes at once
client.pipelines.pause("etl_pipeline_seeded")
client.pipelines.resume("etl_pipeline_seeded")

See Pipelines for DAG construction, discovery semantics, redrive, rollback, and stored-order validation.

Data Placement and Replication

DeltaCAT catalogs can span multiple storage backends (S3, SwiftStack, Lustre). Data placement lets you replicate tables across roots so readers access data from the closest location.

# List available data roots for the catalog. The keys of this dict
# are the valid values to pass as `roots=` and `root=` below.
available_roots = client.catalog.list_data_roots("default")
for root_name, root_info in available_roots.items():
    print(root_name, root_info)

# Example output (dict[str, DataRootInfoSummary]):
# aws_s3_iad       {'root': 's3://bucket-iad/',        'storage_type': 's3',
#                   'region': 'us-east-1', 'endpoint_url': None}
# aws_s3_pdx       {'root': 's3://bucket-pdx/',        'storage_type': 's3',
#                   'region': 'us-west-2', 'endpoint_url': None}
# swiftstack_pdx   {'root': 's3://swiftstack-bucket/', 'storage_type': 's3',
#                   'region': None,        'endpoint_url': 'https://pdx.swiftstack.example.com'}

# Pick a target root from what the catalog actually advertises.
# In production you'd choose based on region, storage class, or policy.
if not available_roots:
    print("This catalog uses a single default root; no placement root to select.")
else:
    target_root = next(iter(available_roots))

    # Place a table on an additional storage root for replication
    client.catalog.place(
        namespace="robotics",
        table="episodes",
        roots=[target_root],         # Replicate to this root
        backfill=True,               # Copy existing data too
    )

    # Check replication status
    status = client.catalog.replication_status(namespace="robotics", table="episodes")
    print(f"Roots: {status}")

    # Read from the closest root (server resolves automatically)
    df = client.catalog.read(
        namespace="robotics",
        table="episodes",
        read_as="pandas",
        root=target_root,            # Prefer this root for file paths
    )

    # Remove a replication target
    client.catalog.unplace(namespace="robotics", table="episodes", root=target_root)

New writes are automatically replicated to all placed roots via a background subscriber. Reads with root= get file paths rewritten through the preferred root when data is available there.

See Configuration for data root setup and multi-root catalog configuration.

Discovery Root Administration

Catalog admins can publish read-only discovery roots for system crawler, GC, relay, replication, and placement jobs. Discovery roots are separate from managed data roots, so they can describe authority roots such as s3:// without becoming valid table write targets.

client.catalog.create_storage_binding(
    "default",
    name="swiftstack_pdx_root",
    storage_type="s3",
    storage_class="discovery",
    uri_scheme="s3",
    endpoint_url="https://pdx.swiftstack.example.com",
    region="us-west-2",
    credential_source={"kind": "aws_profile", "profile": "pdx"},
)

client.catalog.create_discovery_root(
    "default",
    name="swiftstack_pdx",
    binding_name="swiftstack_pdx_root",
    root_uri="s3://",
    authorization={"principals": ["service:system-crawler"]},
)

roots = client.catalog.list_discovery_roots(
    "default",
    include_disabled=True,
    include_history=True,
)
details = client.catalog.describe_discovery_root(
    "default",
    "swiftstack_pdx",
    include_history=True,
)

# Updates create a new immutable discovery-root version.
client.catalog.update_discovery_root(
    "default",
    "swiftstack_pdx",
    binding_name="swiftstack_pdx_root",
    root_uri="s3://",
    description="PDX authority root",
    authorization={"principals": ["service:system-crawler"]},
)

# Deletes are logical disables and remain visible with include_disabled=True.
client.catalog.delete_discovery_root("default", "swiftstack_pdx")

Discovery-root detail responses include the effective binding revision, endpoint, region, authorization summary, lifecycle state, sanitized credential-source metadata, and effective storage snapshot.

Scheduled Processing

Subscriptions and transforms can run on a schedule instead of being triggered manually. DeltaCAT supports interval-based and cron-based scheduling.

# Process new data every 5 minutes
client.subscriptions.create(
    subscriber_id="metrics_ingester",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "schedule", "schedule": {"interval_seconds": 300}},
)

# Process at 2am UTC daily using a cron expression
client.subscriptions.create(
    subscriber_id="nightly_aggregator",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "schedule", "schedule": {"cron": "0 2 * * *", "timezone": "UTC"}},
)

# Event-driven: only runs when triggered manually or by an upstream pipeline node
client.subscriptions.create(
    subscriber_id="on_demand_processor",
    source_tables=[{"namespace": "telemetry", "table": "raw_metrics"}],
    subscriber_type="custom",
    dispatch_mode="custom",
    trigger={"mode": "event"},
)

See Configuration for trigger options and scheduling details.

Agentic Access (MCP)

DeltaCAT ships with a built-in Model Context Protocol server so AI agents (for example, Claude Code) can browse catalogs, inspect schemas, plan reads, and stage writes through natural language instead of hand-written Python.

Most hand-written code should just use the REST Client(...) shown above. Reach for MCP when you want to:

• let an agent explore and operate on your catalog conversationally
• embed DeltaCAT as a tool in an agentic application
• use a typed async Python wrapper over the MCP HTTP surface (deltacat-client[mcp]) from code that is already agentic in shape

See the MCP Server guide for the full tool reference, the typed async client, and recipes for agent-driven catalog workflows.

Server Setup

The DeltaCAT client connects to a DeltaCAT API server. For setup instructions, see:

• Server Setup Guide -- start a local or production server
• REST API Reference -- full REST endpoint documentation
• MCP Server Guide -- agentic access via MCP

Additional Resources

Guide	Description
Reading and Writing	Read plans, write modes, staged writes, supported formats
Training Data	PackDS v5 tables, episodes companion, scoped training reads
Transactions	Transaction lifecycle, time travel, rules and limitations
Jobs and Workers	Job types, claiming, heartbeat, worker routing, authentication
Pipelines	Publications, transforms, subscriptions, DAGs, redrive
Configuration	Auth, dispatch modes, triggers, data placement
Maintainer Workflow	Relationship between REST and MCP, generated bindings, facade updates, and validation guards
Architecture	Package boundary, generated client, development notes
Client Compatibility Runbook	Promotion-time packaged client/server compatibility validation

For the core DeltaCAT data model, storage architecture, and catalog APIs, see the DeltaCAT documentation.