Selected work

A decade, in one dimension

The same instinct at rising levels of abstraction, latest first. Read up the middle column: an expression, a rule graph, a formal definition, a pipeline config, a whole integration, an arbitrary dataflow program.

Mojart — a declarative pipeline compiler

Clojure · author and project lead · 2026

What it is for. The platform being built around it separates three things: connectors bring raw data in, datasets are the materialised artifacts that get exposed, and models are what a person actually queries. Mojart owns the middle — the language and compiler that turn connector data into exposed datasets, plus the metadata the serving layer binds to.

The goal is that adding or changing a pipeline becomes authoring a spec rather than writing pipeline code, and that changing one afterwards stays cheap and safe. Nearly every piece of machinery below exists to buy one of those two properties:

• Content identity and diff-driven deploy, so an edit costs only what actually changed. This is the one that matters most. If every schema change means a full recompute, change becomes unaffordable, so it stops happening, and the platform ossifies around whatever shape it had on day one. I had spent years working inside a system that had gone that way.
• A per-cell run-ledger, so a failed or partial run resumes instead of starting over — the same cost-of-retry argument that drove the ingestion rewrite below.
• A generational manifest with atomic swap, and coordinated removal, because downstream systems are first-class: the exposed schema must never shift under them without coordination, and nobody should read a half-published dataset.

The person this is ultimately for never touches any of it. They ask a question of their marketing data and get an answer they can trust; content identity is what keeps that answer both current and cheap enough to keep producing.

An author writes a program in EDN: sources, objects, datasets as pipes of steps, destinations. Eleven pure passes lower it into a task DAG that carries a Merkle content-identity on every node and every field. Deploy diffs those identities against the last committed generation manifest, so only what actually changed is rebuilt, and the swap is atomic. At runtime the executor reconciles a per-cell ledger and reproduces only dirty (artifact × partition) cells.

The compiler is pure and recompiles from scratch every time. Nothing drift-prone is stored; the graph is re-derived, and only the fingerprint persists.

Some of the work I'd point at: object identity moved onto native UUIDs after I found the old key concatenation collided on NULL versus empty string; an orderable, content-derived primary key so keyset pagination works over persistent state; and normalising NULLs once, where values are born, instead of scattering coalesces downstream.

I also raised checkpoint write throughput from 2.70 to 10.58 million rows per minute at hundred-million-row scale, by sweeping write-ahead-log budget against merge concurrency — and found that concurrency, not the budget knob everyone assumed, is the dominant lever, and nearly free on memory. The storage engine benchmarked is a colleague's; the measurement and the recommendation are mine.

The runtime is the other half, and it is not a thin one. Every task the compiler emits runs as a circuit in the sense of the DBSP paper: a page operator D, the task's operator ^f, an append I, and a z⁻¹ feedback edge that is the checkpoint. The runner owns the loop; an operator only supplies D and ^f. After every page the checkpoint records the cursor, the output location and a fingerprint of the page domain, so a crash resumes from the last page — and a resume whose domain has changed underneath it (a different row count, a different bucket width) is detected rather than silently continued. Partially written pages are undone by a single monotone sweep before the walk restarts. Four paging axes make this work on different shapes: dense rowid windows for transforms, hash-buckets for aggregates — every row of a key lands in one page, so a per-page GROUP BY equals the global one — branch-concatenated rowids for unions, and encoded-primary-key keysets for the columnar destination that has no stable rowid. Pages are aligned to DuckDB's 122,880-row storage groups because a misaligned page pays for rows it discards. A circuit declares whether it is thread-safe; only those fan out across pages, the rest run sequentially regardless of the configured width.

Two things sit on top of the circuits. A persist step lets an author declare that an intermediate result is worth keeping: it becomes its own task and a materialised, versioned table that downstream work resumes from when anything below it changes, instead of recomputing the program from its sources. And the run-ledger records, per produced cell, not only the definition it was built under (content identity) but the version of the source data it was built from — an opaque token the source reports per partition. A cell is dirty when it is missing, when its definition changed, or when its source data advanced; the third case is how late and restated data gets picked up without a rebuild. The two axes are orthogonal, and the ledger folds the data axis into a freshness version on every published dataset. Detection is per partition; narrowing execution to only the dirty partitions is the next step, not done yet.

