DEV Community: Wonder-David Efe

Streaming 16 GB of data on a budget: server-side cursors and parallel workers

Wonder-David Efe — Fri, 24 Jul 2026 08:30:00 +0000

A data processing pipeline that processes 100,000 rows is not the same as one that processes 10 million. At a small scale, a single-threaded Python script does the job. At tens of millions of rows, the first bottleneck you hit is time: the pipeline takes too long to finish. The obvious fix is to run the work in parallel across multiple workers, and that's where the second bottleneck shows up: memory.

Sequential is too slow and naive parallelizing hits memory

A typical pipeline reads from a database, transforms each row in Python, and writes the results back. The first pass is a straight loop against that database: open one connection, SELECT ... FROM records, iterate, transform each row, write it back. It works.

However, at tens of millions of rows, the CPU is mostly idle and it takes days. Every iteration waits on the database or the network. Tens of millions of round-trips of "fetch a row, do a bit of Python, write back" saturate nothing except your patience.

The obvious move is to parallelize. Split the work across N workers, done in roughly 1/N of the time. In Python, the database read can be written in two natural shapes:

# Shape A: fetch everything in the main process, distribute to workers via Pool.map
cur.execute("SELECT * FROM records")
rows = cur.fetchall()
with mp.Pool(8) as pool:
    pool.map(transform, rows)

# Shape B: split by row_id range, each worker fetches its own slice
def worker(lo, hi):
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("SELECT * FROM records WHERE record_id BETWEEN %s AND %s",
                (lo, hi))
    for row in cur.fetchall():
        transform(row)

Both shapes use the default psycopg2 cursor, and that's what makes both fail the same way. The default cursor pulls the entire result set into client memory the moment execute() returns; fetchall() just iterates over what's already been loaded. Shape A OOMs in the main process before any worker spins up. Shape B splits the OOM across N workers: each worker tries to hold its share of the data (still in the gigabytes for large tables), and N processes racing to eat the machine either OOM together or swap so hard that "faster than sequential" becomes slower than sequential.

You've traded a time bottleneck for a memory bottleneck. Neither budget is available.

Keep the results on the server while keeping reads parallel

The problem isn't parallelism, it's how each worker reads. The fix is for the data to be streamed: the result set stays on the Postgres server, and only itersize rows are streamed to the client at a time, on demand, as it iterates. In psycopg2, you get this behavior by using a named cursor.

One knock-on: a named cursor holds an open read transaction on its connection, which means you can't commit writes on the same connection while it's open. This requires each worker to use two connections in practice, one for the streaming read and one for the batched write.

def worker(lo, hi):
    read_conn = get_conn()
    read_conn.set_session(readonly=True)
    cur = read_conn.cursor(name=f"worker_{lo}")   # named = server-side
    cur.itersize = 10_000
    cur.execute("SELECT * FROM records WHERE record_id BETWEEN %s AND %s",
                (lo, hi))
    for row in cur:
        transform(row)                            # only itersize rows in memory

Each worker holds only a few MB of client memory at a time (itersize rows times row width, plus Python overhead), regardless of how big its slice is. The server holds the query state and streams. N workers together hold tens of MB in memory, not the full 16 GB. Parallelism is preserved. The memory problem disappears.

Distributing the chunks: SQL ID ranges

The way you split work across workers matters. Each worker owning a non-overlapping range of the primary key is the simplest coordination-free option. No shared queue, no producer/consumer starvation, no locks.

Divide the row_id space evenly:

lo, hi = get_bounds()               # SELECT MIN(record_id), MAX(record_id)
chunk  = (hi - lo + 1) // workers
jobs   = [
    (lo + i * chunk,
     lo + (i + 1) * chunk - 1 if i < workers - 1 else hi)
    for i in range(workers)
]

with mp.Pool(processes=workers) as pool:
    pool.map(worker, jobs)

The primary key's uniqueness means the ranges never overlap, so workers can process their slices independently without any runtime coordination. The split is done once, at startup, and never revisited. Modulo hashing (WHERE record_id % N = worker_id) also works, but range splits are usually faster because they let the query use an index range scan instead of a full scan with a modulo filter.

With streaming reads on the input side and range partitioning across workers, a 16 GB read-transform-write job fits comfortably on a modest machine. The write side has its own scaling story, covered in Part 3 of this series (CTAS-and-swap).

Architecting lean LLM caching: how to drop a 20M-row table without losing your AI memory

Wonder-David Efe — Sat, 18 Jul 2026 13:28:25 +0000

