deltacat-client 0.1.67

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",
)

Operators running catalog/controller operations against one immutable server artifact can opt in to an exact response fence. When enabled, each response on the shared REST request path must carry the exact X-DeltaCAT-Server-Build-Id value, including redirects, error responses, and retries; missing or different values fail without retrying the identity error.

client = Client(
    "https://deltacat-api.example.com",
    bearer_token="your-api-token",
    expected_server_build_id="2.1.0-immutable-build-id",
)

The option defaults to None, preserving the existing client behavior.

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="newadmin@example.com",
    display_name="New Admin",
    email="newadmin@example.com",
    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.

# 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.

Computed fields derive values while a PyArrow read is materialized. They may be supplied for one request or persisted through schema evolution so later reads resolve them automatically. Physical inputs can stay hidden from the result:

import pyarrow.compute as pc

from deltacat_client import Client, ComputedField

client = Client("http://localhost:8080")
action = pc.field("action:action")

steps = client.catalog.read(
    namespace="robotics",
    table="training_steps",
    read_as="pyarrow",
    columns=["episode_id", "step_index"],
    computed_fields=[
        ComputedField(
            name="action_42",
            expression=pc.list_element(action, 42),
            source_fields=("action:action",),
        ),
    ],
)

# episode_id, step_index, and the extracted float32 action_42 are returned;
# the fixed_size_list action:action input is not.
print(steps.schema)

See Computed Fields for the PackDS range-extraction schema-evolution example, expression composition, projection and plan-reuse semantics, validation, performance, and current limits.

describe_table reads only the table, table-version and stream metadata by default, whatever the table's size. The partition-derived counters (total_records, total_bytes, total_files, delta_count, partition_count) and event_time_watermark are None, and stats_omitted["reason"] is "not_requested":

description = client.catalog.describe_table(
    namespace="iris",
    table="packds_steps",
    table_version="1",
)
schema = description.schema

Pass include_stats=True to fold the counters from the partition-stats system table. This still enumerates no partitions, deltas or manifests and is bounded; description.stats_source is "partition_stats". The stats rows are maintained by a post-commit background task, so these counters are eventually consistent with commits (sub-second lag in practice), are not pinned to the request's transaction, and can undercount when a partition has no stats row (only the no-rows case is detected). When the stats cannot be trusted the counters stay None and stats_omitted["reason"] says why ("partition_stats_unavailable", "partition_stats_dirty", "too_many_partitions"). Code that fences on the counters must treat an omission as unknown rather than as zero:

description = client.catalog.describe_table(
    namespace="iris", table="packds_steps", include_stats=True
)
if description.stats_omitted is not None:
    raise RuntimeError(f"no trusted counters: {description.stats_omitted}")
frontier = (description.stream_id, description.delta_count, description.total_records)

Pass exact_stats=True for exact, snapshot-pinned counters (stats_source == "exact"). That reads one metafile per partition and per delta, is unbounded, and is the only mode that derives event_time_watermark; use it for before/after fences on small sealed tables or for batch callers prepared to wait. A failed walk reports stats_omitted["reason"] == "exact_stats_failed". metadata_only=True remains accepted as an explicit spelling of the default.

query_partition_stats requires catalog ADMIN and is a best-effort diagnostic surface for investigating hot, dirty, or growing partitions. It is not a reader-facing partition-discovery API and must not be used to construct training-read membership.

# Administrative diagnostics only.
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:
        print(row)
    token = res.next_token
    if not token:
        break

PackDS _steps data reads must instead use plan/read with a bounded episode_id partition predicate derived from the application's authenticated membership or source frontier. Do not infer a complete read set from this best-effort statistics page.

An ordinary unfiltered current-snapshot plan(..., include_files=False) request without a manifest-identity or partition-frontier attestation 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. The fast path proves one complete table-scoped statistics page and one complete bounded partition page, requires the runtime-obligation scope to be empty, and matches the same exact (stream_id, partition_id) cohort. Every accepted row must be a full baseline built in one catalog snapshot no later than the reader's MVCC snapshot; delta-updated, future-snapshot, legacy, or unversioned rows are refused. Runtime baseline repair examines at most 1,000 visible deltas per partition; a larger or indeterminate history remains refused. It never turns an oversized or indeterminate metadata request into an unbounded partition/file enumeration. Explicit attestation modes use separate contracts and may resolve the complete snapshot; only a supplied partition-frontier operation makes that acquisition checkpointed.