Above the pipes sits a model of the data itself. Advertising metrics are not flat columns: ~30 source events explode into 800–1,000+ concrete metrics across variant axes — attribution window, attribution type, cohort period. Mojart's design treats each metric as a point in an n-dimensional tensor: a plain metric like spend is the degenerate rank-0 case, and a coalesce/broadcast rule (union the axes; an absent axis means constant along it) makes arithmetic over mixed-rank operands well-defined, so a ratio of a rank-3 conversion count over rank-0 spend needs no special case. The DSL declares the space generatively — never enumerating combinations — and a fixed-point compilation loop resolves selectors against a column registry that both compile-time derivations and runtime-discovered events write into. On the read side, every published dataset carries its metadata as a Model — typed columns, grain, tensor-valued measures — which the semantic layer binds to a shared vocabulary so one widget can compose several models at whatever grain the user picks, with no producer renaming a physical column. Mojart is the write path; the vocabulary and the binding live above it, deliberately.

The most recent piece (late August 2026) is multi-tenancy: a per-workspace tenant context at the bottom of the namespace graph that routes the metadata-database connection and configuration, so one process or worker serves many workspaces. Mechanism first; the workspace registry that will populate it is deferred and additive.

Status, stated plainly: merged and running end to end in an alpha environment, with two years of a real customer's data through it. Not in production. The launch date has moved three times.

Mu — distributed ingestion

Scala / Spark · designed and built it from scratch, principal author · 2022–2025

What it replaced. Customer data arrived through a Ruby ingestion framework grown inside the Rails monolith over several years, backed by a separate Go service over an Apache Drill cluster whose job was querying flat files — CSV, JSON — because the Ruby side couldn't. Colleagues referred to the pair as one thing: the old Ruby/Drill pipeline. Every new source type meant more bespoke code on the Ruby side and more load on the Drill side.

I started Mu from an empty repository and designed it as the replacement for both: a single general-purpose ingestion layer where a channel is a configuration compiled into a distributed plan, rather than a codepath. Eleven source types, ten processors and three sinks behind one config surface, so adding a source stopped being an engineering project. Mu's rollout is what took Drill's usage down, and once the last workload left it I deleted the cluster and the service in front of it — described further down.

A five-million-row weekly customer ingest took about a day and a half. At that duration the job structurally cannot meet a daily data-availability deadline — the customer's morning arrives before the job does. Mu's launch measurement brought it to about ninety minutes; later optimisation took equivalent volumes under an hour.

The mechanism: batch the dereference step at a thousand rows per request across thirty-two Spark executors instead of looking rows up one at a time; read Parquet and BigQuery natively rather than transcoding everything to CSV first, so filters push down to the source; retry a chunk instead of a whole job. A 500k-row daily job went from an hour to ten minutes on the same change.

I verified it rather than asserting it: I ran a live channel and a clone of it through both pipelines and compared the rendered dashboards before merging. Exact match.

It reached 100% of the platform's customer-data and custom-analytics channels and the old pipeline was retired. It compiles a channel config into a Spark plan across eleven source types, ten processors and three sinks. It is a team system — I'm the largest single contributor, not the only one.

Four years on, it is still the ingestion path. On 31-day production means to August 2026: ~680 million rows a day, peaking at 1.33 billion in a day and 120 million in a single hour; ~8,900 jobs a day, 73% of them inside a six-hour morning window (peak 1,889 in one hour, 7× the hourly mean). The shape is the part that matters: the work arrives as a wave against a next-morning deadline, and the twelve workers meet it at 74% occupancy through that window (all twelve busy 15% of the time, idle 3%), then average about two workers busy — fully idle 39% of the time — for the other eighteen hours. The daily total is a measure of demand, not of the ceiling; the demonstrated envelope is twelve-wide and 120 million rows in an hour.

The hardest single problem it posed arrived in 2024: a customer product feed of 17 million objects a day whose job had a four-hour timeout, and whose dereference step was timing out. I wrote the budget first — 500 partitions at parallelism 8 leaves 3.75 hours, so 34 requests per partition must average under six seconds, against a median of twenty — then found a bimodal 60-second stall in a metrics-reporter thread the team had lived with for seven months, batched the cache reads, and moved the 3.5 MB request payload past the web framework's parser (about 700 ms saved on each of 17,500 requests). Dereference finished in 53 minutes; the job in 1.1 hours — and stayed there: over the next ten months the job's mean time fell from 140 to 54 minutes while the feed grew from 8 to 12 million rows a day (21.7 million on its biggest day), with no run over the four-hour timeout after August. Object sync on the annotation side went from 1,100 to 6,700 objects a second by raising a batch size 50×. I also introduced a memory leak along the way that a colleague caught — it is in the record.

