Delta examples

Short, runnable scenarios covering the write/read patterns you’ll reach for most. Each one uses the real API and runs as written (requires pyrula-connectors[delta] and pyarrow).

Write an Arrow table, then read it back

import pyarrow as pa
from pyrula.connectors.delta import DeltaTable

data = pa.table({"id": [1, 2, 3], "amount": [10.0, 20.0, 30.0]})
DeltaTable.write_arrow("/tmp/transactions", data)

table = DeltaTable("/tmp/transactions")
snapshot = table.read_arrow().value   # pyarrow.Table
print(snapshot.num_rows)              # 3

write_arrow creates the table if it does not exist. The default mode="append" means calling it again adds rows; pass mode="overwrite" to replace the contents.

Append vs overwrite with typed models

from pyrula import case, IList
from pyrula.connectors.delta import DeltaTable

@case
class Event:
    id: int
    kind: str

table = DeltaTable.create("/tmp/events", Event).value

# Append two batches; each call bumps the version
table.append(IList([Event(1, "click"), Event(2, "view")]))
table.append(IList([Event(3, "click")]))

# Overwrite replaces all data; the table is now version 3
table.overwrite(IList([Event(99, "reset")]))

rows = table.read(Event).value   # IList([Event(id=99, kind='reset')])

append and overwrite both return Either[WriteReport]. Check result.is_ok() before accessing .value in production code.

Merge (upsert) with MergeReport

from pyrula import case, IList
from pyrula.connectors.delta import DeltaTable

@case
class Order:
    id: int
    status: str

table = DeltaTable.create("/tmp/orders", Order).value
table.append(IList([Order(1, "pending"), Order(2, "pending")]))

incoming = IList([Order(1, "shipped"), Order(3, "new")])

result = (
    table
    .merge(incoming, "target.id = source.id")
    .when_matched_update_all()
    .when_not_matched_insert_all()
    .execute()
)

report = result.value
print(report.rows_inserted)   # 1  (id=3 was new)
print(report.rows_updated)    # 1  (id=1 updated to "shipped")
print(report.version)         # 2

The merge builder chains .when_matched_update_all(), .when_matched_delete(), and .when_not_matched_insert_all() before calling .execute(). The result is Either[MergeReport].

Writing to S3 via storage_options

import pyarrow as pa
from pyrula.connectors.delta import DeltaTable

s3_opts = {
    "AWS_ACCESS_KEY_ID": "...",
    "AWS_SECRET_ACCESS_KEY": "...",
    "AWS_REGION": "us-east-1",
}

data = pa.table({"id": [1, 2], "value": ["a", "b"]})
DeltaTable.write_arrow("s3://my-bucket/warehouse/events", data, storage_options=s3_opts)

table = DeltaTable("s3://my-bucket/warehouse/events", storage_options=s3_opts)
print(table.read_arrow().value.num_rows)   # 2

storage_options accepts a plain dict and passes through to delta-rs. Azure Blob Storage and GCS work the same way with their respective credential keys. See the delta-rs storage docs for the full list of keys per backend.