# Metadata-only full-table stats, no per-file resolution. PackDS requires the
# explicit full-scan override even when include_files=False.
plan = client.catalog.plan(
    namespace="robotics",
    table="steps",
    include_files=False,
    allow_full_packds_scan=True,
)

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")

A file-enumerating request can legitimately resolve no files, for example after removing every selected episode. The SDK retains include_files=True in the plan's serializable request context, so read returns a schema-aligned empty result instead of treating it as metadata-only. Explicit metadata-only requests still cannot be executed as data reads. Legacy plans with snapshot/layout metadata but no recorded file-enumeration intent retain that refusal.

Long-running training brokers should use the bounded preparation facade. It deduplicates physical planning while preserving the exact logical episode order and multiplicity, waits for one immutable snapshot, renews server authority, and returns an executable authorized plan:

import uuid

prepared = client.catalog.prepare_packds_episode_plan(
    namespace="robotics",
    table="steps",
    operation_id=f"prepare-{uuid.uuid4().hex}",
    authority_operation_id=f"authority-{uuid.uuid4().hex}",
    episode_ids=["ep_2", "ep_1", "ep_2"],
    as_of=1712697600000000000,
    columns=["episode_id", "step_index", "state", "action"],
    timeout_seconds=300,
    run_id="n2-training-20260712-001",
)
authorized_plan = prepared.composition.plan
assert prepared.composition.requested_episode_ids == ("ep_2", "ep_1", "ep_2")
assert prepared.composition.composed_episode_ids == ("ep_2", "ep_1")
rows = client.catalog.read(
    plan=authorized_plan,
    read_as="pyarrow",
    plan_authority_verification_keys=prepared.plan_authority_verification_keys,
)

Trusted brokers can prepare overlapping consumers with one structural union while retaining exact authority boundaries:

prepared_by_rank = client.catalog.prepare_packds_episode_plans(
    namespace="robotics",
    table="steps",
    operation_id=f"prepare-bundle-{uuid.uuid4().hex}",
    authority_operation_id=f"authority-bundle-{uuid.uuid4().hex}",
    episode_id_groups={
        "rank-0000": ["ep_2", "ep_1", "ep_2"],
        "rank-0001": ["ep_1", "ep_3"],
    },
    as_of=1712697600000000000,
    columns=["episode_id", "step_index", "state", "action"],
    timeout_seconds=300,
)
rank_zero = prepared_by_rank["rank-0000"]
assert rank_zero.composition.requested_episode_ids == (
    "ep_2",
    "ep_1",
    "ep_2",
)

Each named group receives an independently signed exact authority with a common bundle expiration. The helper never slices or reuses a covering union authority; the entire grouped refresh fails closed if any child cannot be vended, signed, persisted, replayed, or reauthorized.

The caller supplies stable operation IDs so transport retries remain idempotent. run_id is an optional, non-secret provenance identifier sent on every preparation, poll, authority, and verification-key request. Keep it stable across retries for coherent telemetry. It does not alter operation, cache, preparation, or authority identity. Launchers can validate their final Gear ID before submission with deltacat_client.validate_packds_fragment_run_id(...). Pass a threading.Event as cancellation_event for bounded cooperative cancellation. The low-level fragment facade raises UnsupportedFeatureError explicitly for old servers; the bounded high-level facade raises PackDSEpisodePlanPreparationUnsupported without retrying it, so callers can opt into a narrow legacy fallback without treating transient HTTP failures as unsupported servers. Preparation telemetry contains fixed-cardinality timing and retry counts and never embeds episode, fragment, credential, or token values.

The lower-level split/compose primitives remain available for trusted brokers that manage fragment persistence and authority refresh themselves:

from deltacat_client import (
    compose_packds_episode_plan,
    split_packds_episode_plan,
)

plan = client.catalog.plan(
    namespace="robotics",
    table="steps",
    as_of=1712697600000000000,
    filter_predicate={"in": ["episode_id", ["ep_1", "ep_2"]]},
    include_files=True,
)
fragments = split_packds_episode_plan(plan)
cache.update({episode_id: fragment.to_dict() for episode_id, fragment in fragments.items()})
combined = compose_packds_episode_plan([fragments["ep_2"], fragments["ep_1"]])
assert combined.credentials is None