I also benchmarked the per-row validation path itself — deserialise, type-cast every cell, date-parse, write Parquet — against a 4.76 billion row, 2.2 TB source: 52,149,325 rows in 3.2 minutes on 16 cores with an 8 GB heap, roughly 16 million rows a minute with date parsing applied to every row. Knowing the per-row ceiling is what tells you whether a slow job is slow because of the data or because of the plan.

Retiring a query cluster

2024

An Apache Drill cluster, fronted by a small Go service, answered aggregation queries over customer CSVs. By 2024 it was the last thing keeping either alive.

I set removing it as a team objective in September 2024, moved Bing reporting and CSV aggregation in-process onto Polars, deleted the service that October, and drove the Kubernetes teardown of both namespaces the same month. What went with it, read off the manifests at the commit before deletion: a four-replica Drill StatefulSet on a dedicated non-preemptible node pool reserving 16 vCPU and roughly 107 GiB of memory around the clock, plus a two-replica service in front of it. Reserved capacity, not utilisation — that hardware was held whether or not a query arrived.

The arc is the part I would actually point at. That microservice existed because of a recommendation I made in 2019: the postmortem on a failed runtime migration concluded we should carve the CPU-bound work out into a dedicated service. Five and a half years later I deleted it, by making the in-process path faster than the service was. Retiring your own earlier conclusion is cheaper than defending it, and I would rather be the person who does that than the person who was right the first time.

Per-object annotation

Co-designed the architecture, led delivery · 2019–2020

Customers define dimensions as rules over their ad objects; every object then has to carry the right tag, consistently, across a hierarchy. The system doing that had no predictable failure rate and no way to assert its own sanity.

I wrote the v2 specification. Rather than opening with a design, it opens with the properties the system must have — determinism, atomicity, eventual causal consistency, isolation, visibility, checkpointing, fault tolerance — then evaluates three candidate architectures against a failure-mode analysis and picks one. On the execution side I implemented the PostgreSQL-based work executor and the JIT-compiled Ruby evaluator — rules lowered to generated code instead of interpreted per object; a colleague added the C-codegen sibling and later optimised the Ruby path. I built the metadata service that versions the rule graph so a tag can be attributed to a specific rule-graph version.

Six months later I validated the design where it was most likely to break. The architecture put Postgres in the critical path as the work executor — each worker asks the database which objects to claim next — so the question was whether that layer could sustain the rate the design assumed. A targeted check on the component the whole scheme rested on, not an end-to-end measurement of the pipeline.

It could, once one thing was fixed. Toggling a single Postgres planner setting and nothing else took the executor from 2–3,000 to 17–19,000 objects a second. At a larger working set, splitting out one finalisation query and reworking its index cut that query from twenty and a half minutes to 111 seconds. I tracked the degradation curve as the working set grew and held disk health as a controlled variable so a failing disk could not be mistaken for a regression. Written up here.

Years later, as team lead, I set the performance objective for the same engine: after a customer edits a tagging rule, every channel must finish a full re-tag within twelve hours. The twelve hours only means something next to the volume it covers. The engine holds roughly 4.7 billion objects across 217 customer schemas and rewrites about 12% of them every day — some 545 million updates and 37 million inserts, around 6,700 writes a second averaged over the day and far burstier in practice. The largest single channel holds about 440 million; re-tagging it took about an hour at the median and two to five hours at the 90th percentile — inside the ceiling, on the whole pipeline: rule evaluation, tagging and persistence to every store, not just the executor.

It was met in September 2024 and beaten on the largest channels. The annotation team built the mechanism — caching system dimensions, scaling workers dynamically — and took the measurements; the target, the sequencing and the accountability were mine.

The team wrote most of the implementation; the design, the specification and the 2020 benchmark are mine. It is still the platform's annotation engine six years later.

Earlier

A derived-metrics compiler. Customers type formulas; this lowers the parsed expression tree into a MongoDB aggregation expression, recursively inlining nested derived metrics so the emitted query touches only base metrics the database can compute. I extended the language with conditional metrics and conditional dimensions, and contributed operator introspection back to the expression library upstream. Still in production. I wrote up the dependency-resolution half of it at the time: Resolving metric dependency & expression with DAG & AST (2019).

A dataframe migration. A colleague proposed Polars and the team was already inclined towards it. What I contributed was the evidence and the plumbing. I ran the benchmark — 150× on time and about 73,000× on memory against the incumbent — and when it was challenged as too good to believe, re-ran it on a hundred times the data, where the incumbent was OOM-killed doing a group-by aggregate. Then I made it shippable: upstream had dropped support for the Ruby version the platform was pinned to, so I forked the gem, downgraded the requirement, verified the build, and sorted out the native build dependencies. The team moved the workloads across.