Any team running an agentic pipeline on a periodically-reloaded dataset caches the output of inputs that repeat across periods to save cost. How you store that cache matters. Storing it in the same table as the data is safe, but makes you unable to wipe the data from the previous period before loading the new one. Keeping both on disk to use the cache can work, but if you're in a storage-constrained environment, you can't keep both on disk and have to find a way to remove the old one before loading the new one while still keeping the cache. The fix is to move the cache into a small, orthogonal lookup table keyed on a stable identifier, so it survives the wipe AND takes almost no disk.

This article walks through that pattern: the discovery that made it possible, the two-table shape, and the one operational gotcha you'll hit if you don't watch for it.

The discovery: your inputs are much smaller than your rows

The move to a lookup table starts with noticing something about your data.

A pipeline can process 20 million rows every period and only pass 60,000 distinct inputs to the LLM. The other 19.94 million rows are duplicates of those 60k values, spread across various dates, event types, geographies. Same key, different context.

Once you see that number, the design writes itself. You don't need to store 20 million enrichment results. You need to store 60,000. The 20 million rows can each reference one of those 60k answers.

Concretely:

Working table (data): 20M rows every 2 weeks. ~16 GB. Gets wiped and reloaded when the next period's data arrives.
Cache table (enrichment_cache): 60k rows. Two columns, the input key (whatever you feed the LLM) and the enriched output. A few MB.

The compression ratio is doing the work here. A 16 GB table you can't afford to keep two copies of becomes a few-MB table you can afford to keep forever. That's what makes wiping the working table safe: the cache is the durable output of every past LLM call, and it's tiny.

The architecture: a four-step sequence

When the next data arrives, the pipeline runs this sequence:

Wipe the previous period's data. The cache is untouched because it lives in a different table entirely, so nothing valuable is lost.
Load the new data. New rows go into the freshly-empty data table via whatever ingest mechanism you use.
Propagate cache hits with SQL. One statement joins data against enrichment_cache on the input key, copying every already-known enrichment onto the new rows.
Call the LLM for cache misses only, and write to both places. Query data for rows still missing enrichment. Those are the genuinely-new inputs. Every LLM response gets written to the row in data AND appended to enrichment_cache so the input never needs another LLM call.

The SQL for step 3 is the shortest and most important:

UPDATE data d
SET    enrichment = c.enrichment
FROM   enrichment_cache c
WHERE  d.input_key = c.input_key
  AND  d.enrichment IS NULL;

While this UPDATE syntax works perfectly on smaller datasets, running a bulk in-place update across 20 million rows in production triggers a massive MVCC performance penalty. Postgres creates millions of dead tuples, causing table bloat and heavy disk I/O serialization. To bypass this bottleneck, we eventually migrated this step from an in-place UPDATE to a CTAS (Create Table As Select) and swap architecture during the initial ingestion phase. Part 3 of this series breaks down the mechanics of that zero-bloat swap pattern.

Engineering trade-offs and one gotcha

Why keep the cache in Postgres rather than an in-memory store like Redis?

Any relational database already gives you durable, transactional, indexed key-value storage for free. Adding a separate in-memory cache is another moving part in the deploy, another failure mode to reason about, and another operational surface (memory limits, eviction, replication). If the cache fits comfortably in a relational table and the lookup is an exact match on a stable key, keep infrastructure count low unless a hard constraint forces you to add pieces.

The indexing gotcha.

The cache table is small, so it's tempting to think the join will be trivially fast. It won't be, not at 20M rows on the working side. Without an index on enrichment_cache.input_key, the join has to hash the entire cache table for each batch. That's fine on 60k rows once, but Postgres's planner may still pick a plan that costs you more than you expect at scale, and the equivalent per-row lookup inside step 4 (SELECT ... WHERE input_key = %s) will be a sequential scan every time.

Fix once, forever:

CREATE UNIQUE INDEX enrichment_cache_key_idx ON enrichment_cache (input_key);

A B-tree index on the lookup column turns every cache read into a sub-millisecond index lookup and lets the planner use an index join for the bulk propagate in step 3. This is the one line that separates "cache works in principle" from "cache is fast enough that you notice the pipeline is done."

The point in one sentence

The cheapest LLM call is the one you don't make. The cheapest byte of disk is the one you don't store.

Move the durable, expensive knowledge (LLM answers) into a small, orthogonal lookup table. Let the big working table remain a disposable window on the current period. The wipe becomes safe, the cache stays warm, and the bill collapses.

How do I answer "what did my data look like last month" in Postgres?

Wonder-David Efe — Thu, 09 Jul 2026 18:06:54 +0000