Fragment composition is intentionally limited to complete, non-MOR, nontransactional PackDS episode plans with an explicit as_of. The hashes detect cache corruption; they are not server signatures. The credential-free composed plan is deliberately non-executable. A trusted broker must obtain a fresh Ed25519-signed server authority covering the exact fragment fingerprints, snapshot, roots, bindings, and credentials, then use compose_authorized_packds_episode_plan(...) with a trusted server verification key before executing it.

The metadata-only fast path fails closed on stale, dirty, incomplete, malformed, or over-ceiling partition statistics. It raises a bounded-planning error instead of returning a stale count or silently falling back to unknown-cardinality partition/delta enumeration. Allow bounded runtime maintenance to rebuild the statistics, or narrow the request before retrying.

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.

Retained references written before root_id was recorded remain readable when their recorded file path, exact binding revision, scope and credentials are present in the plan. The reader uses that revision's managed data_root prefix when published, not a broader bucket or a root from the current snapshot. It never invents a missing root ID. Explicit root IDs must still match their published association; missing historical scope or malformed identities produce structured refusals. Obtain a complete plan for the original reference instead of replacing its revision with the current one. New writer contexts still need their own root identity; this compatibility is for retained file references.

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

# Credential-free, process-local counters for attribution/benchmarking. Bytes
# count successful returned range payloads; wait covers the exact authority-
# created filesystem read. Reset only at a quiescent run boundary.
metrics = client.catalog.packds_video_read_metrics(reset=False)

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.

create_placer(...) builds a managed crawler → differ → relay pipeline. Its ExactPathRelaySubscriber copies new and changed files byte-for-byte to the same relative paths beneath a destination root. Unchanged files are skipped, and destination copies are retained when source files are deleted.

Use an existing source discovery-root binding and a destination DataRoot with usage=["relay_target"], configured by your catalog administrator. This example copies a source bucket into a replica prefix and checks for changes nightly:

from deltacat_client import (
    Client,
    PlacementDestination,
    PlacementSource,
    create_placer,
)

client = Client("https://deltacat-api.example.com", bearer_token="your-api-token")
replica = create_placer(
    client,
    sources=[
        PlacementSource(
            name="source",
            uri="s3://source-bucket/",
            binding="source-discovery",
            region="us-east-1",
            reflect_deletions=False,  # Retain copies if source files are deleted.
        ),
    ],
    destinations={
        "source": [
            PlacementDestination(
                name="replica",
                binding="replica-s3",
                uri="s3://replica-bucket/source-bucket/",
            ),
        ],
    },
    schedule_cron="0 0 * * *",
    schedule_timezone="America/Los_Angeles",
    name="nightly-replica",
    run_id="v1",
    include_version_changes=True,  # Also relay timestamp-only changes.
    trigger_now=True,  # Start the initial copy, then check nightly.
)

See Incremental exact-path copies for the full reference, including multiple sources, destination fanout, and version-change relays.

Declare an autonomous crawl → diff → placement → conversion pipeline with one thin-client call. By default each source converges into a shared corpus in the auto-created iris namespace:

• iris.packds_steps — the unified steps table, identity-partitioned by episode_id; conversions contribute episode-scoped ADD/REPLACE partitions here.
• iris.packds_episodes — the rich application episodes table (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.

client.training.create_converter supports these source adapters: XDOF (source_format="xdof" or omitted), Dexmate ("dexmate"), YAM in-house ("gear_yam_raw_v1"), Mecka ("mecka"), canonical DROID ("droid"), canonical ABC LeRobot ("abc_lerobot"), UMI ("umi_aligned_zarr"), and a generic LeRobot v2 file set ("lerobot_v2"). DROID and ABC require their canonical dataset/source and a manual crawl. Unknown selectors fail closed. plugin_source stages the implementation expected by the selected strategy; it does not dynamically register an arbitrary new source_format.

from deltacat_client import Client

client = Client(server_url="https://deltacat-api.example.com", bearer_token="...")
result = client.training.create_converter(
    dataset="my_xdof_source",
    seed_prefixes=["s3://my-source-bucket/episodes/"],
    plugin_source="./xdof_packds_loader.py",
    source_format="xdof",
    crawl_schedule=None,  # validate manual runs before enabling a schedule
)
print(result.pipeline_id, result.crawler_id)

For these file sources, the method returns a PackDSConverterResult with 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); the first run still fires immediately by default (trigger_now=True). Pass dry_run=True to build the declaration without server mutation. 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.

