Metadata-Version: 2.5
Name: litelink
Version: 0.10.0
Summary: An embedded Iceberg storage engine for append-only data.
Project-URL: Source, https://github.com/nhobin219/litelink
Project-URL: Specification, https://github.com/nhobin219/litelink/blob/main/docs/SPEC.md
Author-email: Nico Bautista Hobin <nico@penca.io>
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: append-only,capture,iceberg,parquet,sqlite
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Database :: Database Engines/Servers
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.11
Requires-Dist: duckdb<1.5.6,>=1.5.5
Requires-Dist: pyarrow>=17.0.0
Requires-Dist: pyiceberg[pyarrow,sql-sqlite]<0.13,>=0.12.0
Requires-Dist: xxhash>=3.6.0
Description-Content-Type: text/markdown

<p align="center">
  <img src="docs/assets/litelink-logo.svg" alt="litelink" width="330">
</p>

[![CI](https://github.com/nhobin219/litelink/actions/workflows/ci.yml/badge.svg)](https://github.com/nhobin219/litelink/actions/workflows/ci.yml)
[![license](https://img.shields.io/badge/license-Apache%20v2-blue)](LICENSE)
[![Python](https://img.shields.io/badge/python-3.11%20%7C%203.12%20%7C%203.13%20%7C%203.14-blue)](pyproject.toml)
[![Iceberg](https://img.shields.io/badge/Apache%20Iceberg-v2-4B8BBE)](https://iceberg.apache.org/)

# An embedded Iceberg storage engine for append-only data

|  | DuckDB | litelink |
|---|---|---|
| runs | in your process | in your process |
| without | a server, daemon or cluster | a server, daemon or catalog service |
| owns | the query | the durable write path |
| speaks | SQL over Parquet and Arrow | Iceberg v2, on disk and in object storage |

What DuckDB is to query execution, litelink is to the durable write path. It takes
high-throughput transactional appends and turns them into one well-sized Iceberg table per
log — on S3, or in a local directory. `append()` commits to a SQLite buffer and returns once
the row is durable. Behind it, the library seals rows into sorted Parquet, compacts small files
up to a target size, publishes settled files to that table, and evicts from local disk what it
already holds. Through all of it the log stays one queryable unit: a read sees every row
exactly once, whichever tier holds it, and maintenance runs beside appends rather than in
front of them.

Rows move through three tables: buffer, staging and published. `advance()` runs steps 1–3 to
move them along, then steps 4–10 to clean up what they left behind:

```
  log.append(row) · log.extend(rows)        durable on return: SQLite, synchronous=FULL
                │
                ▼
  ┌───────────────────────────┐
  │ buffer         buffer.db  │   rows wait until a file's worth has arrived
  └───────────────────────────┘
                │  1. seal       writes what the size trigger cut       flush: everything
                ▼
  ┌───────────────────────────┐ ◄──┐
  │ staging    local Iceberg  │    │  2. compact   merges runs of small files
  └───────────────────────────┘ ───┘
                │  3. publish    what compaction is finished with       flush: everything
                ▼
  ┌───────────────────────────┐
  │ published        Iceberg  │   local by default, or s3://
  └───────────────────────────┘

  then cleanup:
     4. evict("buffer")        rows staging holds (published, with wal_replication)
     5. evict("staging")       files published holds, never before (I4)
     6. reclaim("buffer")      VACUUM buffer.db         only with vacuum_free_ratio
     7. reclaim("staging")     expire old snapshots, then delete files past their grace
     8. sweep("staging")       files a lost or crashed commit left behind
     9. reclaim("published")   the same, on the published table
    10. sweep("published")
```

One set of verbs for every table, each taking the table as an argument:

| | `evict`: drop what the next copy holds | `reclaim`: turn that into free disk | `sweep` |
|---|---|---|---|
| `"buffer"` | rows staging holds, or published with `wal_replication` | `VACUUM` | — |
| `"staging"` | files published holds (I4) | expire snapshots, then delete | stranded metadata |
| `"published"` | — (it is the durable copy) | expire snapshots, then delete | stranded metadata |

`reclaim` on staging or published does not delete a file in the call that frees it: the file
waits out its table's snapshot retention first, so a scan already reading it can finish. A
later `reclaim` deletes it.

- **`flush=True` means the same everywhere**: push everything through this stage now,
  regardless of thresholds. `advance(flush=True)` passes it to `seal` and `publish`, so one
  pass leaves nothing buffered and nothing unpublished. Use it at shutdown, not in a loop: it
  leaves undersized files behind.
- **Nothing runs unless you call it.** The library owns no thread. Call `seal()` often (it is
  one indexed read when there is nothing to do) and `advance()` rarely.
- **Each step is a routine of its own** for an orchestrator that wants them on different
  schedules or in different processes; API.md's "Process split" is the recommended layout.
  `advance()` is the one-process version, and raises if another process holds a claim it needs.
- **A publish that fails stops only the published steps.** Steps 4–8 still run, so a machine
  cut off from S3 keeps reclaiming local storage; then the error is raised.

`ingest()` loads data that is already durable, a Parquet corpus say: it skips the buffer and
writes straight into staging at `target_compact_size`, then publishes, flushing its short last
file by default only when the log replicates its WAL. Eviction, reclaiming and
the sweeps are left to the next `advance()`, which a log that only ever ingests still needs.
[SPEC §1](docs/SPEC.md#the-pipeline) diagrams both paths.

**The Iceberg table is the product.** The usual shape is a write path in one system and an
analytical store in another, with a job copying between them. Here they are one log, and the
Parquet a row is sealed into is the Parquet DuckDB, or any other Iceberg engine, reads. litelink
reads across every tier, so **no read on the hot path touches the network**; everything else
reads the published table through its `version-hint.text`, with no catalog and no litelink:

```python
import duckdb
import litelink

# Written, durable on return.
log = litelink.open("data", "trades")
log.append({"trade_id": 624438572, "event_ts": 1787772776240000,
            "price": 78501.62, "amount": 0.0076})

# Read on the same box, across whichever tiers could hold a match. Inside sql(), the log
# is always the relation `log`, whatever it is named.
log.sql("SELECT count(*), max(price) FROM log").read_all()

# Read the published table with any Iceberg engine, and litelink not installed at all —
# from S3, or for a local-only log from data/trades/published/trades.
duckdb.sql("""
    SELECT count(*), max(price)
    FROM iceberg_scan('s3://bucket/prefix/trades',
                      version_name_format = '%s%s.metadata.json')
""")
```

The published table holds what `publish` has pushed, which trails the buffer by the publish
interval; rows newer than that are readable through litelink on the writer's machine.

## Install

```bash
pip install litelink        # or: uv add litelink
```

**Nothing else is required** — no producer, no credentials, no maintainer process, no
container. Object storage and WAL replication are opt-in, and each is one setting; another
machine reads the published table with no litelink at all.

Wheels for Linux and macOS on x86-64 and arm64 carry a checksum-verified litestream and the
DuckDB extensions litelink loads, so a box with no egress still reads, writes and restores.
That costs ~152 MB. Run `python -m litelink` to check a machine before you rely on it; see
[`docs/RUNTIME.md`](docs/RUNTIME.md) for anywhere else.

## API

```python
litelink.new(root, name, *, schema, sort_by=None, config=None, published=None,
             s3_options=None, start_offset=1)                      -> WriteHandle
litelink.open(root, name, *, s3_options=None)                      -> WriteHandle
litelink.open(root, name, *, read_only=True, ...)                  -> LocalReadHandle
litelink.restore(root, name, *, published, s3_options=None, ...)   -> WriteHandle
litelink.validate_row(schema, row)                                 # raises as append would
litelink.preflight(...)                                            # what python -m litelink runs

# Every handle reads:
    log.scan(*, columns=None, where=None, start_offset=None, end_offset=None, published=True)
    log.sql(query, *, published=True)                 # the log is `log`; both stream Arrow
    log.column_statistics(*, tier=None) · log.coverage(*, published=True)   # tier: staging|published|buffer|None
    log.end_offset() · buffered_rows() · staging_rows() · staging_files() · published_through()
    log.schema · sort_by · config · published

# A WriteHandle also writes:
    log.append(row) -> int                          # durable on return
    log.extend(rows) -> list[int]                   # ONE transaction, one fsync
    log.ingest(table_or_reader)                     # Arrow straight to Parquet, then published
    log.advance(*, flush=False)                     # the whole pipeline above, in order
    log.seal(*, flush=False) · compact() · publish(*, flush=False)
    log.evict(table=None) · reclaim(table=None) · sweep(table=None) # its steps: clean up
    log.retire()                                    # end the log: all published, none local
    log.set_config(...)                             # the policy; schema, sort_by, published are fixed
```

The deliberate choices:

- **Handles, not logs.** A read handle has no write methods at all, rather than ones that
  raise, and `open(..., read_only=True)` is typed so misuse is caught before it runs.
- **`new` takes the shape; `open` takes none of it.** Schema, sort order, config and published table
  live in the log, so nothing at the call site can disagree with what is on disk.
- **The library owns no thread.** Nothing seals unless you call `seal()` or `advance()`;
  your loop is the schedule.
- **Which tiers a query reads is decided per query**, from its predicates. A query bounded
  inside the staging window never touches the network, however much has been evicted.

Full reference in [`docs/API.md`](docs/API.md).

## Writing

```python
import litelink
import pyarrow as pa

schema = pa.schema([
    pa.field("trade_id", pa.int64()),
    pa.field("event_ts", pa.int64()),    # microseconds, as the exchange sends them
    pa.field("price", pa.float64()),
    pa.field("amount", pa.float64()),
])

log = litelink.new("data", "trades", schema=schema, sort_by=("event_ts",))

log.append({"trade_id": 624438572, "event_ts": 1787772776240000,
            "price": 78501.62, "amount": 0.0076})     # durable on return
log.extend(group_of_rows)                             # the throughput lever
log.advance()                                        # seal, compact, publish, evict, expire, sweep
```

`extend()` commits the whole group in one transaction, so it is one fsync for the batch
rather than one per row, and that call size is the write-throughput lever. Loading history is
`ingest()`, which writes Arrow straight to Parquet.

## Reading

`sql` exposes the log as `log`; `scan(where=…, columns=…)` is the typed equivalent, and both
return a `pa.RecordBatchReader` rather than a table, so materialising is yours to choose. A
reader can open the same log alongside a live writer with
`litelink.open("data", "trades", read_only=True)`.

**litelink decides which tiers a query reads.** Every query reads the buffer; the staging table
and the published table are read only when they could hold a matching row — for the published
table, a file below the staging table. That is decided from per-column bounds for each tier —
the staging table's from its own Iceberg manifests, the published table's kept in `buffer.db` —
so the decision itself never touches the network:

```python
log.scan(where="event_ts > 1787772000000000")   # recent: local disk only
log.scan(where="event_ts < 1700000000000000")   # history: reads the published table too
log.scan()                                      # the whole log
log.scan(published=False)                       # local disk only, whatever it asks
```

So a query's latency follows its predicates. Bound it on a leading column of `sort_by` and a
recent window stays local; leave it unbounded and it reads every tier, because the whole log
is the right answer. Anything the decision cannot read — an OR, a subquery, a comparison with
something other than a constant — reads the published table rather than risk skipping a
row. The buffer keeps no column statistics, so only an offset bound (`scan(start_offset=…,
end_offset=…)`) can skip it.
`column_statistics(tier=…)` gives every column's bounds and counts without opening a data
file, per tier (`"staging"`, `"published"` below it, `"buffer"`) or for the whole log.

**Reads are cached in memory, and a reader on another machine can also cache on disk.** Every
`duckdb_connection` keeps DuckDB's memory cache on, local reads included (`memory_cache=False`
turns it off). One built with `s3_options` reads S3, and with `disk_cache=True` also caches what
it reads on disk across restarts, under `~/.cache/litelink/<cache_key>` (or `$XDG_CACHE_HOME`),
shared by every process using the key. The disk cache is off by default and never used by a
log's own handles: on the host that writes a log, it would put back on disk exactly what
eviction removed.

```python
options = litelink.S3Options()
con = litelink.duckdb_connection(s3_options=options, disk_cache=True, cache_key="trades-reader")
metadata = litelink.current_metadata("s3://bucket/prefix/trades", s3_options=options)
con.sql(f"SELECT count(*) FROM iceberg_scan('{metadata}')")
```

**With the disk cache, resolve the table with `current_metadata`, per read.** It reads
`version-hint.text` outside DuckDB and returns the current `metadata.json`. Scanning the table
by its directory instead resolves the hint through the cache, whose file handle outlives a new
publish: the scan fails DuckDB's ETag check instead of reading the new snapshot.

## Reading from another machine

litelink reads on the primary: every handle is on the host that holds the log's root. Off that
host, the published table **is** the interface. It is an ordinary Iceberg table that publishes
`version-hint.text` at every commit, so an engine pointed at the prefix resolves the current
metadata itself. No catalog service, no local root, no litelink install:

```python
import duckdb

con = duckdb.connect()
con.execute("CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION 'us-east-1');")

table = con.execute("""
    SELECT count(*), max(litelink_offset)
    FROM iceberg_scan('s3://bucket/prefix/trades',
                      version_name_format = '%s%s.metadata.json')
""").arrow().read_all()
```

Point it at the table DIRECTORY — `<published>/<name>` — not at a metadata JSON.
**`version_name_format` is not optional**: DuckDB defaults to the Hadoop `v%s%s.metadata.json`
while pyiceberg names its metadata `00003-<uuid>.metadata.json`, so the format has to stop
prepending the `v`. `credential_chain` is the ordinary AWS resolution — profile, instance
metadata, SSO; against another endpoint pass `KEY_ID`, `SECRET`, `ENDPOINT` and
`URL_STYLE 'path'` instead. No `INSTALL`/`LOAD` is needed — DuckDB autoloads `iceberg`, `avro`
and `httpfs` when a query names them, and `just bootstrap` provisions them ahead of time so the
first read is not a download.

It reads what the published table holds, which on a quiet stream can lag the writer indefinitely
rather than by the publish interval, because `publish` holds back a trailing run under
`target_compact_size`. Rows still in the primary's buffer or staging table are only readable on
the primary.

`litelink_offset` is monotonic and never reused, so a reader keeps the highest it has seen and
asks for what came after — which is how you poll the published table as it grows.

## Demos and recovery

```bash
just demo-websocket    # a live public feed, one process, ~30 seconds
just demo-capture      # a synthetic feed, driven as hard as you like
just demo-maintain     # in another terminal: seal, compact, publish, evict
just rustfs            # object storage in a container, to publish to S3
just demo-replicate    # ship the SQLite WAL, to survive losing the machine
```

Clone the repo for these; `just bootstrap` sets up the toolchain. Credentials are never
written to the log directory — the library reads them from the environment through the
ordinary AWS chain, so a profile, instance metadata or SSO all work untouched.
`litelink.restore(root, name, published="s3://...")` rebuilds a log on another box, reserving an
offset window so nothing the dead machine served is reissued.

litelink emits the litestream config; your supervisor runs the binary. Full walkthrough in
[`examples/`](examples/) and [`docs/RUNTIME.md`](docs/RUNTIME.md).

**On a KVM guest, run on the `kvm-clock` clocksource before turning on `wal_replication`.**
litestream panics on a single backwards tick of the monotonic clock, and a KVM guest on `tsc`
produces them. The sidecar then crash-loops while logging healthy syncs, and the WAL stops
being replicated without anything saying so. `python -m litelink` warns about it, and
[`docs/API.md`](docs/API.md#the-sidecar-needs-a-monotonic-clock) has a systemd unit that makes
`kvm-clock` stick across reboots.

## On disk

One directory per stream, holding everything that stream owns — and the published prefix
mirrors it, so a stream can be copied, replicated or deleted whole in either tier:

```
data/trades/                     s3://bucket/prefix/trades/
    buffer.db                        _wal/
    catalog.db                           buffer.db/
    published.db                         catalog.db/
    litestream.yml                       published.db/
    data/                            data/
        *.parquet                        *.parquet
        compacted/*.parquet              compacted/*.parquet
        ingested/*.parquet               ingested/*.parquet
    metadata/                        metadata/
        *.metadata.json                  *.metadata.json
        *.avro                           *.avro
                                         version-hint.text
```

Data files sit under the table's own location, so the path an engine reads
(`s3://bucket/prefix/trades`) is the directory that holds both halves of the table. A log with
no S3 published table keeps the same table on local disk, at `data/trades/published/trades/`,
with no `_wal/`. A log written before 0.6 keeps `archive.db` where a new one has `published.db`.

Upgrading a log written by 0.1.0 takes litelink 0.5.1 first: see
[Migrating from 0.1](docs/RUNTIME.md#migrating-from-01).

## What it is not

- **Not a mutable store.** Rows are only appended, never updated or deleted in place, and a
  log's schema, sort order and published table are fixed when it is created. To change any of
  them, `retire()` the log and start a new one where it ended:
  `new(root, "trades-v2", schema=…, sort_by=…, published=…, start_offset=old.end_offset())`.
  Offsets stay dense across the two, and any engine reads both as one sequence.

- **Not an unbounded staging table.** A seal's cost tracks what the table's metadata holds, so
  a log that never runs `advance()` and never evicts gets slower on the write path over time.
  `advance()` arrests the larger factor; a retention, with `publish()` running, bounds the
  rest. Numbers and the reasoning are in [`docs/SPEC.md`](docs/SPEC.md) §13.7.

## Not implemented yet

- **Indexes for point lookups.** A lookup by key scans the tiers, pruned only by min/max
  statistics, so finding one row costs a scan rather than a seek — least on `sort_by`'s leading
  column, where the statistics are tight.

- **Blob fields** — large payloads that bypass the buffer — are specified and unbuilt;
  `binary` columns are carried, for ids and other small values rather than payloads
  ([SPEC](docs/SPEC.md) §15).

## Documentation

- [`docs/API.md`](docs/API.md) — every public call, on one page
- [`docs/SPEC.md`](docs/SPEC.md) — the design, and in places still ahead of the code
- [`docs/RUNTIME.md`](docs/RUNTIME.md) — writer and maintainer, threads, processes, what crosses between them
- [`examples/`](examples/) — the websocket capture, and the synthetic feed with one process per role
- [`benchmarks/`](benchmarks/) — the harness, including what litelink costs over raw SQLite
- [`CONTRIBUTING.md`](CONTRIBUTING.md) — setup, the gates, and what a good PR here looks like
- [`SECURITY.md`](SECURITY.md) — what to report privately, and what is a known limit instead

## Development

```bash
just bootstrap          # uv sync + git hooks + DuckDB extensions + litestream
just check              # lint + format-check + typecheck + tests, same as CI
just --list             # the rest
```

A checkout downloads the DuckDB extensions and litestream that an installed wheel carries, so
a contributor provisions what a user does not. Tooling is uv + ruff +
[ty](https://github.com/astral-sh/ty) + pytest; commits follow
[Conventional Commits](https://www.conventionalcommits.org), enforced by a hook. See
[`CONTRIBUTING.md`](CONTRIBUTING.md).

## License

Apache License 2.0 — see [LICENSE](LICENSE) and [NOTICE](NOTICE).