Sooner or later, someone asks: "What did this customer's plan look like when they signed up last February?" The current row won't tell you that; it's been edited three times since. This is the classic historical/temporal data problem: how do you manage changes to dimensional data over time so you can answer "row as of T" on demand. This article is about the schema pattern that answers that question in one SELECT.

There are several approaches to modeling and solving this problem, and I'll discuss some of those approaches below:

a) Event-driven/event sourcing. This has to do with when the domain naturally consists of discrete meaningful facts, e.g., sensor readings, prescriptions, stock movements, financial transactions. Here the change of records events is the natural state and the database is designed for recording these events. It is a Bad fit when the domain thinks in records (customers, products, contracts) whose changes don't have natural event semantics.

b) Versioning in place (this is what SCD Type 2 refers to, if you want to look it up — SCD explainer). The database table holds every version of every record with valid_from / valid_to denoting the change time; the domain still thinks in records. Getting the records state at a certain point in time requires filtering for versions in the valid from/to window. Great when versions have domain meaning. Bad fit when the app is already built; adding versioning in place means every existing query, view, and join needs a WHERE valid_to IS NULL filter. In greenfield the choice between this and a sibling history table (SCD Type 4) is close, and comes down to whether "one table with a filter convention" or "current and history split" fits the team's mental model.

c) Snapshotting a full copy on some cadence or event. Nightly snapshots into a separate table; or freeze a snapshot into a downstream entity when a business event happens (approval, invoice). This is efficient when the ask is anchored to those specific moments and storage is tight. When asking for arbitrary times ("what did this look like Tuesday at 15:37") and for records that change often, you get the values at snapshot time and miss a lot of the changes in between snapshots.

d) Slowly Changing Dimension (SCD) Type 4. In SCD Type 4, you solve by creating a history table in addition to your main table.

The SQL below is Postgres; the pattern translates to any RDBMS that supports row-level triggers, just with different syntax for the trigger body and the point-in-time query.

So you originally have your main table:

CREATE TABLE records (
    record_id  BIGSERIAL PRIMARY KEY,
    -- your app's existing columns
);

Alongside it, you create a history table that mirrors the same shape plus a few bookkeeping columns:

CREATE TABLE records_history (
    history_id  BIGSERIAL PRIMARY KEY,
    valid_from  TIMESTAMPTZ NOT NULL,
    valid_to    TIMESTAMPTZ,
    operation   TEXT NOT NULL,
    LIKE records INCLUDING DEFAULTS
);
CREATE INDEX ON records_history (record_id, valid_from DESC);

LIKE records INCLUDING DEFAULTS copies every column from records. The history table auto-inherits the schema. Only the currently-valid version of a record has valid_to = NULL; older versions have their valid_to filled in with the moment the next edit closed them out.

Every INSERT, UPDATE, and DELETE on records needs to be mirrored into records_history. A row-level trigger keeps them in sync so the app itself never has to write to two tables:

CREATE FUNCTION records_history_trg() RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP IN ('UPDATE','DELETE') THEN
        UPDATE records_history
           SET valid_to = now()
         WHERE record_id = OLD.record_id AND valid_to IS NULL;
    END IF;
    IF TG_OP = 'DELETE' THEN
        INSERT INTO records_history
        SELECT nextval('records_history_history_id_seq'),
               now(), now(), 'DELETE', OLD.*;
        RETURN OLD;
    END IF;
    INSERT INTO records_history
    SELECT nextval('records_history_history_id_seq'),
           now(), NULL, TG_OP, NEW.*;
    RETURN NEW;
END $$ LANGUAGE plpgsql;

CREATE TRIGGER records_history_trg
AFTER INSERT OR UPDATE OR DELETE ON records
FOR EACH ROW EXECUTE FUNCTION records_history_trg();

On every UPDATE or DELETE the previously-current version has its valid_to stamped to now(). On INSERT or UPDATE a fresh row is written with valid_from = now() and valid_to = NULL; the new current version. On DELETE we also write a tombstone row with valid_from = valid_to = now() so we can tell later that the record ceased to exist at exactly that moment.

With this in place, asking "what did the table look like on date T?" is one query:

SELECT DISTINCT ON (record_id) *
FROM records_history
WHERE valid_from <= :as_of
  AND (valid_to IS NULL OR valid_to > :as_of)
  AND operation != 'DELETE'
ORDER BY record_id, valid_from DESC;

For each record we pick the version whose validity window contains :as_of, and we skip records that were deleted before that moment. The DISTINCT ON (record_id) with ORDER BY record_id, valid_from DESC gives you one row per record, the most recent one that was valid at :as_of which is exactly what a SELECT * FROM records would have returned if you'd run it at that moment.