For the complete strategy table, data model, B0/B1/B2 stages, annotation contract, scheduling policy, and retry behavior, see PackDS converter pipelines.

Multiple converters, one unified corpus (multi-writer). Several training 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 graph mutation, the PackDS adapter reads existing server declarations and 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"),
)

client.training.create_converter(...) provides a source-neutral namespace for converter creation. Omitted source/output options retain the existing PackDS pipeline defaults. Creation success means graph installation, not completed conversion. The same converter can be reused for later repairs; no per-episode pipeline is required.

from deltacat_client import Client, ConversionOutput

client = Client("http://localhost:8080", catalog="training")
converter = client.training.create_converter(
    dataset="my_dataset",
    source_format="xdof",
    seed_prefixes=["file:///datasets/my_dataset/"],
    staged_plugin_working_dir="file:///plugins/xdof/",
    output_formats=[ConversionOutput(format="packds", version="6")],
    trigger_now=False,
    dry_run=True,
    allow_duplicate=True,  # Skip server duplicate lookup for this offline preview.
)

See Training converters for the full reference, including source options, reprocessing selected episodes, observing conversion and required copies, and explicit staged-repair promotion.

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, adds subsequent unkeyed rows) and 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.

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: select an episode from application inventory or the converter's rich _episodes table, then rewrite it 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 creates the
#    table with episode_id identity partitioning and no merge keys.
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="training_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. Plans for a terminal _steps name advertise the converter's rich sibling.
#    This metadata does not create or query that table.
plan = client.catalog.plan(
    namespace="robotics",
    table="training_steps",
    filter_predicate={"eq": ["episode_id", "ep_0001"]},
    include_files=False,
)
assert plan.episodes_table == "robotics.training_episodes"
target_id = "ep_0001"
target_len = 3

# 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="training_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="training_steps",
    read_as="pyarrow",
    filter_predicate={"eq": ["episode_id", target_id]},
    columns=["episode_id", "step_index", "task", "annotation"],
))

Larger enrichment loops typically batch over application-owned episode IDs 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 guidance on choosing between PackDS episode rewrites and a normal merge-key table.

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.

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}")

Admins can submit a placement metadata hash baseline through client.jobs.submit_placement_hash_baseline(...), supplying the crawl table and version, placement table, and source binding. It defaults to verify_only=True, apply=False. An apply request must explicitly set both flags and provide the existing server confirmation; the client never supplies confirmation automatically. An active job for the same subject is deduplicated. Submission is not completion: monitor the returned job through client.jobs.

To inspect an existing host-local frontier operation, use client.catalog.get_frontier_progress(table=..., frontier_operation_id=..., frontier_authority_id=...) on a client connected directly to its owning server. This read-only call never starts a scan or retries against a different owner. Unknown totals remain unknown; progress is not proof that a frontier can be adopted or that a repair has completed.

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 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 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 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 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.

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.

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.

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.

Are merge keys, merge order, and physical sort the same thing?

No. Declare each for its own purpose:

Without declared merge-order/event-time keys, ordered upserts use their delta stream order, not a timestamp inferred from a column name. There is an important current limitation: the default merge engine sorts incremental rows by stream position before the schema's merge-order keys. Those keys therefore resolve ties within a position; they do not promise that an earlier delta with a larger event time will defeat a later delta. Do not treat either a timestamp sort scheme or the event-time watermark as a global greatest-event-time-per-key guarantee. See the schema declarations and the shared merge engine.

Which write mode should I choose?

Pass the intended mode explicitly when ordering, upserts or replacement matter.

"auto" is not a synonym for ordered "append": it can create a missing table, uses upserts for a merge-key schema, and uses unordered ADD for an existing unkeyed table. Schema and format restrictions still apply: ADD/APPEND are not substitutes for MERGE on a keyed table, and PackDS layout does not support row merge keys. See staged-write usage, write-mode validation, and repair choices.

Does MERGE mean old files disappear immediately?

No. Merge-on-read reconciles committed deltas for a read without persisting a new compacted image. Compaction materializes the merged result using the shared merge rules; it is a read-cost optimization, not a new winner-selection policy. Use catalog reads/plans that honor merge-on-read requirements rather than concatenating the underlying files and assuming one row per key.

