deltalite

Streaming, partition-level upsert for Delta Lake tables that replaces delta-rs's SQL MERGE with a bounded-memory merge engine. Memory is bounded by the size of the incoming batch and a few concurrency knobs — never by the size of the target table.

delta-rs stays the storage and protocol layer (transaction log, checkpoints, Parquet encoding, S3 conditional-put commits, conflict resolution). deltalite replaces only the merge execution, and writes output files through a streaming writer that produces the same files and Add-action statistics as delta-rs's RecordBatchWriter without holding a second copy of each file.

import deltalite

table = deltalite.DeltaLiteTable.open("s3://bucket/my_table")
stats = table.upsert(record_batch, primary_keys=["id"], partition_key="day")
print(f"v{stats.version}: +{stats.rows_inserted} / ~{stats.rows_updated}")

Why

delta-rs MERGE executes a DataFusion hash join whose memory scales with the scanned target, and it can deadlock under a bounded memory pool (delta-io/delta-rs#4614). For a large, slowly-changing table merged against a comparatively small batch — the typical incremental-sync shape — that means either OOM risk or a hang.

deltalite takes a different route:

1. Build a primary-key hash set over the (small) source batch.
2. Stream the (large) target one Parquet row group at a time, dropping rows whose key is in the source set.
3. Write survivors plus the source rows into new files.
4. Commit every touched partition in one atomic Delta commit.

Peak memory is bounded by the source batch and the concurrency knobs, not by the table. In validation, resident memory stayed flat from 62k- to 1M-row partitions (~4× below MERGE) at matching write volume, via exact content-based file selection.

Installation

pip install deltalite
# or
uv add deltalite

Prebuilt cp312-abi3 wheels are published for manylinux (2_28) and musllinux on x86_64/aarch64, and macOS on arm64/x86_64. A single wheel works on any CPython 3.12 or newer. No Rust toolchain is needed to install.

Usage

upsert accepts anything with the pyarrow C-stream interface — a pyarrow.Table, a RecordBatch, or a RecordBatchReader:

import pyarrow as pa
import deltalite

table = deltalite.DeltaLiteTable.open(
    "s3://bucket/events",
    storage_options={
        "AWS_REGION": "us-east-1",
        "AWS_ACCESS_KEY_ID": "...",
        "AWS_SECRET_ACCESS_KEY": "...",
    },
)

batch = pa.table({
    "id":  [1, 2, 3],
    "day": ["2026-01-01", "2026-01-01", "2026-01-02"],
    "val": ["a", "b", "c"],
})

stats = table.upsert(
    batch,
    primary_keys=["id"],
    partition_key="day",          # omit for an unpartitioned table
    commit_metadata={"source": "my-sync"},
)

print(stats)
# UpsertStats(version=42, partitions_touched=2, files_added=2, files_removed=1, ...)

Rows in the batch whose primary key already exists are replaced; new keys are inserted. Deletes are not expressed through upsert — it is an insert-or-replace by key. Duplicate primary keys within a single batch are rejected (raising DeltaLiteError) rather than silently double-inserted.

API

DeltaLiteTable

version, table_id, configuration, schema_*, partition_columns, num_files, files and file_uris read the loaded snapshot in memory and do no I/O; call reload() first when the table may have moved. history reads the log.

UpsertStats

Returned by upsert. Counts: version, partitions_touched, files_added, files_removed, files_carried_over, files_probed (candidate files the probe read), files_probe_cached (candidate files the probe decided from the probe cache), rows_updated, rows_inserted, rows_copied, source_rows, null_pk_rows. Per-phase wall-clock timings (milliseconds): ingest_ms (importing the pyarrow source), initial_open_ms (the full snapshot load in DeltaLiteTable.open, reported on the first upsert through a handle and 0 afterwards, so a sum over a handle's upserts counts it once), open_ms (snapshot refreshes around the upsert), relax_ms (nullability-relax check), plan_ms (listing + pruning files), rewrite_ms (reading + rewriting the touched partitions), probe_ms (probing, summed over partitions, part of rewrite_ms), commit_ms (committing to the Delta log), maintenance_ms (checkpoint/log cleanup, non-zero only on checkpoint-boundary commits). columns_relaxed counts non-nullable columns flipped to nullable before the write.

History upsert

history_upsert writes an SCD2 history table: one row per version of a key. Open versions (valid_to_column null) live in the current_partition value of partition_column. A closed version lives in the partition named by its valid_to_column as %Y-%m in UTC, which is what pyarrow.compute.strftime(valid_to, "%Y-%m") returns for a UTC timestamp. The table must be partitioned by exactly partition_column, a string column, or the call raises DeltaLiteUnsupportedTableError.

Per key (key_columns, compared null-safely: a null key part matches a null key part):

1. Rows with the same order_column value collapse to the one with the highest tie_break_column value. On an equal tie-break value, the later row wins.
2. Only rows with an order value strictly greater than the key's open row are kept. Equal rows count as rows_reread, lower rows as rows_backwards.
3. The kept rows form a chain: each closes at the next kept row's valid_from_column, and the newest stays open.
4. When a key keeps a row and has an open row, the open row closes at the first kept row's valid_from_column and moves to its month partition.

The source must not carry valid_to_column or partition_column, since the operation sets them. A null order value or a null valid_from_column raises DeltaLiteError. tie_break_column must have a fixed-width type, such as an integer or a timestamp, and is written only when the table has that column. Closed rows are only appended, so earlier closed versions of a key stay. A source that keeps no row, such as a re-run of the same input, makes no commit and returns committed=False, so a re-run after a crash is safe. That includes the nullability relax: it commits only when the call also writes rows.

Each call reads the key and order columns of every current-partition file that its stats do not rule out, so a file that holds only re-read keys is not rewritten. The probe cache is not used. The knobs are the upsert ones that apply: commit_metadata, max_parallel_files, max_buffered_bytes, max_fetch_bytes, probe_concurrency, read_batch_size, target_file_size, max_row_group_bytes, max_source_bytes, read_ahead_row_groups and the multipart and ranged-read settings. max_parallel_partitions is accepted and has no effect, because only the current partition is rewritten. A commit that finds a newer table version is never rebased onto it, since delta-rs does not check a concurrent commit that only adds files: the call plans again against the new version, up to five times, then raises DeltaLiteCommitConflictError.

HistoryUpsertStats counts rows_in, rows_deduped, rows_reread, rows_backwards, versions_opened, versions_closed (open rows closed), chain_rows_closed (kept rows written closed), rows_copied, files_added, files_removed, files_carried_over, files_probed, bytes_read (stored size of the rewritten files) and bytes_written, with version, committed and the upsert timings.

DeltaLiteTable.key_buckets(data, key_columns, n) returns a pyarrow.UInt32Array with a bucket id in 0..n for every row, for splitting a source that is too large for one call into calls on disjoint keys. The key columns are first cast to the table's types, as history_upsert casts them before it matches keys, so source values that become one table key always share a bucket. Each key is then encoded with arrow-row (null-safe) and hashed with xxHash64 under a fixed seed, then reduced modulo n. The ids are the same in every process that runs one deltalite build, for keys of the same Arrow types. arrow-row may change its encoding between arrow releases, so do not persist the ids.

Compaction

compact plans bins exactly as delta-rs's optimize.compact does, from the loaded log with no storage reads, and commits them the same way (ZSTD level 4, one OPTIMIZE commit, dataChange=false). Each bin is streamed one file and one byte-bounded batch at a time through the upsert's fetch and decode budgets, so memory follows the knobs, not the bin. Output files and row groups are also capped by decoded size, so a bin that decodes past a cap writes more than one file where delta-rs writes one.

Conflicts: when another writer commits first, compaction keeps each bin whose input files are all still live and drops the others (deleting their output), so it never resurrects rows a merge replaced. A concurrent schema, partitioning or protocol change raises DeltaLiteCommitConflictError. Tables the upsert refuses are refused here too (DeltaLiteUnsupportedTableError). The result dict carries deltalite's counters plus delta-rs's numFilesAdded, numFilesRemoved, partitionsOptimized, numBatches, totalConsideredFiles and totalFilesSkipped.

Exceptions

All inherit from DeltaLiteError, so you can catch the base or branch on kind:

Operational knobs

Per-call — keyword arguments to upsert (defaults in parentheses):

Process-global — environment variables, enforced on top of the per-call knobs so that many concurrent upsert threads in one process cannot multiply the budgets:

DELTALITE_PROCESS_MAX_PARALLEL_PARTITIONS (8), DELTALITE_PROCESS_MAX_PARALLEL_FILES (16), DELTALITE_PROCESS_MAX_BUFFERED_BYTES (256 MiB), DELTALITE_PROCESS_MAX_FETCH_BYTES (256 MiB), DELTALITE_MAX_SOURCE_BYTES, DELTALITE_MULTIPART_THRESHOLD_BYTES, DELTALITE_MULTIPART_PART_SIZE_BYTES, DELTALITE_MULTIPART_CONCURRENCY, DELTALITE_RANGE_GET_PART_SIZE_BYTES, DELTALITE_RANGE_GET_CONCURRENCY. The last five are the defaults of the per-call arguments of the same names.

Timeouts. All upserts and compactions of one process share the process-wide budgets above, and their waits are fair. So one call that holds permits and stops making progress can make every other call wait. Three bounds stop that:

Only S3 tables get the request defaults. object_store uses a 30s request timeout, which cuts a 128 MiB checkpoint read or a 64 MiB single PUT on a connection slower than about 4 MB/s.

deltalite.limits_snapshot() returns the free and total permits of each process-wide budget and the number of acquisitions that wait for it: {"partitions": {"available": 8, "total": 8, "waiters": 0}, "files": ..., "buffer_kb": ..., "fetch_kb": ...}. It does no I/O and takes no lock, so a caller can read it while another call of the process hangs. A waiter that needs more permits than are free takes the free ones and waits for the rest, so available can be 0 while a part of the budget belongs to a waiter.

Probe cache. The probe reads a footer and the PK column of every candidate file, so on object storage its cost grows with the number of files in the touched partitions. A Delta data file never changes, so deltalite keeps the 64-bit hashes of each file's PK tuples in one process-wide LRU cache, keyed by table URL and file path. A probe that reads a file and finds no match caches it, and an upsert caches the files it writes once its commit lands. A later probe decides a cached file with no read. Equal keys always give equal hashes, so a cached file that holds a key is never skipped. A hash collision only adds a file to the rewrite, which copies its rows unchanged. A file the commit removes leaves the cache. DELTALITE_PROBE_CACHE_BYTES (128 MiB, 0 turns the cache off) caps the entries, and one entry can use at most an eighth of the cap, so a file with more keys is probed every time. Hashes collected for files still being read or written use at most a quarter of the cap more, so the cache uses at most 1.25 times the cap. deltalite.set_probe_cache_max_bytes(n) changes the cap at run time, deltalite.clear_probe_cache() empties the cache, and deltalite.probe_cache_stats() returns its size and hit counters.

Checkpoint prefetch. Opening or reloading a table reads its latest checkpoint Parquet file whole in one GET and serves the Parquet reader's many small range reads from memory, instead of one round trip per footer, metadata block and column chunk. DELTALITE_CHECKPOINT_PREFETCH_MAX_BYTES (128 MiB) bounds the total checkpoint bytes cached by one table load; larger files or multipart checkpoints that exhaust the budget fall back to range reads, and 0 disables the prefetch. Cached bytes also reserve space from DELTALITE_PROCESS_MAX_BUFFERED_BYTES, so concurrent table opens share the process-wide budget. The bytes are released as soon as the load finishes.

Commit snapshot adoption. After a successful upsert the handle keeps the table state delta-rs derived for the commit it just wrote (the same state every delta-rs operation returns) instead of listing the log and reading that commit back. The next upsert still refreshes before it plans, so commits from other writers are observed exactly as before, and a checkpoint the upsert's own maintenance wrote is adopted on that refresh. DELTALITE_ADOPT_COMMIT_SNAPSHOT=0 restores the post-commit refresh.

Data-file reads. Every Parquet data file is opened with a 64 KiB footer size hint, so a footer that fits arrives in one GET instead of two, and a file the probe found a match in hands its parsed footer to the rewrite, so the file is not opened a second time.

Log requests. Four changes remove LIST requests of _delta_log/ and repeated reads. Each one has a kill switch that restores the earlier requests: set the variable to 0 (false, off and no also work).

The probe is used only while a 404 is proof:

• The handle lists the log when the last LIST is older than 10 minutes, or older than half of delta.logRetentionDuration if that is shorter. Log cleanup removes only commit files older than the retention, so inside this time a missing commit file was never written.
• The handle lists the log when the last load read no commit file (the loaded version is a checkpoint version), when the store gives no ETag, and after its own commit on a checkpoint boundary.
• A log store that does not commit with a create-only put (for example the DynamoDB lock store) keeps the LIST before the commit put.

The probe needs a store that answers a GET or a HEAD with the current state of the object. S3 does this (reads are strongly consistent after a write or a delete), and so do SeaweedFS, the local file system and the in-memory store. Set DELTALITE_PROBE_REFRESH=0 and DELTALITE_OPTIMISTIC_COMMIT_VERSION=0 for a store, or a cache in front of a store, that can answer 404 for an object that exists.

Metrics

deltalite emits via the Rust metrics facade (static labels only): deltalite_upserts_total (outcome, prune_strategy, error_kind), deltalite_upsert_duration_seconds, deltalite_files_{added,removed,carried_over,probed,probe_cached}_total, deltalite_rows_{updated,inserted,copied}_total, deltalite_checkpoint_prefetch_total (outcome), deltalite_refresh_probe_total (outcome: current, new_commits, replaced, unknown, listed, no_anchor), and deltalite_commit_probe_total (outcome).

Compatibility & status

• Built against deltalake (delta-rs) 0.32.x as the storage/protocol layer. Correctness is guaranteed by a differential parity suite that runs the same batch sequences through real delta-rs MERGE and through deltalite.upsert and asserts identical logical content — not by version equality.
• Not supported: tables with deletion vectors or column mapping (detected and raised as DeltaLiteUnsupportedTableError). SCD2 history tables are written with history_upsert, not with upsert.
• deltalite rejects duplicate source primary keys that MERGE silently double-inserts — check pre-existing data if you migrate an existing pipeline.

This package is developed in the PostHog monorepo under rust/deltalite/. Issues and source live there.