One thing to handle at install time: if rows existed before the trigger was added, they'll have no history entries yet. Backfill once:

INSERT INTO records_history (valid_from, valid_to, operation, ...)
SELECT now(), NULL, 'BOOTSTRAP', r.*
FROM records r;

Every existing record now has a BOOTSTRAP version starting at install. Point-in-time queries before that return empty (which is honest as you didn't have history yet). Anything on or after is queryable.

Two things to bake into ops before you ship. Schema drift. When you add a column to records, records_history doesn't gain it automatically. LIKE ... INCLUDING DEFAULTS copies the schema at CREATE time, not on future changes. The trigger's NEW.* copy will fail on the column-count mismatch until you ALTER TABLE records_history ADD COLUMN to match. Bake that ALTER into whatever automation runs your migrations so nobody has to remember.

Filesystem for AI Agents: What I Learned Building One

Wonder-David Efe — Thu, 02 Apr 2026 22:08:48 +0000

Filesystem for AI Agents: What I Learned Building One

Most agentic systems, like Claude Code, that run on laptops and servers, interact with files natively through bash. But building an agentic system that allows users to upload and work with files comes with its own limitations that make you unable to store files on the server the agent runs on, and give the agent the bash tool:

The fact that it's exposed to users anywhere — bad actors can get it to run commands that can crash the server or exploit other stuffs, so you want only file operations
Even if you allow only file operations, you can't store every user's files on the server due to storage limits, so you'll have to store files in remote storage like S3 or Azure — but mounting them will make native commands like grep slow, as it has to download the full file first
Even if you had unlimited storage and didn't need mounting, you still need isolation — where the agent cannot access files uploaded by another user, or by the same user in another session

There are other solutions to these problems, but they each come with their own tradeoffs:

VM/sandbox platforms (E2B, Northflank) — spin up an isolated environment per conversation, which solves security and isolation. But they have cold start latency, operational overhead, and cost that compound at scale. You're managing servers again, just indirectly.
S3 mounting (mountpoint-s3, JuiceFS, s3fs) — mount remote object storage as a local filesystem. Grep and similar commands work, but inefficiently — each scan triggers sequential HTTP range requests that essentially download the whole file in chunks. Too slow for agent workloads.
just-bash (Vercel Labs) — a TypeScript reimplementation of bash with a pluggable filesystem backend. Closest to what I wanted, but TypeScript only. My pipeline is Python.
Localsandbox (CoPlane) — Python wrapper around just-bash, which would have solved the language problem. But it bridges Python to just-bash via a Deno runtime, adding a deployment dependency I didn't want in a Celery environment.

I ran into this problem recently while building a legal AI agentic system where users had to upload files for the agent to work with. The solution I needed had to be database-like storage that doesn't need to be spun up and down like a server, but supports native file operations that can be exposed as tools to the agent, with the agent unable to access anything outside its own scoped workspace.

Then I found AgentFS — a filesystem built specifically for AI agents, backed by Turso/SQLite. It provides scoped, isolated storage per user and session, with file operations that can be wired directly as agent tools.

Of the integration options — Python SDK, AgentFS + just-bash, AgentFS + FUSE — I went with the Python SDK. Unlike FUSE, which gives the agent a real mount but leaves the rest of the server exposed, the Python SDK puts you in full control. The agent can only do what you explicitly wire up as a tool. No shell escape, no arbitrary commands, no environment variable leaks. The isolation is in the design, not bolted on afterward.

The trade-off is that you're responsible for the tool surface. The SDK ships with the basics — read, write, list — but search operations were missing. No grep, no find, no wc. For an agent that needs to navigate files without dumping everything into context, those aren't optional. So I built them and raised a PR to have them integrated directly into the SDK.

AgentFS relies on Turso DB for hosted production use. Locally, the pattern already works — one SQLite file per user, each opened independently with full read-write access. But on a production server, you can't manage hundreds of separate database files manually. You need a single server process that can route connections to the right user's database.

Turso Cloud solves part of this — it supports creating thousands of separate databases and even lets you query across them using ATTACH. But attached databases are currently read-only. You can read from multiple user databases in one session, but you can't write to them. For an agentic system where the agent needs to create, modify, and delete files in a user's scoped workspace, read-only access isn't enough.

Turso has confirmed that full read-write ATTACH support is on their roadmap. On the AgentFS side, the open() call goes through a connect() function that can be pointed at a Turso-managed database instead of a local file — so the SDK integration path is straightforward once Turso ships the write support. Until then, full production multi-user AgentFS is blocked on this upstream feature.