Retention is separate again: it removes eligible replaced partition/stream data. The default retention limits are unlimited unless overridden by inherited configuration. Removing old data limits time travel and rollback; retained metadata does not recreate deleted data or externally referenced media. A successful commit, compaction or progress watermark is not a backup checkpoint. Retention does not consult subscriber checkpoints. Set retention with consumer lag and replay needs in mind; a subscriber watermark alone does not protect data. See merge-on-read and retention and GC.

Why do writes to different keys conflict?

The conflict boundary is metadata, not a row key. Ordered writes to the same partition update its position/revision, so concurrent writers can collide even when their merge keys are disjoint. An unpartitioned table has one internal partition. Bound concurrency per hot partition; choose a partition scheme that spreads independent work while keeping updates for one logical key together. Different partitions can progress independently, but table/stream replacement or shared schema changes can still conflict with them.

Unordered ADD avoids the ordered-position update for an existing partition; it still validates its parent and can conflict with creation or replacement. Switch to ADD only if unordered, non-upsert semantics fit the application. See transaction isolation and the ADD path.

How long should a write session stay open, and what can I retry?

Prepare inputs before staging where possible, then stage, upload and commit promptly. Use one session for one logical commit, retaining its session_id and operation_id. Session creation captures target/schema/storage validation state; it does not reserve the partition's next position. For a standalone session, the server creates the native write transaction during commit. An explicit transaction has its own longer-lived snapshot. Native delta/partition staging and publication checks determine the actual conflict window.

An OPEN session that has not begun a commit does not itself claim a partition revision. Abort unused sessions, but do not assume an abandoned or timed-out commit did nothing. Inspect the structured error and the original session:

• Retryable admission or transport failure: the SDK already performs bounded backoff/status recovery. Retrying the same logical commit must retain the same session, operation ID, entries and commit obligations. Do not change the payload or create a second session while the first commit's outcome is unknown. On existing automatic retry paths, a numeric Retry-After is a minimum delay: client backoff caps and jitter do not shorten it. If it cannot fit the remaining retry or lease budget, the client preserves the bounded failure instead of issuing another request early. The header does not make an otherwise unsafe mutation eligible for replay; ambiguity still requires the original session's recovery path.
• AmbiguousCommitError: recover with client.catalog.get_write_session(session_id) and the original operation ID. A timeout is not evidence of an abort. Follow any server-required recovery rather than blindly replaying a write.
• Definitive failed transaction: concurrent_modification with outcome="aborted_uncommitted", or write_session_transaction_failed_terminal, cannot be repaired by repeating that failed commit. Resolve contention, refresh the inputs/target as needed, and start a new session/transaction with bounded backoff. Do not classify every predecessor mismatch or HTTP 409 as this same outcome.

write_session_commit_quiescence_operator_required preserves an unsealed transaction for operator recovery; it is not permission to replace that transaction. A write_key is also not a durable completion receipt: it may recover an OPEN session, but staging after completion can create a new session. Retain the original session ID across process restarts. See registering prewritten files, transactions, and the write-session protocol.

Registering prewritten files

For files copied from an older table definition, session.entry(...) and session.entry_for(...) accept schema_id and sort_scheme_id. Supply the IDs that describe the actual payload, not the current table defaults. An explicit sort_scheme_id=None or an omitted sort ID records unknown order. Data-bearing writes record the scheme actually applied by the writer; a destination's current scheme cannot certify a prewritten file. Registration preserves historical schema/sort metadata and does not sort or rewrite a copied file. It does not retain referenced external media.

Retain the original session.session_id before commit and recover it with client.catalog.get_write_session(...) after an interruption. An explicit write_key can recover an OPEN session, but staging with that key after commit creates a new session for a later write. Retry the original session with the same operation ID and entries; do not treat a recycled key as proof of completion.

For a destination captured earlier, pass expected_table_id and an explicit table_version to client.catalog.stage(...). The existing table and version must match; create_if_missing cannot create a target for this conditional session. The server retains the condition through replay and checks it inside commit validation, refreshing mutable names so rename/recreate cannot redirect the write. Omission keeps ordinary staging behavior. Use a fully upgraded server and recovery-worker fleet: older servers do not understand the stored condition.