Introduction

dataferry is an open-source data migration engine written in Rust. You describe a migration in a few YAML files: which databases to read from and write to, and how each source record becomes rows in the target tables. dataferry plans it, copies the data in partitions that each commit on their own, and then checks that what landed is what it computed.

This release, 0.1.0-beta.1, moves data from MongoDB or PostgreSQL into PostgreSQL, as a single-process command-line tool. It is a beta: the config grammar and the exit codes are meant to stay, but expect rough edges, and read Limitations and roadmap before you point it at data you care about.

What it does

• Reshapes while it copies. One flow can copy a table 1:1, split one record into several tables, unnest an array into a child table, or merge several entities into one row with joins and nested arrays (Transforms). Values are computed with a small, pure, SQL-like expression language that is type-checked before anything runs (Expressions).
• Survives being stopped. Each partition (a slice of a source entity) is written in one transaction and committed once. A run that is killed, interrupted or loses its connection resumes from the partitions that committed and writes nothing twice (Running migrations).
• Tells you what happened. Every record a run reads ends up written, filtered by a where: you wrote, or rejected with a reason (and, if you ask for it, kept in a dead-letter table). After the run, the target is read back and compared with checksums taken at write time, constraints are checked, and your own expect: assertions are evaluated. The result is a report in Markdown and JSON (Verification).
• Fails before it starts. dataferry validate checks a config without connecting to anything, and dataferry plan discovers the sources, type-checks every expression and prints the DDL it would run, without writing a byte. Errors point at the file, line and column you wrote.

Who it is for

Engineers who need to move a database’s data, once or a few times, into a differently shaped PostgreSQL schema: a MongoDB-to-Postgres migration, a consolidation of legacy tables into a reporting schema, a re-modelling between two Postgres databases. The config is plain files you can review and commit, and validate and plan run in CI without credentials to production.

What it is not

• Not a sync or CDC tool. A run copies what the source holds when it reads it. It doesn’t follow changes afterwards, and mode: upsert never deletes target rows.
• Not a general ETL scheduler. There are no triggers, schedules or web UI. Each run is one command.
• Not distributed. One process runs every flow, with as many workers as you configure.
• Not (yet) a many-to-many connector hub. The built-in connectors are a PostgreSQL source and sink and a MongoDB source. You can add your own connector in Rust and ship it in your own binary.

A migration in 60 seconds

A config is a directory, conventionally called ferry/:

ferry/
  ferry.yaml          # version: 1 and the migration's name
  connections.yaml    # the databases, with credentials from ${ENV}
  runtime.yaml        # where run state lives, workers, error policy
  flows/
    orders.yaml       # one or more flows: from, sink, to, columns

A flow reads one source entity (here, a Mongo collection) and writes rows into one or more tables of one sink. This one, from the quickstart, turns each order document into a row in orders and one row per element of its items array in order_items:

flows:
  - name: orders
    from: shop.orders
    sink: warehouse
    mode: insert
    # ...
    to:
      orders:
        key: [id]
        columns:
          id: string(_id)
          customer_id: customer.id          # null when customer is null (…000a)
          status: lower(status)
          placed_at:
            expr: created_at                # null when missing (…0009)
            type: timestamp(UTC)
          total_cents: int(round(total * 100))   # decimal half-up: 1.005 → 101

      # Part of S2: the first end-to-end run writes both tables.
      order_items:
        parent: orders
        unnest: items                       # empty (…0002) or missing (…0003): no rows
        key: [order_id, line_no]
        columns:
          id: $key                          # UUIDv5 over (_id, items, $index)
          order_id: $parent.id              # the mapped parent row
          line_no: $index
          sku: sku
          qty: int(qty)                     # "three" (…0006) rejects the whole document
          unit_cents: int(round(price * 100))

Then:

dataferry validate      # check the config, offline
dataferry plan          # discover sources, type-check, show DDL and strategy
dataferry run           # copy, verify, write the report

run prints a run id first. If it is stopped for any reason, dataferry resume <run-id> continues from the last committed partitions. When it finishes, the report says how many records were read, written, filtered and dead-lettered, whether every table’s checksum matched, and whether your assertions held. The exit code says the same thing to a script: 0 for success, 6 when partitions failed, 7 when verification failed, and so on (CLI reference).

How this guide is organised

• Getting started: install dataferry and run the quickstart, a real Mongo-to-Postgres migration on Docker, in about five minutes.
• User guide: the concepts, then each part of a config and a run in the order you meet them.
• Reference: every command and flag, every config key, the JSON Schemas for editor completion, and the decision records that explain why things are the way they are.

A note on names: the project, the crates and the command are dataferry. The config directory (ferry/, with ferry.yaml), the local work directory (.ferry/), the tables dataferry creates in your database (_ferry_*) and the environment variables it reads (FERRY_*) keep their shorter names.

Installation

dataferry is one binary, dataferry, with the PostgreSQL and MongoDB connectors built in. There are three ways to get it.

From crates.io

You need a Rust toolchain, 1.94 or newer (install it with rustup). Then:

cargo install dataferry --version 0.1.0-beta.1 --locked

Because this release is a pre-release, cargo install dataferry without --version doesn’t pick it: Cargo only installs pre-releases you ask for by version. --locked builds with the exact dependency versions the release was tested with.

The build takes a few minutes: the expression language is built on Apache DataFusion and Arrow, which are large.

Prebuilt binaries

Each release on the GitHub Releases page has an archive (.tar.gz) for each of:

next to a .sha256 file holding its SHA-256 checksum. Download both for your platform, check the archive against the checksum, unpack it and put dataferry on your PATH. With the two files in the current directory (the names below are placeholders for the ones you downloaded):

shasum -a 256 -c dataferry-<version>-<target>.tar.gz.sha256
tar -xzf dataferry-<version>-<target>.tar.gz

shasum -c prints <archive>: OK when the archive matches. On Linux, sha256sum -c does the same. If the checksum file holds only the hash, compare it by eye with the output of shasum -a 256 <archive>. Never run a binary whose checksum doesn’t match.

On macOS, a binary downloaded with a browser is quarantined by Gatekeeper. If macOS refuses to run it, clear the flag:

xattr -d com.apple.quarantine dataferry

From the Git repository

To build the latest code on main (or any branch or tag) instead of a release:

cargo install --git https://github.com/0xvasanth/dataferry dataferry --locked

Add --tag v0.1.0-beta.1 to build exactly that release, or --branch <name> for a branch. From a clone, cargo build --release puts the binary in target/release/.

Leaving a connector out

The postgres and mongo connectors are default Cargo features of the dataferry crate. A binary that only moves Postgres to Postgres can leave the MongoDB driver out:

cargo install dataferry --version 0.1.0-beta.1 --locked --no-default-features --features postgres

postgres also brings the Postgres state store (runtime.state.backend: postgres). Without mongo, a type: mongo connection is an unknown type.

Check the installation

dataferry --version
dataferry --help

--version prints dataferry 0.1.0-beta.1. --help lists the commands, which the CLI reference describes in full.

Shell completion

The binary doesn’t generate shell completion scripts in this release.

Editor support

dataferry can print a JSON Schema for each of its config files, so editors can complete and check them as you type. See JSON Schemas.

Uninstall

cargo uninstall dataferry removes a binary Cargo installed. A prebuilt binary is just the file you put on your PATH.

dataferry doesn’t install anything else. What it creates lives where you point it: a .ferry/ work directory next to each config you run, and _ferry_* tables in the databases a run writes to or keeps its state in (Concepts).

Quickstart

This walks through a real migration on your machine: ten MongoDB order documents become rows in two PostgreSQL tables, orders and order_items. It takes about five minutes and needs:

• dataferry on your PATH (Installation),
• Docker, to run a throwaway MongoDB and PostgreSQL,
• git, to fetch the example config and data.

Every command and every output on this page is checked by the project’s tests against real containers, so what you see here is what you should get. Run ids, times and the order of a few lines in docker output will differ.

1. Get the example

The example lives in the repository: the config in examples/golden/ferry/ and the ten documents in fixtures/golden/source/orders.json.

git clone --depth 1 https://github.com/0xvasanth/dataferry
cd dataferry

2. Start the databases

A MongoDB 7 and an empty PostgreSQL 16, on their usual ports. The password is a throwaway for a container on your machine:

docker run -d --name dataferry-mongo -p 27017:27017 mongo:7
docker run -d --name dataferry-pg -p 5432:5432 -e POSTGRES_PASSWORD=quickstart-pw postgres:16

Wait until both accept connections, then load the documents into the orders collection of a shop database:

until docker exec dataferry-mongo mongosh --quiet --eval 'db.runCommand({ ping: 1 })' >/dev/null 2>&1; do sleep 1; done
until docker exec dataferry-pg pg_isready -h 127.0.0.1 -U postgres >/dev/null 2>&1; do sleep 1; done
docker exec -i dataferry-mongo mongoimport --quiet --db shop --collection orders --jsonArray < fixtures/golden/source/orders.json

If ports 5432 or 27017 are taken on your machine, publish different ones (-p 15432:5432) and use them in the URLs below.

3. Point the config at them

The config never holds credentials. Its connections.yaml names two environment variables:

# `name` and `type` are fixed by 0003 (`description` is reserved). The other
# fields are each type's own, per 0004 §2. `schema` defaults to public.
connections:
  - name: shop
    type: mongo
    url: ${FERRY_GOLDEN_MONGO_URL}
    database: shop

  - name: warehouse
    type: postgres
    url: ${FERRY_GOLDEN_PG_URL}

Set them:

export FERRY_GOLDEN_MONGO_URL=mongodb://localhost:27017
export FERRY_GOLDEN_PG_URL=postgres://postgres:quickstart-pw@localhost:5432/postgres

4. Validate, offline

validate loads every file of the config, substitutes the variables and checks each connection’s options and every flow, without connecting to anything:

dataferry validate --config examples/golden/ferry

`golden_orders` is valid: 2 connections, 1 flow

With -v, it also says how it understood each connection. Secrets never show: the Mongo URL came from the environment, so all of it is ***, and the Postgres one is shown without its password:

dataferry -v validate --config examples/golden/ferry

`golden_orders` is valid: 2 connections, 1 flow
  shop (mongo, used as source): ***, database shop
  warehouse (postgres, used as sink): postgres://postgres@localhost:5432/postgres, schema public

The flow it found, examples/golden/ferry/flows/orders.yaml, is the whole migration. Each key is explained in Configuration and Transforms; the comments say what each line does to the ten documents:

flows:
  - name: orders
    from: shop.orders
    sink: warehouse
    mode: insert

    # True when deleted_at is null *or* missing (B2), so …0001 and …0005 are
    # kept and only …0004 is filtered. A flow filter removes the record from
    # every target, so …0004's item is dropped too.
    where:
      not_deleted: deleted_at == null

    # Ten documents in partitions of four gives three partitions, which is
    # enough to exercise resume. Allowed per flow because it only sizes this
    # flow's own work (0003 §6), and left out of the config hash (§10.5).
    runtime:
      partition_rows: 4

    to:
      orders:
        key: [id]
        columns:
          id: string(_id)
          customer_id: customer.id          # null when customer is null (…000a)
          status: lower(status)
          placed_at:
            expr: created_at                # null when missing (…0009)
            type: timestamp(UTC)
          total_cents: int(round(total * 100))   # decimal half-up: 1.005 → 101

      # Part of S2: the first end-to-end run writes both tables.
      order_items:
        parent: orders
        unnest: items                       # empty (…0002) or missing (…0003): no rows
        key: [order_id, line_no]
        columns:
          id: $key                          # UUIDv5 over (_id, items, $index)
          order_id: $parent.id              # the mapped parent row
          line_no: $index
          sku: sku
          qty: int(qty)                     # "three" (…0006) rejects the whole document
          unit_cents: int(round(price * 100))

    expect:
      order_count: count(orders) == source.count()
      order_total: sum(orders.total_cents) == source.sum(int(round(total * 100)))
      item_count: count(order_items) == source.sum(coalesce(cardinality(items), 0))

5. Plan

plan connects read-only: it samples the Mongo collection to learn its fields and types, asks Postgres what it can do, type-checks every expression and prints what a run would do, including the DDL. It writes nothing:

dataferry plan --config examples/golden/ferry

plan for `golden_orders`, config sha256:c4e1f41657f0c914abccab48976f2122ca2041b6769bd83a031f758b787109e4
state store: postgres, on connection `warehouse`
on_error: dead_letter, max_errors: 10

connections
  shop       mongo (source): ***, database shop
  warehouse  postgres (sink, state store): postgres://postgres@localhost:5432/postgres, schema public

run order
  1. orders

flow `orders`
  from:        shop.orders (driver, key _id)
  sink:        warehouse, mode insert
  strategy:    TxnWithState
  partitions:  about 4 rows each (the flow's runtime:), stored as `orders/<partition>`
  where:       not_deleted: deleted_at == null
  write order: orders → order_items
  targets:
    orders (key id)
      id           utf8
      customer_id  utf8, nullable
      status       utf8, nullable
      placed_at    timestamp(UTC), nullable
      total_cents  int64, nullable
    order_items (key order_id, line_no; parent orders; unnest items)
      id          uuid
      order_id    utf8
      line_no     int64
      sku         utf8, nullable
      qty         int64, nullable
      unit_cents  int64, nullable
      foreign key (order_id) → orders (id)
  expect:
    order_count: count(orders) == source.count()
    order_total: sum(orders.total_cents) == source.sum(int(round(total * 100)))
    item_count: count(order_items) == source.sum(coalesce(cardinality(items), 0))
  ddl (not run by plan):
    CREATE SCHEMA IF NOT EXISTS "public";
    CREATE TABLE "public"."orders" (
      "id" text NOT NULL,
      "customer_id" text,
      "status" text,
      "placed_at" timestamp with time zone,
      "total_cents" bigint,
      PRIMARY KEY ("id")
    );
    CREATE TABLE "public"."order_items" (
      "id" uuid NOT NULL,
      "order_id" text NOT NULL,
      "line_no" bigint NOT NULL,
      "sku" text,
      "qty" bigint,
      "unit_cents" bigint,
      PRIMARY KEY ("order_id", "line_no")
    );
    CREATE TABLE "public"."_ferry_dlq" (
      "flow" text NOT NULL,
      "source_id" text NOT NULL,
      "run" text NOT NULL,
      "partition" text NOT NULL,
      "target" text,
      "error" text NOT NULL,
      "record" jsonb,
      PRIMARY KEY ("flow", "source_id")
    );
    ALTER TABLE "public"."order_items" ADD CONSTRAINT "order_items_order_id_fkey" FOREIGN KEY ("order_id") REFERENCES "public"."orders"("id") NOT VALID;
    ALTER TABLE "public"."order_items" VALIDATE CONSTRAINT "order_items_order_id_fkey";

A few things to notice:

• The Mongo URL shows as ***: every value that came from the environment is redacted from everything dataferry prints. The Postgres one is shown without its password.
• strategy: TxnWithState is the exactly-once strategy: each partition’s rows and the record that it is done commit in one transaction, because the run’s state store is on the same connection as the sink (runtime.yaml, Running migrations).
• The flow is split into partitions of about four documents, so ten documents make three partitions, each committed on its own.
• _ferry_dlq is the dead-letter table, created because the config says on_error: dead_letter (Errors and dead letters).

6. Run

dataferry run --config examples/golden/ferry

run ef3a454bec5a4a0ca03e74dc1ffcb6e0
  flow `orders`: TxnWithState
run ef3a454bec5a4a0ca03e74dc1ffcb6e0: complete; 3 partitions; read 10, dead-lettered 1; wrote order_items 7, orders 8
verification of `warehouse` passed; report: examples/golden/.ferry/runs/ef3a454bec5a4a0ca03e74dc1ffcb6e0/report.md

The first line is the run id. Keep it: resume and retry take it. The exit code is 0.

Ten documents were read. One (…0004) was filtered by the flow’s where: because it is soft-deleted, one (…0006) was dead-lettered because an item has qty: "three", and eight were written, with seven items between them. Then the target was read back and verified.

7. Look at the result

docker exec dataferry-pg psql -U postgres -c 'select * from orders order by id'

            id            | customer_id | status  |       placed_at        | total_cents 
--------------------------+-------------+---------+------------------------+-------------
 650000000000000000000001 | c1          | paid    | 2026-03-01 10:00:00+00 |        2550
 650000000000000000000002 | c1          | paid    | 2026-03-02 10:00:00+00 |           0
 650000000000000000000003 | c2          | pending | 2026-03-03 10:00:00+00 |        1200
 650000000000000000000005 | c2          | paid    | 2026-03-05 10:00:00+00 |         725
 650000000000000000000007 | c1          | paid    | 2026-03-07 10:00:00+00 |         101
 650000000000000000000008 | c3          | shipped | 2026-03-08 10:00:00+00 |         300
 650000000000000000000009 | c2          | paid    |                        |         400
 65000000000000000000000a |             | pending | 2026-03-10 10:00:00+00 |          50
(8 rows)

Order …0007 cost 1.005: int(round(total * 100)) rounds the decimal half away from zero, so it is 101 cents, where floating-point arithmetic would give 100. …0009 has no created_at, so placed_at is null, and …000a has customer: null, so customer_id is null.

docker exec dataferry-pg psql -U postgres -c 'select order_id, line_no, sku, qty, unit_cents from order_items order by order_id, line_no'

         order_id         | line_no |  sku  | qty | unit_cents 
--------------------------+---------+-------+-----+------------
 650000000000000000000001 |       0 | SKU-1 |   2 |       1000
 650000000000000000000001 |       1 | SKU-2 |   1 |        550
 650000000000000000000005 |       0 | SKU-3 |   1 |        725
 650000000000000000000007 |       0 | SKU-4 |   1 |        101
 650000000000000000000008 |       0 | SKU-5 |   3 |        100
 650000000000000000000009 |       0 | SKU-1 |   4 |        100
 65000000000000000000000a |       0 | SKU-6 |   1 |         50
(7 rows)

Orders …0002 and …0003 have no items (an empty and a missing items array), so they have no rows here, and the filtered order’s item is gone with it. The dead-lettered document is in _ferry_dlq, with the reason:

docker exec dataferry-pg psql -U postgres -c 'select source_id, target, error from _ferry_dlq'

          source_id           |   target    |                                     error                                      
------------------------------+-------------+--------------------------------------------------------------------------------
 ["650000000000000000000006"] | order_items | order_items.qty = `int(qty)`: int(): `"three"` is not an integer (at items[1])
(1 row)

8. Read the report

The run wrote report.md and report.json next to the config, under .ferry/runs/<run-id>/ (the path run printed):

cat examples/golden/.ferry/runs/<run-id>/report.md

# Run ef3a454bec5a4a0ca03e74dc1ffcb6e0: verification PASSED

Verification proves that what the run computed is what landed: every row the runner handed the sink for a done partition is in the target, exactly once, with the same values after normalization, and the target holds no rows the run didn't write. Counts are what the run read, recorded in the state store at read time, so changes to the source after it was read don't affect them.

It does not prove the mapping is correct: a wrong expression computes wrong values, and they land faithfully. Only `expect:` assertions check the mapping. Differences below a target column's timestamp precision, between equal numbers of different width or scale, and in JSON key order or whitespace are normalized away. The checksum covers whole tables, so it assumes a full load into empty targets: rows already in a table count as rows the run didn't write.

## Run

- Config hash: `sha256:c4e1f41657f0c914abccab48976f2122ca2041b6769bd83a031f758b787109e4`
- Engine version: 0.1.0-beta.1
- Verifier version: 0.1.0-beta.1
- Target: postgres, schema public
- Connection `shop`: mongo
- Connection `warehouse`: postgres
- Run: 2026-10-18T06:39:47.378Z to 2026-10-18T06:39:49.010Z (1.6s)
- Verification: 2026-10-18T06:39:49.010Z to 2026-10-18T06:39:49.472Z (0.5s)
- Partitions: 3 done, 0 failed, 0 pending, 0 leased

## Counts

10 driver rows read, 1 dead-lettered (from the state store).

| Flow | Read | Written | Filtered | Dead-lettered | Unaccounted |
|---|---:|---:|---:|---:|---:|
| orders | 10 | 8 | 1 | 1 | 0 |

## Tables

| Table | Committed | Written | Read back | Checksum | Result |
|---|---:|---:|---:|---|---|
| orders | 8 | 8 | 8 | matches | pass |
| order_items | 7 | 7 | 7 | matches | pass |

## Constraints

| Table | Constraint | Violations | Result |
|---|---|---:|---|
| orders | `id` not null | 0 | pass |
| orders | unique (id) | 0 | pass |
| order_items | `id` not null | 0 | pass |
| order_items | `order_id` not null | 0 | pass |
| order_items | `line_no` not null | 0 | pass |
| order_items | unique (order_id, line_no) | 0 | pass |
| order_items | (order_id) references orders (id) | 0 | pass |

## Assertions

| Flow | Name | Assertion | Values | Result |
|---|---|---|---|---|
| orders | order_count | `count(orders) == source.count()` | 8 == 8 | pass |
| orders | order_total | `sum(orders.total_cents) == source.sum(int(round(total * 100)))` | 5326 == 5326 | pass |
| orders | item_count | `count(order_items) == source.sum(coalesce(cardinality(items), 0))` | 7 == 7 | pass |

## Dead letters

1 records dead-lettered (from the state store); `_ferry_dlq` holds 1 rows for this run: pass.

| Flow | Target | Error | Records | Example `source_id` |
|---|---|---|---|---|
| orders | order_items | order_items.qty = \`int(qty)\`: int(): \`"three"\` is not an integer (at items[1]) | 1 | `["650000000000000000000006"]` |

Verification explains each section. In short: every row the run computed is in the target exactly once with the same values (Tables), every record read is accounted for (Counts), the keys and the foreign key hold (Constraints), and the three expect: assertions in the flow hold (Assertions).

9. Resume and retry

A run that stopped part-way (Ctrl-C, a crash, a lost connection) continues from the partitions that committed. A run that finished with failed partitions re-runs only those. Both take the run id. On this finished run, they find nothing left to do, write nothing twice, and verify again:

dataferry resume --config examples/golden/ferry <run-id>

resuming run ef3a454bec5a4a0ca03e74dc1ffcb6e0
  flow `orders`: TxnWithState
run ef3a454bec5a4a0ca03e74dc1ffcb6e0: complete; 3 partitions; read 10, dead-lettered 1; wrote order_items 7, orders 8
verification of `warehouse` passed; report: examples/golden/.ferry/runs/ef3a454bec5a4a0ca03e74dc1ffcb6e0/report.md

dataferry retry --config examples/golden/ferry --failed <run-id>

retrying the failed partitions of run ef3a454bec5a4a0ca03e74dc1ffcb6e0
  flow `orders`: TxnWithState
run ef3a454bec5a4a0ca03e74dc1ffcb6e0: complete; 3 partitions; read 10, dead-lettered 1; wrote order_items 7, orders 8
verification of `warehouse` passed; report: examples/golden/.ferry/runs/ef3a454bec5a4a0ca03e74dc1ffcb6e0/report.md

Running migrations shows what happens when a run really is interrupted.

10. Clean up

docker rm -f dataferry-mongo dataferry-pg

The run’s work directory, examples/golden/.ferry/, holds the reports and the write-time records verification used; delete it when you no longer need them.

Try this

• Break the mapping. In examples/golden/ferry/flows/orders.yaml, change total_cents: int(round(total * 100)) to int(round(total * 100)) * 10, start again from an empty Postgres and run. The tables still verify, because the wrong values land faithfully, but the order_total assertion fails with 53260 == 5326, the report says verification FAILED, and run exits 7.
• Interrupt it. Press Ctrl-C during a run: it exits 130 and tells you the resume command to type.
• Write your own. Configuration explains the ferry/ directory, and Transforms every shape a flow can take.

Concepts

This page names the moving parts and shows how a run uses them. Later chapters go into each one.

Connections, sources and sinks

A connection is a database dataferry talks to, declared in connections.yaml with a name, a type and that type’s own options (usually a url from an environment variable). The type is a connector: this release has postgres and mongo.

A connection is used as a source (read from), a sink (written to), or both. Each connector says which roles it can play:

dataferry validate checks that every connection a flow reads from is a source and every one it writes to is a sink.

Entities and records

An entity is what a source holds rows of: a MongoDB collection, or a PostgreSQL table in the connection’s schema. Flows name it as <connection>.<entity>, for example shop.orders. For Postgres, the entity may also carry the connection’s schema, as psql would write it: on a connection with schema: public, app.public.orders and app.orders are the same table.

Each row an entity yields is a record. dataferry learns an entity’s fields and their types by asking the source to discover it: Postgres reads its catalog, MongoDB samples documents (Connectors). Every entity needs a primary key (Mongo’s _id, a Postgres primary key): it identifies a record in reports and dead letters, and it is what partitions are cut on.

Flows, targets and rows

A flow is one unit of migration, written in a file under flows/:

flows:
  - name: orders
    from: shop.orders
    sink: warehouse
    mode: insert
    to:
      orders:
        key: [id]
        columns:
          id: string(_id)
          customer_id: customer.id          # null when customer is null (…000a)
          status: lower(status)
          placed_at:
            expr: created_at                # null when missing (…0009)
            type: timestamp(UTC)
          total_cents: int(round(total * 100))   # decimal half-up: 1.005 → 101

• from: is the driver entity. Every row a flow writes is attributed to one driver record. A flow can also join other entities to its driver (a merge).
• sink: is the one connection the flow writes to.
• to: lists the flow’s targets: the tables it writes. A record can become rows in several targets (a split), and several rows in one target (an unnest).
• columns: says how each target column is computed from the record, in the expression language.

Each driver record ends a run in exactly one of three states, and the report counts each:

• written: its rows are in the targets;
• filtered: a where: you wrote excluded it, so it has no rows anywhere;
• rejected: computing one of its rows failed (int("three"), a null in a key column, …), so none of its rows are written. Depending on on_error, the run either stops, or keeps the record in the dead-letter table _ferry_dlq and carries on (Errors).

A record is never half-written: all its rows in all targets land together, or none do.

Partitions

A flow’s driver entity is cut into partitions: slices of about partition_rows records (default 100,000), each a range of the primary key. A partition is the unit of work and of commit: one worker reads it, transforms it and writes every target row it produces in one transaction, committed once (decision 0001).

That is what makes a run safe to stop. A partition is either committed, with all its rows, or not at all; a run that stops part-way loses at most the partitions in flight, and running them again can’t duplicate rows.

Partitions are planned once, when a run starts, and stored with the run. A resumed run uses the same partitions. Entities whose key can’t be cut into ranges (a composite key, or a key type a connector can’t range over) are read as one partition.

Plans

Before anything is written, dataferry plans the config:

1. It loads and validates every file (this alone is dataferry validate).
2. It discovers every entity the flows read, and asks each sink what it can do.
3. It type-checks and compiles every expression against the discovered types.
4. It orders the flows: a flow waits for the flows in its depends_on:, and for any flow writing a table it reads.
5. For each flow, it picks an idempotency strategy (how a partition is made safe to write twice), the join strategies of a merge, and the DDL the sink would run.

dataferry plan prints all of this and stops. dataferry run plans the same way, then runs.

Runs and the state store

A run is one execution of the plan, with a run id (32 hex digits) printed first. The run, its partitions and their status are kept in the state store:

• backend: file: files in the work directory, the default;
• backend: postgres: tables in a Postgres connection you name.

The run pins two things when it starts: the config hash (a SHA-256 of the config’s meaning, not its layout) and the engine version. dataferry resume <run-id> continues a run from its stored partitions, and refuses if either changed, so a resumed run computes exactly what the interrupted one would have (Running migrations).

Leases and epochs

A worker takes a partition by leasing it from the state store. A lease lasts 30 seconds and the worker renews it every 10 while it works. Each lease has an epoch, a number that grows with every lease ever granted.

If a process dies, its leases expire, and a resumed run can lease those partitions again, at a higher epoch. Writes are fenced by epoch: a worker whose lease was superseded can’t record its partition done. Under the exactly-once strategy it can’t commit its rows either, since the epoch is checked inside the commit. Under the other strategy the epoch is checked just before the commit, and a stale write that slips past that check writes the same rows the new lease writes, which replace them by key.

Idempotency strategies

How a partition is made safe to write twice depends on where the state store is:

• TxnWithState, exactly-once: when the state store is on the flow’s sink connection, the partition’s rows and the record that it is done commit in the same transaction. Either both are there or neither is.
• DeletePartitionThenInsert: otherwise, the sink first deletes the target rows whose keys the partition is about to write, then writes them. A partition interrupted between its commit and the state store update is written again, and the second write replaces the first.

The config never names a strategy: dataferry plan shows which one each flow gets and why. Keep the state store on the sink (runtime.state.connection: <sink>) to get TxnWithState.

Verification and the report

After a run, dataferry verifies each sink: it reads the target tables back and compares them with checksums it took of every row as it wrote them, checks keys, not-nulls and foreign keys, accounts for every record read, checks the dead-letter table, and evaluates the flows’ expect: assertions. The result is a report, report.md and report.json (Verification).

Where things are kept

The work directory, .ferry/ next to the config directory by default (--work-dir or FERRY_WORK_DIR to move it), holds what a run keeps outside the databases:

Use one work directory per process: two runs at the same time must not share it.

In your databases, dataferry creates tables whose names start with _ferry_ (a target table can’t use that prefix):

How a run proceeds

 dataferry run
   │
   ├─ plan ─────────── load + validate config ─► discover sources ─► type-check
   │                   ─► order flows ─► strategies, partitions, DDL
   │
   ├─ create the run ─ pin config hash + engine version in the state store
   │                   print "run <run-id>"
   │
   ├─ for every flow, up front:
   │     prepare the sink (CREATE TABLE ..., drop deferred constraints)
   │     plan and store the flow's partitions (first attempt only)
   │
   ├─ workers, in parallel, each flow once the flows it waits for are done:
   │     lease a partition ─► read ─► transform ─► write every target
   │     ─► commit once ─► partition done
   │
   ├─ once every flow has stopped: finalize the flows that are done
   │     (restore constraints and indexes, validate foreign keys)
   │
   ├─ verify each sink ─ checksums, counts, constraints, dead letters, expect:
   │
   └─ write report.md + report.json, exit 0 / 6 / 7

The first failed partition stops the run from leasing new ones; the ones in flight finish. dataferry retry --failed <run-id> puts the failed partitions back and runs them again.

Configuration

A dataferry config is a directory of YAML files, conventionally named ferry/. This chapter explains how the directory is read, what each file holds, how environment variables and secrets work, and which settings change what a run writes. Every key is listed in the config reference.

The ferry/ directory

ferry/
  ferry.yaml          # required: version and name, optional include:
  connections.yaml    # required once there is a flow: the databases
  runtime.yaml        # optional: state store, workers, memory, error policy
  flows/              # every *.yaml / *.yml below here, at any depth
    orders.yaml
    sales/
      refunds.yml

Commands read the directory given by --config <dir> (or the FERRY_CONFIG environment variable), and ./ferry by default. The directory must contain ferry.yaml: that file is what marks it as a config. Pointing --config at a file instead is an error that suggests its directory.

How the files are loaded:

• Each file holds only its own top-level key: ferry.yaml holds version:, name: and include:; connections.yaml only connections:; runtime.yaml only runtime:; a flow file only flows:.
• .yaml and .yml are both accepted. Having connections.yaml and connections.yml side by side is an error.
• Every *.yaml and *.yml file under flows/ is a flow file, at any depth. Other files (a README.md) are ignored, and so are files and directories whose names start with . (editor swap files). A flow file must contain flows:; write flows: [] to keep a placeholder. A config with no flows at all is an error.
• Order doesn’t matter. Files are read sorted by path, which only decides which error is reported first. Moving a flow to another file changes nothing.
• One YAML document per file. A second document (---) is an error, which keeps every error’s file and line exact.
• Names of connections and flows are lowercase letters, digits and _, starting with a letter. Flow names are unique across all files, and a duplicate is reported with both places.
• Unknown keys are errors at every level, so a typo fails validate instead of being ignored.

error[F0105]: runtime.wrokers: unknown key `wrokers`
 --> ferry/runtime.yaml:7:3
  |
7 |   wrokers: 4
  |   ^^^^^^^^
  = help: did you mean `workers`?
config is invalid: 1 error

YAML details that matter

The loader reads YAML 1.2, where on, yes and no are strings, not booleans (many editors still assume YAML 1.1). The grammar never uses them as keys: joins say keys:, not on:.

Expressions and other values must be YAML strings. A bare true, null or 5 where an expression is expected is an error, so YAML’s type guessing never silently changes an expression. Quote it ("true"):

error[F0105]: flows[products].to.products.columns.in_stock: expected an expression (a string), `{expr, type, ...}` or `{nest, columns, ...}`, found boolean `true`
  --> ferry/flows/products.yaml:12:21
   |
12 |           in_stock: true
   |                     ^^^^
config is invalid: 1 error

For the same reason, quote any expression that starts with *, [, {, &, !, |, >, ', ", % or @, or that contains : (a colon and a space). status: lower(status) needs no quotes; label: "'id: ' || id" does.

ferry.yaml

# Postgres → Postgres merge: one order summary row per order, with the
# customer looked up and the order's items nested as an ordered array.
version: 1
name: pg_order_summaries

# Escape hatch for files outside the convention (0003 §8). Here, flows kept in
# a legacy directory that is not flows/.
include:
  - legacy/*.yaml

• version: 1 is the version of the config grammar, and the only one.
• name: names the migration in the state store and in reports.
• include: is an escape hatch for files outside the convention.

include:

include: is a list of paths or globs (*, **), relative to the ferry/ directory. They may point outside it (../shared/connections.yaml), so several configs can share one file of connections. A config with flows still needs its own connections.yaml: when all its connections come from an include, keep a local one holding connections: [].

• An included file holds flows:, connections: or both, never runtime: (all runtime settings live in runtime.yaml).
• Its entries merge with the conventional files’ under the same rules: a duplicate flow or connection name is an error naming both files.
• A path that doesn’t exist, or a glob that matches nothing, is an error.
• A file reached twice (by the flows/ convention and a glob, or by two globs) is loaded once.
• Only ferry.yaml may have include:, so includes can’t nest.

connections.yaml

# `name` and `type` are fixed by 0003 (`description` is reserved). The other
# fields are each type's own, per 0004 §2. `schema` defaults to public.
connections:
  - name: shop
    type: mongo
    url: ${FERRY_GOLDEN_MONGO_URL}
    database: shop

  - name: warehouse
    type: postgres
    url: ${FERRY_GOLDEN_PG_URL}

Each entry has a name and a type, and every other key belongs to the type: PostgreSQL has url, password, schema, drop_indexes and foreign_keys; MongoDB has url and database. An unknown type is an error that lists the types the binary has. name, type and description are reserved in every entry (description has no meaning yet and is rejected).

validate also checks that each connection is used in a role its connector has:

error[F0303]: flows[products].sink: flow `products` writes to connection `shop`, but the `mongo` connector in this build has no sink
 --> ferry/flows/products.yaml:4:11
  |
4 |     sink: shop
  |           ^^^^
  = help: `mongo` provides: source
config is invalid: 1 error

Flows refer to a connection in two ways: an entity as <connection>.<entity> (split at the first dot, so app.public.orders is entity public.orders on connection app, which Postgres also resolves as table orders in the connection’s schema:), and a connection on its own in sink: and runtime.state.connection.

Environment variables and secrets

A config never needs to hold a credential. ${NAME} in a string value is replaced with the environment variable NAME:

• Only in connections.yaml entries (including connections from included files) and in runtime.state.path. runtime.state’s backend: and connection: are checked as written, before substitution, so ${...} there is an error too. A ${ anywhere in a flow is an error, because the config hash is taken before substitution: an environment-driven filter could change between a run and its resume, and the resumed run would write different rows.
• NAME is letters, digits and _, not starting with a digit. ${NAME} is the only form: there is no default value syntax and no ${file:...}. To use a secret mounted as a file, export it first: export PG_URL=$(cat /run/secrets/pg_url).
• A variable that isn’t set is an error that names it, without printing any other value.
• A variable set to the empty string is a value, the empty string. There is no way to write a literal ${ in a connection value.
• Substitution works on parsed values, so a variable can’t add YAML structure, and its value is never expanded again.

A ${ in a flow, and a variable that isn’t set:

error[F0108]: flows[products].to.product_dimensions.where: `${` in a flow: environment variables are substituted only in connections and runtime.state
  --> ferry/flows/products.yaml:23:16
   |
23 |         where: dimensions.width_cm > ${MIN_WIDTH}
   |                ^^^^^^^^^^^^^^^^^^^
   = help: write the value into the flow, so it is part of the config hash (0003 §9)
config is invalid: 1 error

error[F0108]: connections[warehouse].url: environment variable not set: CATALOG_PG_URI (connections.yaml:9:10)
 --> ferry/connections.yaml:9:10
  |
9 |     url: ${CATALOG_PG_URI}
  |          ^^^^^^^^^^^^^^^^^
  = help: export it before running dataferry; values are read only from the environment (0003 §9)
config is invalid: 1 error

Every substituted value, and the value of every secret field (Postgres url and password, Mongo url), is replaced with *** in everything dataferry prints and writes: diagnostics, plan output, logs and reports. That is a backstop for driver errors that echo a connection string. A secret field that holds a literal value is accepted with a warning, since config directories get committed:

warning[W0301]: connections[warehouse].url: holds a secret as a literal value, and config directories get committed
 --> ferry/connections.yaml:9:10
  |
9 |     url: ***
  |          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  = help: write `${NAME}` and export the value in the environment instead
`catalog` is valid: 2 connections, 1 flow, 1 warning

runtime.yaml

Every key is optional; without the file, every default applies.

runtime:
  # The state store is on the flow's sink connection, so data and the
  # partition's done-record commit in one transaction (TxnWithState, 0004 §8).
  # Any other connection here would make the flow fall back to
  # DeletePartitionThenInsert.
  state:
    backend: postgres
    connection: warehouse
  # Tuning: workers and memory_budget are not in the config hash (0003 §10.5).
  workers: 2
  memory_budget: 256MiB
  # …0006 has one bad item: the whole document goes to _ferry_dlq and the
  # run carries on. One dead letter is under max_errors, so `dataferry run`
  # still exits 0.
  on_error: dead_letter
  max_errors: 10

Behavioural settings change what a run writes or how it treats errors, so they are part of the config hash: change one and resume refuses the run. Tuning settings only change how fast, or how much at a time, so they are left out of the hash: a run that ran out of memory can be resumed with a bigger memory_budget, or with more workers. Partitions are stored when a run starts, so a changed partition_rows only applies to new runs.

The state store and TxnWithState

Where the state store is decides each flow’s idempotency strategy. A flow gets exactly-once TxnWithState only when state.connection is the flow’s sink:; every other flow falls back to DeletePartitionThenInsert, and plan and run say why. The catalog example with state: {backend: file} runs as:

  flow `products`: DeletePartitionThenInsert (the state store is a file (`backend: file`), not on "warehouse")

validate checks that state.connection names a connection whose type is backend: and that this type can hold the state store (only postgres in this release).

Per-flow settings

A flow may override a setting only if its effect is limited to that flow:

    # Ten documents in partitions of four gives three partitions, which is
    # enough to exercise resume. Allowed per flow because it only sizes this
    # flow's own work (0003 §6), and left out of the config hash (§10.5).
    runtime:
      partition_rows: 4

• partition_rows is honoured per flow.
• on_error and max_errors are accepted in a flow’s runtime: only with the same value as in runtime.yaml. The runner applies them to the whole run in this release, so plan refuses a flow that sets a different value (exit 4), rather than silently ignoring it. Per-flow budgets are planned for a later release.
• state, workers, memory_budget and the two source limits are process-wide, and setting them on a flow is an error.

Ordering flows

Flows with no relation between them run at the same time, sharing the workers. A flow starts only after every flow it waits for has committed all its partitions:

• the flows it names in depends_on:;
• any flow that writes a table it reads, on the same connection. When flow b reads reporting.customers and flow a writes customers to reporting, b waits for a.

Only the reading waits. Every flow’s entities are discovered, and its partitions planned, when the run starts, so a table one flow reads from another must exist by then, and its partitions are cut on the rows it holds at that moment (the first and last ranges are open-ended, so rows added later are still read, in uneven partitions).

flows:
  - name: order_summaries
    # customer_id points at reporting.customers. FKs are validated at
    # finalize anyway (A6), so this ordering is an optimization.
    depends_on: [customers]

depends_on: names must exist, appear once, not name the flow itself and not form a cycle. validate checks all of this without connecting. If a flow fails, the flows that wait for it don’t start, and the run reports them as blocked. Within one flow, the order of its tables comes from parent: (Transforms), not from depends_on:.

You rarely need depends_on: for foreign keys: a Postgres sink drops the foreign keys of existing target tables during a load and validates them at the end (PostgreSQL), so write order is an optimization, not what keeps the data consistent.

The config hash

A run pins a hash of the config, and resume and retry refuse to continue (exit 8) when it has changed. The hash is a SHA-256 of the config’s meaning:

• It is taken after defaults are filled in, so writing a default out (mode: insert) doesn’t change it.
• It is taken before ${ENV} substitution: url: ${PG_URL} is hashed as that text, so rotating a password doesn’t block a resume, and secret values never reach the hash.
• File layout doesn’t count: moving a flow to another file, renaming files, reordering flows or connections, editing comments or changing include: leaves it unchanged, as long as the model is the same.
• What has an order keeps it: columns: (the DDL column order), merge from: entries, named where: predicates and expect:.
• Tuning settings are left out: workers, memory_budget, partition_rows, source_max_connections and source_rows_per_sec, in runtime.yaml and in flows.

plan prints the hash on its first line, and the report records it.

Editor support

dataferry schema --out schemas/ writes a JSON Schema for each kind of file, including the options of every connector the binary has, so YAML editors can complete keys and flag mistakes as you type. See JSON Schemas.

Transforms

A flow reshapes records while it copies them. There are no split: or merge: keywords: each shape is a way of writing from: and to:.

They combine: one flow can merge three entities, write a summary table and unnest a child table from the same driver record.

This page covers what every shape shares. Each shape’s page has a worked example that the project’s tests run against real databases, with the rows it writes.

Anatomy of a flow

flows:
  - name: products
    from: shop.products
    sink: warehouse
    to:
      products:
        key: [id]
        columns:
          id: _id
          name: name
          price: {expr: price, type: "decimal(10,2)", allow_lossy: true}
          price_cents: int(round(price * 100))
          tag_count: int(coalesce(cardinality(tags), 0))

A target (a key under to:) is a table name, without a schema: the sink connection decides the schema (Postgres schema:, default public). One table can be written by only one flow.

Columns

columns: is an ordered map from target column to a column spec. Its order is the column order of the table dataferry creates. A spec is one of:

• An expression, as a string: status: lower(status). Its type is inferred, and becomes the column’s type.
• A typed column, {expr, type, nullable, allow_lossy}, when the type matters. type: is a canonical type (int64, decimal(12,2), utf8, timestamp(UTC), uuid, json, …). nullable: false makes the column NOT NULL. If converting the expression’s type to type: can lose information, the plan refuses it unless the column says allow_lossy: true.
• A nest column, {nest: <many alias>, columns: ...}, built from a many join (Merge and nest).
• columns: "*", instead of a map: every field of the entity, unchanged (Copy a table).

A typed column from the catalog example: the source’s price is a decimal(38,3), and the column narrows it to two decimal places, rounding half away from zero (7.125 becomes 7.13):

          price: {expr: price, type: "decimal(10,2)", allow_lossy: true}

Scopes: what a bare name means

Every target evaluates its expressions in one scope:

Sigils reach the rest: $parent is the parent target’s row, $index the element’s position, $item the element itself and $key a generated key (Expressions). An unnested target sees the element, not the driver record: it reaches values of the record through $parent, the parent’s mapped columns, so a value parent and child share is mapped once, in one place.

Keys and modes

key: lists target columns. They become the table’s primary key, and are NOT NULL. There is no hidden or generated key: a child that needs a surrogate key maps one, for example id: $key.

mode: is set per flow:

• insert (the default): the run creates the target rows. Re-running a partition is made safe by the flow’s idempotency strategy. On PostgreSQL, a partition’s rows replace any rows already in the table with the same key (the sink deletes them first), so a run into tables that already hold those keys overwrites them rather than failing.
• upsert: rows may already exist and are updated by key (INSERT ... ON CONFLICT ... DO UPDATE). Every target must have key:. Children of a parent are replaced: for each parent row in a batch, its existing child rows are deleted and the new ones written. Upsert never deletes rows for records a filter excludes, and never propagates deletes from the source.

Keys matter for re-runs. In insert mode a target without key: is allowed, and plan warns about a child target without one, but a PostgreSQL sink can only write a keyless table under TxnWithState (state store on the sink): with any other strategy, a re-run couldn’t tell which rows to replace, and the sink refuses the flow. Give every target a key.

The replacement of child rows relies on the foreign key the planner derives from parent: (below): it exists only when the child maps each of its parent’s key columns as exactly $parent.<column>.

Parents and write order

parent: <table> makes a target the child of another target of the same flow. That one declaration drives three things:

• Write order. Parents are written before their children (ordered by parent:, ties broken by table name), whatever the order of keys in the file. plan prints it: write order: products → product_dimensions → product_tags → variants → variant_options.
• Foreign keys. If a child maps every key column of its parent from $parent.<column>, the planner adds a foreign key from those columns to the parent’s key, created and validated at the end of the load.
• Filter propagation, below.

A parent: cycle, or a parent that isn’t a target of the flow, is an error.

Filters: where:

where: is one expression, or an ordered map of named predicates that must all hold:

    where:
      not_deleted: deleted_at == null

• A record or row is kept only when every predicate is true. A predicate that is null counts as false, so discontinued != true drops rows where discontinued is null or missing; write not coalesce(discontinued, false) to keep them.
• A predicate that fails to evaluate (an error, not false) doesn’t filter: the record is rejected.
• Reports name the first predicate that failed (not_deleted), or the expression text of an unnamed where:.

There are two levels, and one rule connects them:

• A flow where: applies to the driver record (in a merge, with its one joins). A record that fails it is filtered: it has no rows in any target, and is counted once.
• A target where: applies to that target’s rows, in the target’s scope, and is evaluated on the source values, before the columns are mapped.
• Propagation: when a row is filtered out of a target, every row that reaches it through parent: (children, grandchildren) is dropped with it. Filters never travel upward or sideways: a sibling target keeps its rows, and the record isn’t rejected.

Errors

Any error computing any row of a record (a failed conversion, a null in a NOT NULL column, an array that isn’t one) rejects the whole record: none of its rows in any target are written. Other records are not affected. What happens next is the run’s on_error policy (Errors and dead letters).

Copy a table 1:1

The simplest flow copies one entity into one table, unchanged. In a large migration most tables are like this, and each takes a few lines.

columns: "*"

From the pg-merge example, which copies a Postgres customers table into a reporting database:

# Loaded through `include:` in ferry.yaml, not by the flows/ convention.
flows:
  - name: customers
    from: app.public.customers
    sink: reporting
    mode: upsert
    to:
      customers:
        key: [id]
        columns: "*"            # 1:1 copy of every declared field

columns: "*" writes every field of the entity, with its name and type, in the order the source declares them. dataferry plan shows the result:

flow `customers`
  from:        app.public.customers (driver, key id)
  sink:        reporting, mode upsert
  strategy:    TxnWithState
  partitions:  about 50000 rows each (runtime.yaml), stored as `customers/<partition>`
  write order: customers
  targets:
    customers (key id)
      id     int64
      name   utf8
      email  utf8, nullable
  ddl (not run by plan):
    CREATE SCHEMA IF NOT EXISTS "public";
    CREATE TABLE "public"."customers" (
      "id" bigint NOT NULL,
      "name" text NOT NULL,
      "email" text,
      PRIMARY KEY ("id")
    );

The rules:

• "*" is allowed only in a flow with a single from: entity, on a target with no parent: and no unnest:.
• It can’t be combined with other columns in this release: to rename, convert or add one column, list them all.
• Types map from the source’s to the sink’s through dataferry’s canonical types (PostgreSQL, MongoDB). Copying from MongoDB works the same way, with the fields schema inference found.

Renaming, converting, filtering

Listing the columns gives you control over each:

      products:
        key: [id]
        columns:
          id: _id
          name: name
          price: {expr: price, type: "decimal(10,2)", allow_lossy: true}
          price_cents: int(round(price * 100))
          tag_count: int(coalesce(cardinality(tags), 0))

Here _id becomes id, price is narrowed to two decimal places, and two columns are computed. A flow where: would drop records, and a target where: rows (Transforms).

Copying into tables that exist

dataferry creates each target table that doesn’t exist, with the columns, types and primary key the plan shows. If a table already exists, it must fit: every column the flow writes must exist with a compatible type, and any column the flow doesn’t write must be nullable or have a default. A mismatch fails before anything is written, with every problem listed (PostgreSQL).

Two things to know before loading into a table that already holds rows:

• Rows with the same key are replaced, in either mode: under insert the sink deletes the target rows whose keys a partition writes, then inserts; under upsert it updates them in place (INSERT ... ON CONFLICT DO UPDATE). Rows with other keys are left alone. A target without key: can only be written under TxnWithState.
• Verification assumes empty targets. It compares whole tables with what the run wrote, so rows the run didn’t rewrite make it fail (Verification).

For repeated loads into the same tables, use mode: upsert and expect verification to report the rows already there that the run doesn’t rewrite.

Split and unnest

A split writes one source record into several tables. An unnest writes one row per element of an array. Together they turn a nested document into a set of related tables.

The worked example

Three MongoDB product documents, each with a nested dimensions object, an array of tags (strings) and an array of variants (objects), each variant with an array of options:

[
  {
    "_id": "p-1",
    "name": "Desk lamp",
    "price": { "$numberDecimal": "24.50" },
    "tags": ["lighting", "office"],
    "dimensions": { "width_cm": 15, "height_cm": 40 },
    "variants": [
      {
        "sku": "LAMP-BLK", "color": "black", "stock": 12,
        "options": [{ "name": "bulb", "value": "LED" }, { "name": "plug", "value": "EU" }]
      },
      {
        "sku": "LAMP-WHT", "color": "white", "stock": 0, "discontinued": true,
        "options": [{ "name": "bulb", "value": "LED" }]
      }
    ]
  },
  {
    "_id": "p-2",
    "name": "Notebook",
    "price": { "$numberDecimal": "3.99" },
    "tags": [],
    "variants": [
      { "sku": "NB-A5", "color": "red", "stock": 100, "discontinued": false, "options": [] }
    ]
  },
  {
    "_id": "p-3",
    "name": "Cable",
    "price": { "$numberDecimal": "7.125" },
    "tags": ["office"],
    "dimensions": null
  }
]

The edge cases: p-2 has an empty tags array and a variant marked discontinued: false; p-3 has dimensions: null and no variants at all; the second variant of p-1 is discontinued; p-3’s price has three decimal places.

The flow, docs/src/examples/catalog/ferry/flows/products.yaml, writes five tables. Taken a target at a time:

One row per record

flows:
  - name: products
    from: shop.products
    sink: warehouse
    to:
      products:
        key: [id]
        columns:
          id: _id
          name: name
          price: {expr: price, type: "decimal(10,2)", allow_lossy: true}
          price_cents: int(round(price * 100))
          tag_count: int(coalesce(cardinality(tags), 0))

A target with no parent: and no unnest: gets one row per record. Bare names are the document’s fields.

A split: another table from the same record

      # A split: a second table from the same document, one row per product
      # that has dimensions.
      product_dimensions:
        parent: products
        key: [product_id]
        where: dimensions != null
        columns:
          product_id: $parent.id
          width_cm: dimensions.width_cm
          height_cm: dimensions.height_cm

parent: products makes this a child of products: it is written after it, it sees the parent’s mapped row as $parent ($parent.id is the products.id column, already computed), and since it maps the parent’s whole key from $parent, it gets a foreign key to products. Without an unnest:, a child has one row per parent row, and its scope is still the document (dimensions.width_cm). Its where: keeps only products that have dimensions: p-3’s null and p-2’s missing field both fail it.

Unnest an array of scalars

      # An array of strings: `$item` is the element itself.
      product_tags:
        parent: products
        unnest: tags
        key: [product_id, position]
        columns:
          product_id: $parent.id
          position: $index
          tag: $item

unnest: tags writes one row per element of tags. $index is the element’s 0-based position and $item the element itself, for arrays of scalars. An empty array, a missing field and null all give no rows, and the parent row is still written.

Unnest an array of objects

      # An array of objects: bare names are the element's fields.
      variants:
        parent: products
        unnest: variants
        key: [product_id, position]
        where:
          on_sale: not coalesce(discontinued, false)
        columns:
          product_id: $parent.id
          position: $index
          sku: sku
          color: color
          stock: stock

When the elements are objects, bare names are the element’s fields (sku is short for $item.sku). The target’s where: drops the discontinued variant. discontinued is missing in most variants, and a missing value compares as null, so coalesce() gives it a default: a plain discontinued != true would drop every variant without the field.

Unnest inside an unnest

      # Nested: one row per option of each variant written above.
      variant_options:
        parent: variants
        unnest: options
        key: [product_id, variant_position, position]
        columns:
          id: $key
          product_id: $parent.product_id
          variant_position: $parent.position
          position: $index
          name: name
          value: value

The parent is variants, so unnest: options is resolved in the scope of each variant, and $parent is the mapped variant row. $index is always the innermost position. $key is a deterministic UUID for the row: version 5, over the driver record’s key, the unnest path with every outer index, and $index. The same document always gets the same keys, however it is partitioned and however often it is re-run, so $key is safe as a primary key for rows that have no natural one.

Because the discontinued variant was filtered out of variants, its options are dropped with it: a filter propagates to children through parent:.

Assertions

    expect:
      every_product: count(products) == source.count()
      every_tag: count(product_tags) == source.sum(coalesce(cardinality(tags), 0))

source.sum(...) adds up an expression over the documents the run wrote, as it read them; the target side is queried after the run (Verification).

The result

dataferry run, then the five tables, as psql shows them:

run 8f47986ff71c4ecf8d13b0b17d4c5d39
  flow `products`: TxnWithState
run 8f47986ff71c4ecf8d13b0b17d4c5d39: complete; 1 partitions; read 3, dead-lettered 0; wrote product_dimensions 1, product_tags 3, products 3, variant_options 2, variants 2
verification of `warehouse` passed; report: ./.ferry/runs/8f47986ff71c4ecf8d13b0b17d4c5d39/report.md
select * from products order by id;
 id  |   name    | price | price_cents | tag_count 
-----+-----------+-------+-------------+-----------
 p-1 | Desk lamp | 24.50 |        2450 |         2
 p-2 | Notebook  |  3.99 |         399 |         0
 p-3 | Cable     |  7.13 |         713 |         1
(3 rows)

select * from product_dimensions order by product_id;
 product_id | width_cm | height_cm 
------------+----------+-----------
 p-1        |       15 |        40
(1 row)

select * from product_tags order by product_id, position;
 product_id | position |   tag    
------------+----------+----------
 p-1        |        0 | lighting
 p-1        |        1 | office
 p-3        |        0 | office
(3 rows)

select * from variants order by product_id, position;
 product_id | position |   sku    | color | stock 
------------+----------+----------+-------+-------
 p-1        |        0 | LAMP-BLK | black |    12
 p-2        |        0 | NB-A5    | red   |   100
(2 rows)

select * from variant_options order by product_id, variant_position, position;
                  id                  | product_id | variant_position | position | name | value 
--------------------------------------+------------+------------------+----------+------+-------
 9594e786-b1ed-5fb8-894b-86c98f35905f | p-1        |                0 |        0 | bulb | LED
 b2749e44-412c-5285-8b63-6ba77dd8d21b | p-1        |                0 |        1 | plug | EU
(2 rows)

p-3’s price of 7.125 is 7.13 after narrowing to decimal(10,2), and 713 cents. p-2 is in products with no tags and no dimensions. The discontinued LAMP-WHT and its option are nowhere, and p-3, which has no variants field, simply has no variant rows.

The rules

Split

• Each key under to: is a table. Write order follows parent:; tables with no relation are ordered by name.
• A record is written in full or not at all: if computing any row of any target fails, none of the record’s rows are written.

Unnest

• unnest: <path> is resolved in the scope of the target’s parent: the parent’s element if the parent is itself unnested, otherwise the driver record. A target with unnest: but no parent: unnests from the driver record and has no $parent.
• Elements of an empty array, a missing field or null: none, and the parent is still written. This isn’t configurable.
• A value that isn’t an array (tags: "x") is an error: the record is rejected. If the source can’t give a field one type (a MongoDB field that is an array in some documents and a string in others), it is read as json, and an unnest of it takes JSON arrays element by element and rejects anything else.
• A null element still gives a row, with a null $item.
• Paths use dots for nested fields (unnest: shipping.parcels). There is no indexing into an array (items[0]); items[*].sku in an expression is the list of every element’s sku.
• $index, $item and $key exist only in unnest targets ($item only in one that unnests a path, not a many join), and $parent only in a target with parent:.

The golden example

The quickstart’s flow is a split and an unnest in one: orders from each document and order_items from its items array, with id: $key, order_id: $parent.id and line_no: $index:

      # Part of S2: the first end-to-end run writes both tables.
      order_items:
        parent: orders
        unnest: items                       # empty (…0002) or missing (…0003): no rows
        key: [order_id, line_no]
        columns:
          id: $key                          # UUIDv5 over (_id, items, $index)
          order_id: $parent.id              # the mapped parent row
          line_no: $index
          sku: sku
          qty: int(qty)                     # "three" (…0006) rejects the whole document
          unit_cents: int(round(price * 100))

Merge and nest

A merge reads several entities and writes rows built from all of them: from: is a list whose first entry is the driver and whose other entries are joined to it. A nest turns the rows of a many join into an array column. An unnest of a many join writes them as child rows instead.

The worked example

The pg-merge example builds one order_summaries row per order from three Postgres tables:

-- Source schema and a few rows for examples/pg-merge/ferry. Load into the
-- database behind APP_DATABASE_URL.

create table customers (
    id    bigint primary key,
    name  text not null,
    email text
);

create table orders (
    id          bigint primary key,
    customer_id bigint references customers (id),
    status      text not null,
    placed_at   timestamptz not null,
    total_cents bigint not null
);

-- The primary key's leading order_id column gives the range scan that
-- co-partitioning on order_id needs (0002 step 3).
create table order_items (
    order_id   bigint not null references orders (id),
    line_no    int not null,
    sku        text not null,
    qty        int not null,
    unit_cents bigint not null,
    primary key (order_id, line_no)
);

insert into customers (id, name, email) values
    (1, 'Ana', 'ana@example.dev'),
    (2, 'Bo',  null);

insert into orders (id, customer_id, status, placed_at, total_cents) values
    (10, 1, 'paid',      '2026-03-01 10:00:00+00', 2550),
    (11, 1, 'cancelled', '2026-03-02 10:00:00+00',  900),  -- filtered
    (12, 2, 'paid',      '2026-03-03 10:00:00+00',    0),  -- no items: items = []
    (13, null, 'pending','2026-03-04 10:00:00+00',  300);  -- no customer: customer_name = null

insert into order_items (order_id, line_no, sku, qty, unit_cents) values
    (10, 1, 'SKU-2', 1, 550),
    (10, 0, 'SKU-1', 2, 1000),   -- out of insert order; nest sorts by line_no
    (11, 0, 'SKU-1', 1, 900),
    (13, 0, 'SKU-5', 0, 100),    -- qty 0: dropped from the nested array
    (13, 1, 'SKU-6', 3, 100);

The flow:

flows:
  - name: order_summaries
    # customer_id points at reporting.customers. FKs are validated at
    # finalize anyway (A6), so this ordering is an optimization.
    depends_on: [customers]

    from:
      - entity: app.public.orders           # driver: first, no join
        as: o
      - entity: app.public.customers
        as: c
        join:
          cardinality: one                  # a duplicate customer id rejects the order
          keys: {id: o.customer_id}         # c.id = o.customer_id
      - entity: app.public.order_items
        as: i
        join:
          cardinality: many
          keys: {order_id: o.id}
          # Partition order_items on order_id at the driver's boundaries
          # (0002). Fail the plan instead of falling back to staging.
          strategy: co_partition

    where:
      not_cancelled: status != 'cancelled'  # bare names are driver fields

    sink: reporting
    mode: upsert

    to:
      order_summaries:
        key: [order_id]
        columns:
          order_id: id
          customer_id: customer_id
          customer_name: c.name             # null if the customer row is missing
          status: status
          placed_at: placed_at
          total_cents: total_cents
          items:
            nest: i
            order_by: [line_no]             # (order_id, line_no) appended as tie-breaker
            where: qty > 0
            columns:
              sku: sku
              qty: qty
              unit_cents: unit_cents

    # Meant for a full load into an empty reporting database (0003 §5.9).
    expect:
      one_row_per_order: count(order_summaries) == source.count()
      totals_match: sum(order_summaries.total_cents) == source.sum(total_cents)

What dataferry run prints, and what lands in reporting (customers is the 1:1 copy from Copy a table, which this flow waits for):

run a3016d8945ee4df6949b6a8cd3533ac2
  flow `customers`: TxnWithState
  flow `order_summaries`: TxnWithState
run a3016d8945ee4df6949b6a8cd3533ac2: complete; 2 partitions; read 6, dead-lettered 0; wrote customers 2, order_summaries 3
verification of `reporting` passed; report: ./.ferry/runs/a3016d8945ee4df6949b6a8cd3533ac2/report.md
select * from customers order by id;
 id | name |      email      
----+------+-----------------
  1 | Ana  | ana@example.dev
  2 | Bo   | 
(2 rows)

select order_id, customer_name, status, total_cents, items from order_summaries order by order_id;
 order_id | customer_name | status  | total_cents |                                              items                                              
----------+---------------+---------+-------------+-------------------------------------------------------------------------------------------------
       10 | Ana           | paid    |        2550 | [{"qty": 2, "sku": "SKU-1", "unit_cents": 1000}, {"qty": 1, "sku": "SKU-2", "unit_cents": 550}]
       12 | Bo            | paid    |           0 | []
       13 |               | pending |         300 | [{"qty": 3, "sku": "SKU-6", "unit_cents": 100}]
(3 rows)

• Order 11 is cancelled, so the flow where: filters it, items and all.
• Order 12 has no items: its items is [], never null.
• Order 13 has no customer (customer_id is null), so the one join finds nothing and customer_name is null. Its item with qty: 0 is dropped from the array by the nest’s where:.
• Order 10’s items were inserted out of order, and come out sorted by line_no.

from: as a list

    from:
      - entity: app.public.orders           # driver: first, no join
        as: o
      - entity: app.public.customers
        as: c
        join:
          cardinality: one                  # a duplicate customer id rejects the order
          keys: {id: o.customer_id}         # c.id = o.customer_id
      - entity: app.public.order_items
        as: i
        join:
          cardinality: many
          keys: {order_id: o.id}
          # Partition order_items on order_id at the driver's boundaries
          # (0002). Fail the plan instead of falling back to staging.
          strategy: co_partition

• The driver is the first entry, and has no join:. Every other entry has one. Every row the flow writes is attributed to one driver record, and partitions are cut on the driver.
• as: is the alias expressions use. It defaults to the last dot-separated part of the entity name (orders for app.public.orders), and must be unique in the flow. In the flow’s columns, a bare name is a driver field: status and o.status are the same.
• join.keys: maps a column of this entry to a column of an earlier one, {<column>: <alias>.<column>}; each pair is an equality, and several pairs make a composite key. A join may reference the driver or an earlier one entry (orders → customers → regions), but not a many entry. Key types must be comparable (the same type, or a safe widening; text only with text).
• Every join is a left join from the driver: a driver record with no match is kept, with null joined fields. To drop it, filter on a joined field: where: c.id != null.
• Entries may come from different connections.

Cardinality

join.cardinality: is required, and has no default:

• one: at most one match per driver record. Its fields are available as <alias>.<field> everywhere in the driver’s scope, including the flow where:. If a second match turns up at run time, the driver record is rejected (an error row, then on_error): a one join never fans out.
• many: zero or more matches. A many alias can only be the source of a nest: column or of an unnest: target. Used anywhere else it would multiply the driver record, and the plan refuses it: `i` is a `many` join: one record has many rows of it.

Join strategies

How a join is executed is chosen per join, and plan shows it with the reason:

  joins:
    c app.public.customers (one, on c.id = o.customer_id)
      strategy: staged: read once, sorted in 256MiB and spilled to disk
      why:      auto: staged, because `o.customer_id` isn't the driver's key `o.id`
    i app.public.order_items (many, on i.order_id = o.id)
      strategy: co-partitioned on `order_id`, streamed with each driver partition
      why:      pinned by `strategy: co_partition`

• Co-partitioned (streamed): the joined entity is partitioned on its join column at exactly the driver’s partition boundaries, and each pair of partitions is read and joined together, as one unit of work. Nothing is held in memory beyond the partition (decision 0002).
• Staged: the joined entity is read once, sorted and spilled to disk (in the work directory’s stage/), and each driver partition looks its rows up there. Staged joins share a quarter of memory_budget for sorting.

join.strategy: picks one:

• auto (the default) co-partitions when all of these hold: the join has exactly one key pair; its right side is written exactly as the driver’s alias and its single-column primary key (keys: {order_id: o.id}); the entry is on the driver’s connection; and both columns are of the same kind that partitions can be cut on (integer, text or bytes). Otherwise it stages, and plan says which condition failed. If the source can’t split a joined entity on the driver’s boundaries, every co-partitioned join of that flow that auto chose is staged instead.
• co_partition insists on streaming, and fails the plan instead of falling back.
• stage always stages.

A cross-connection join can only be staged. Co-partitioned child partitions can be very uneven (one order with 100,000 items), and that is accepted in this release.

nest: columns

          items:
            nest: i
            order_by: [line_no]             # (order_id, line_no) appended as tie-breaker
            where: qty > 0
            columns:
              sku: sku
              qty: qty
              unit_cents: unit_cents

A nest column builds an array of objects from a many join:

• nest: names the many alias.
• columns: are the fields of each element, evaluated in the joined row’s scope. Each is an expression or a typed column; a nest can’t contain another nest in this release.
• where: drops joined rows from the array (a filter, not an error).
• order_by: is a list of expressions in the joined row’s scope; add desc (or DESC) to sort one descending. The joined entity’s primary key columns not already listed are always appended as the tie-breaker, so the array’s order is deterministic under any partitioning, and plan prints the full ordering (ordered by line_no, order_id). Nulls sort last ascending and first descending.
• A driver record with no matches gets [], never null.

The column’s type is list(struct(...)). PostgreSQL stores it as jsonb (keys in the order Postgres normalizes them, as the output above shows).

Unnest a many join

unnest: <many alias> writes one row per joined row: the relational version of unnesting an array. The golden-pg example reads the quickstart’s orders as two Postgres tables and writes the same orders and order_items:

flows:
  - name: orders
    # The golden mapping over rows: `order_items` joined to its order, and
    # unnested from the join where the documents unnest `items`.
    from:
      - entity: shop.orders                     # driver: first, no join
        as: o
      - entity: shop.order_items
        as: i
        join:
          cardinality: many
          keys: {order_id: o.id}
    sink: warehouse
    mode: insert

    # Only …0004 is soft-deleted; its item is dropped with it.
    where:
      not_deleted: deleted_at == null

    runtime:
      partition_rows: 4

    to:
      orders:
        key: [id]
        columns:
          id: id
          customer_id: customer_id
          status: lower(status)
          placed_at:
            expr: created_at
            type: timestamp(UTC)
          total_cents: int(round(total * 100))   # decimal half-up: 1.005 → 101

      # No `id: $key`: a key generated over a join isn't the one the golden
      # documents' `items` path gives, so the column is left out.
      order_items:
        parent: orders
        unnest: i
        key: [order_id, line_no]
        columns:
          order_id: $parent.id
          line_no: line_no
          sku: sku
          qty: int(qty)                         # "three" (…0006) rejects the whole order
          unit_cents: int(round(price * 100))

    expect:
      order_count: count(orders) == source.count()
      order_total: sum(orders.total_cents) == source.sum(int(round(total * 100)))

run 38dc95d03a27465d8cea2bf42fa890d3
  flow `orders`: TxnWithState
run 38dc95d03a27465d8cea2bf42fa890d3: complete; 3 partitions; read 10, dead-lettered 1; wrote order_items 7, orders 8
verification of `warehouse` passed; report: ./.ferry/runs/38dc95d03a27465d8cea2bf42fa890d3/report.md
select order_id, line_no, sku, qty, unit_cents from order_items order by order_id, line_no;
         order_id         | line_no |  sku  | qty | unit_cents 
--------------------------+---------+-------+-----+------------
 650000000000000000000001 |       0 | SKU-1 |   2 |       1000
 650000000000000000000001 |       1 | SKU-2 |   1 |        550
 650000000000000000000005 |       0 | SKU-3 |   1 |        725
 650000000000000000000007 |       0 | SKU-4 |   1 |        101
 650000000000000000000008 |       0 | SKU-5 |   3 |        100
 650000000000000000000009 |       0 | SKU-1 |   4 |        100
 65000000000000000000000a |       0 | SKU-6 |   1 |         50
(7 rows)

In an unnest of a many alias, bare names are the joined row’s fields, rows come in the joined entity’s primary-key order, and $index and $key are available ($item is not: there is no element value, only a row). Order …0006 has an item whose qty is the text three, so int(qty) rejects the whole order, which is dead-lettered.

Expressions

Every column, filter and assertion in a flow is an expression: one SQL-like scalar expression, evaluated once per record (or per array element, joined row or nested row, depending on the scope).

lower(status)
int(round(total * 100))
coalesce(customer.email, 'unknown')
deleted_at == null
case when qty > 10 then 'bulk' else 'single' end

The language is deliberately small. It is parsed and type-checked against the discovered source types by dataferry plan, before anything runs, and every function in it is pure, so the same record always gives the same row, however the run is partitioned and however often it is re-run.

Under the hood it is the SQL expression syntax of Apache DataFusion, restricted to what is listed on this page.

Literals

Text is in single quotes. Double quotes name a field: "Order Id", or "index" for a field called index.

In YAML, the whole expression is a string: write it plainly unless it starts with a character YAML treats specially, or contains : (Configuration).

Names and paths

• Names are case-sensitive: Status and status are different fields. Function names are not: LOWER(x) is lower(x).
• A path that names nothing is a plan error, with a suggestion: no field `stauts` in entity `shop.orders` / did you mean `status`?.
• A path can have one [*]. Indexing into an array (items[0]) isn’t supported: unnest: gives one row per element instead.
• parent, index, item and key are reserved at the start of a path, because their sigil forms exist. line_no: index is an error that suggests $index; a field with one of those names is written quoted, "index". The check runs in validate:

error[F0110]: flows[products].to.product_tags.columns.position: `index` is a reserved name
   | index
   | ^^^^^
  --> ferry/flows/products.yaml:38:21
   |
38 |           position: index
   |                     ^^^^^
   = help: write `$index` for the engine's value, or `"index"` for a field named index
config is invalid: 1 error

• Names starting with _ferry are reserved for the engine.

Sigils

Using a sigil where it isn’t valid is a plan error, for example `$index` is only available in `unnest:` targets.

Operators

An operation with a null operand gives null, except in the null tests and in coalesce, concat, concat_ws, greatest and least, which skip nulls (concat('x', null) is 'x'). and/or use three-valued logic.

Three things that differ from what you may expect, and that the planner doesn’t catch:

• Integer arithmetic isn’t checked for overflow: it wraps around, so the largest int64 plus 1 is the smallest. Integer / truncates (7 / 2 is 3); write 7.0 / 2 for a decimal result.
• ^ is bitwise XOR, not a power (7 ^ 2 is 5): use power(). & and | are bitwise too.
• cast(x as t) and x::t take SQL type names and convert as Arrow does, silently: 1.7::int is 1. Prefer the strict conversions below, which never truncate.

Not available, each with an error that says so: subqueries and exists (an expression sees one record; use a merge to read another entity), aggregates (sum, count, … belong in expect:), window functions, and *.

Null and missing

A source can say two different things about a field: it is null, or it is missing (the document has no such field). MongoDB sources tell them apart; relational sources have no missing fields. Most of the time the difference doesn’t matter, and the language treats them alike:

• x == null (or x = null, or x is null) is true when x is null or missing. x != null is the opposite.
• A missing value is null in every other expression, and lands in the target as NULL.
• missing(x) is true only when the field is absent. So x == null and not missing(x) means an explicit null. missing() takes a path into the current record (missing(customer.email)); a missing parent object doesn’t make its fields missing, only null.
• Ordering comparisons with the null literal (x > null) are an error: they are never true.

In a where:, a predicate that is null counts as false: the record or row is filtered. Use coalesce() to choose what a null means: where: not coalesce(discontinued, false).

Functions

Only these functions exist. Any other name is an error, with a suggestion when one is close (lowr → lower).

Besides the strict conversions and missing(), they behave as their DataFusion namesakes.

There are no date, time, random or UUID functions: an expression can’t read the clock or generate randomness (see Purity). For a generated key, use $key.

Numbers and decimals

Numbers with a fraction are decimals, and arithmetic on decimals stays exact. round() on a decimal rounds half away from zero:

Money should never go through float(). The golden example’s int(round(total * 100)) turns a Decimal128 total of 1.005 into 101 cents.

Strict conversions

int(), float() and string() convert a value or reject the record, never guess:

A null converts to null. A value that can’t be converted is an error row: the record is rejected, with the reason, for example accounts.seats = `int(seats)`: int(): `ten` is not an integer. That happens at run time, even for a constant: int('three') passes plan and rejects every record.

When a field holds mixed types in MongoDB (numbers in some documents, text in others), it is read as json, and int() parses the value’s JSON text. That is how the golden example rejects qty: "three": its error reads int(): `"three"` is not an integer, quotes included. A JSON string holding digits ("7") is rejected the same way.

Types and conversions

dataferry types every value with one of its canonical types, which connectors map to and from their own (PostgreSQL, MongoDB):

int8, int16, int32, int64, uint8, uint16, uint32, uint64, float32, float64, decimal(<precision>,<scale>) (precision up to 76), bool, utf8, binary, date, timestamp (wall-clock, no zone), timestamp(<zone>) (an instant, e.g. timestamp(UTC)), uuid, json, list(<type>), and the structs of nested objects. Timestamps have microsecond precision.

A typed column ({expr: ..., type: ...}) converts the expression’s type to the declared one. Conversions fall in three classes:

• Safe: every value converts exactly (integer widening, decimal(10,2) → decimal(12,4), anything → utf8). Nothing to do.
• Lossy: some values change or fail. The plan refuses the column unless it says allow_lossy: true; then values change as the table says, and values that can’t convert are error rows.
• Impossible (bool → timestamp, uuid → int64): an error even with allow_lossy.

The opt-in is per column, and part of the config hash. A conversion you write in the expression itself (int(x)) is explicit already.

Purity

An expression’s value depends only on the record. There is no clock (now()), no randomness (random(), uuid()), no aggregates and no reads of other rows. Every function is checked to be immutable, and a property test checks that random expressions give the same result when compiled again, evaluated again, evaluated one row at a time, and over rows in reverse order. That is what makes a re-run partition, a resumed run and a second run over the same data write the same rows.

Errors

At plan time, every error points at the file, line and column of the expression, with the field path and, where it can, a suggestion. plan reports every error it finds, not just the first. Two mistakes in one flow, from the guide’s tests:

# Two mistakes `validate` can't see, because they need the source's
# fields and types: `plan` reports all of them.
flows:
  - name: accounts
    from: crm.signups
    sink: warehouse
    where: emial != null
    to:
      accounts:
        key: [id]
        columns:
          id: _id
          email: lower(email)
          seats: {expr: seats, type: int32}

error[F0110]: flows[accounts].where: no field `emial` in entity `crm.signups`
   | emial != null
   | ^^^^^
 --> ferry/flows/accounts.yaml:7:12
  |
7 |     where: emial != null
  |            ^^^^^
  = help: did you mean `email`?
error[F0111]: flows[accounts].to.accounts.columns.seats: `seats` is utf8, but column `seats` is declared int32: utf8 → int32: text that isn't a valid int32 fails
  --> ferry/flows/accounts.yaml:14:25
   |
14 |           seats: {expr: seats, type: int32}
   |                         ^^^^^^
   = help: add `allow_lossy: true` to the column to accept it (values that don't convert become error rows), or convert explicitly: `int(seats)`

F0110 is an expression that doesn’t parse or names nothing in scope; F0111 a type that doesn’t fit where it is used. plan exits 4.

At run time, an expression that fails for one record (a conversion, a null in a NOT NULL column, an array that isn’t one) rejects that whole record, and the run’s on_error decides what happens next (Errors and dead letters).

Running migrations

This chapter is about plan, run, resume and retry: what they do, what they print, and how a run survives being stopped.

Before a run: validate and plan

dataferry validate reads the config and checks it without connecting to anything; it is safe to run in CI with placeholder URLs.

dataferry plan goes further, read-only: it discovers every entity the flows read, asks each sink for its capabilities, type-checks every expression and prints, per flow, the driver and its key, the strategy, the partition size, the joins and their strategies, the targets with their columns and types, the assertions, and the DDL the sink would run. It creates and writes nothing. Its output goes to stdout and its diagnostics to stderr, so dataferry plan > plan.txt captures just the plan for a review. The quickstart shows a full plan.

run plans exactly the same way first, so anything plan refuses, run refuses before writing.

run

dataferry run --config ferry

1. Plans the config.
2. Creates the run in the state store, pinning the config hash and the engine version, and prints run <run-id> (before any partition starts, so the id is on screen whatever happens next), then one line per flow with its strategy.
3. Prepares every flow, up front: its sink creates the missing tables and drops deferred constraints, and its partitions are planned and stored (on the first attempt only).
4. Runs the flows, in the order plan printed: a flow’s partitions are leased once every flow it waits for has committed all of its own. Each partition is read, transformed and written in one transaction, committed once.
5. Once every flow has stopped, finalizes the flows whose partitions are all done (constraints and indexes dropped for the load are restored, foreign keys validated). A flow that didn’t finish is finalized by a later resume or retry: until then, what its load dropped stays dropped.
6. Verifies each sink and writes the report.
7. Prints a summary and exits.

run a3016d8945ee4df6949b6a8cd3533ac2
  flow `customers`: TxnWithState
  flow `order_summaries`: TxnWithState
run a3016d8945ee4df6949b6a8cd3533ac2: complete; 2 partitions; read 6, dead-lettered 0; wrote customers 2, order_summaries 3
verification of `reporting` passed; report: ./.ferry/runs/a3016d8945ee4df6949b6a8cd3533ac2/report.md

The summary line counts partitions, driver records read and dead-lettered, and rows written per table (the dead-letter table aside). The exit code says how it went: 0 complete and verified, 6 partitions failed, 7 verification failed, 130 interrupted (all exit codes).

There is no progress display during a run in this release. -q hides the per-flow strategy lines and the plan’s warnings; the run id, the summary, the verification lines and the warnings W0601 and W0701 are always printed.

Interrupting and resuming

Press Ctrl-C during a run and it stops at once: partitions in flight are abandoned without committing, and their leases are released, so they can be taken again immediately. The run exits 130 and says how to continue. From a real run of 400,000 Mongo documents in 20 partitions, interrupted after 5 had committed (a transcript: the guide’s tests don’t re-run this one, since where an interrupt lands depends on timing):

run 2b82553a45d04191abfb245b3d6de94c
  flow `events`: TxnWithState
error[F0603]: interrupted: in-flight partitions of run `2b82553a45d04191abfb245b3d6de94c` were abandoned, and 2 leases were released
   = help: continue it with `dataferry resume 2b82553a45d04191abfb245b3d6de94c`

resume continues from the committed partitions:

dataferry resume --config ferry 2b82553a45d04191abfb245b3d6de94c

resuming run 2b82553a45d04191abfb245b3d6de94c
  flow `events`: TxnWithState
run 2b82553a45d04191abfb245b3d6de94c: complete; 20 partitions; read 400000, dead-lettered 0; wrote events 400000
verification of `wh` passed; report: ./.ferry/runs/2b82553a45d04191abfb245b3d6de94c/report.md

The target then held 400,000 rows with 400,000 distinct keys: the 100,000 rows of the first five partitions were not written again, and nothing of the abandoned partitions had landed.

What resume does:

• It plans again, so the sources are discovered again and the config is checked again, then compares the config hash and the engine version with the ones the run pinned. If either changed, it refuses, exit 8 (from the same transcript):

  error[F0801]: run `2b82553a45d04191abfb245b3d6de94c` was started with config hash sha256:71a5e90f626f0d56a4747328e38776672550b31f7e8d9aab3dacc5bbfce0e733, but the config now hashes to sha256:ec9f013ec3da5b9a274d39663467bb3f430af917de7435ff6aa8eba4fee8cf16: the config changed since the run started. Start a new run, or restore the config the run was started with
  

  Undo the change, or start a new run. Resume a run with the same dataferry version that started it.

• It reuses the stored partitions: partitions are planned once per run, so a resumed run covers exactly the key ranges the first attempt did, whatever partition_rows says now.

• It runs every partition that isn’t done. Failed partitions stay failed: use retry for those.

• A run id the state store doesn’t have is a usage error, exit 2.

• Resuming a run that is already complete leases nothing, but still prepares every flow again (on tables that exist, deferred foreign keys and indexes are dropped and then restored by the finalize that follows), verifies again, and rewrites the report.

Interrupted while verifying (after the data is written), the run exits 130 too, and resume verifies it again once there is nothing left to run. Interrupted while still planning, nothing has been created yet: run again.

After a crash or kill -9

A process that dies without Ctrl-C can’t release its leases. They expire 30 seconds after their last renewal, and a resume started before that waits for them, checking every second, then leases those partitions at a new epoch. Whatever the dead process had written for them but not committed was rolled back by the database. If, by any chance, the dead worker were still alive somewhere (a paused VM), epoch fencing stops it from recording its partition done, and under TxnWithState from committing its rows at all.

retry

A run that finished with failed partitions exits 6 and says so:

run ba4e513cd87644dd8b44ad24a254e0e8: not complete (0 done, 1 failed, 0 pending, 0 leased); 1 partitions; read 0, dead-lettered 0; wrote nothing
error[F0602]: flow `accounts`: partition `accounts/signups-0` failed: transform output broke its contract: record [Int(3)] was rejected for `accounts`: accounts.email can't be null, but `lower(email)` is null
verification of `warehouse` incomplete: the run isn't finished; report: ./.ferry/runs/ba4e513cd87644dd8b44ad24a254e0e8/report.md
run `ba4e513cd87644dd8b44ad24a254e0e8` has failed partitions; `dataferry retry --failed ba4e513cd87644dd8b44ad24a254e0e8` re-runs them

Fix the cause (the data, the config’s mapping if the hash allows it, a permission), then:

dataferry retry --config ferry --failed <run-id>

retry checks the pins like resume, puts the failed partitions back to pending, and resumes the run. Flows that didn’t start because a flow they wait for failed start now. Partitions that were done are not touched, including the records they dead-lettered.

A failure in the mapping (an expression that rejects records under on_error: fail) can’t be fixed by a retry alone: changing the flow changes the hash. Start a new run with the fixed config instead; with the same keys, it overwrites the rows the first one wrote.

Idempotency and exactly-once

Every way a run can stop leaves the target correct, because of how each partition is written:

• One transaction per partition. A partition’s rows in all its target tables (and its dead letters) are written in one sink transaction, committed once. A stopped run loses at most the partitions in flight, and those leave nothing behind.
• TxnWithState (when the state store is on the flow’s sink): the record that a partition is done is written in that same transaction, after the epoch check. Rows and done-record commit together or not at all, so a partition is written exactly once.
• DeletePartitionThenInsert (any other placement of the state store): the done-record is written after the commit. A crash between the two re-runs the partition, and the sink first deletes the rows with the keys it is about to write, so the second write replaces the first. The result is the same; the work is done twice.

plan and run print each flow’s strategy, and why when it isn’t TxnWithState. Keep runtime.state.connection on the sink you write to for exactly-once. A run with flows into two sinks can have exactly-once for the flows into the state store’s connection only; the others fall back.

Concurrency, memory and pacing

• workers (default: one per CPU core) is how many partitions are worked on at once, across all flows. Flows that don’t wait for each other run at the same time and share the workers.
• partition_rows sizes partitions, and therefore transactions. Larger partitions mean fewer, longer transactions; a crash loses at most one partition per worker. A merge’s co-partitioned child is cut at the driver’s boundaries, whatever its own size.
• memory_budget (default 1 GiB) is what the runner sizes its writes and read-ahead from: each write batch is about an eighth of the budget divided by the workers (at least 64 KiB), and a quarter is shared out as read-ahead. Staged merge joins get a quarter of the budget for sorting, taken off the rest. A batch the source returns isn’t split, so a run can exceed the budget by about one source batch.
• source_max_connections caps the read streams open at once across all sources; source_rows_per_sec caps the rows read per second across all sources. Use them to protect a production source.

All of these are tuning settings, left out of the config hash: change them between attempts and resume still accepts the run (Configuration).

Running it again

dataferry run always starts a new run, with a new id, new partitions and a new report. On tables the previous run filled, insert replaces the rows with the same keys, and verification fails if the tables then hold rows this run didn’t write (rows deleted from the source since, for example). For a clean re-run, empty or drop the target tables first.

run, resume and retry are meant for one process per run and per work directory. Don’t run two processes on the same run, or two runs writing the same tables, at once.

Leases, for tests only

Leases last 30 seconds, renewed every 10. The environment variable FERRY_LEASE_TTL_MS sets their length in milliseconds, from 1000 to 600000, with renewal at a third of it; any other value is a usage error (exit 2, F0601). It exists so tests can resume right after a kill, and run, resume and retry warn (W0601) whenever it is set. Don’t use it in production: a lease shorter than a slow commit or a network stall expires under a live worker, which then loses its partition (fenced, so nothing is written twice) and the run redoes or fails work it had done.

Errors and dead letters

Real data has records that don’t fit the mapping: a quantity of "three", a missing email in a column that can’t be null. dataferry never drops such a record silently. It rejects it, with a reason, and the run’s on_error policy decides what happens next: stop, or keep the record in a dead-letter table and carry on.

What rejects a record

A record is rejected, as a whole (none of its rows in any target are written), when:

• an expression fails for it: a strict conversion (int('ten')), a value that doesn’t fit a lossy typed column, an error in a where: predicate, a one join that finds two matches, or an unnest: of a value that isn’t an array;
• a value is null in a column that can’t be null (a key column, or one declared nullable: false);
• the source can’t read it as its entity: for MongoDB, a document whose field has a type the sampled schema doesn’t allow, or a field the sample didn’t see (MongoDB).

A record that a where: excludes is filtered, not rejected: that is the mapping working as written, and it is only counted.

on_error: fail

The default. The first rejected record fails its partition: nothing of that partition is committed, the run leases no new partitions, the ones in flight finish, and the run exits 6.

  # The default, written out: the first rejected record fails its partition
  # and the run.
  on_error: fail

With the guide’s signups example (five documents, two of which can’t be mapped):

run ba4e513cd87644dd8b44ad24a254e0e8
  flow `accounts`: TxnWithState
run ba4e513cd87644dd8b44ad24a254e0e8: not complete (0 done, 1 failed, 0 pending, 0 leased); 1 partitions; read 0, dead-lettered 0; wrote nothing
error[F0602]: flow `accounts`: partition `accounts/signups-0` failed: transform output broke its contract: record [Int(3)] was rejected for `accounts`: accounts.email can't be null, but `lower(email)` is null
verification of `warehouse` incomplete: the run isn't finished; report: ./.ferry/runs/ba4e513cd87644dd8b44ad24a254e0e8/report.md
run `ba4e513cd87644dd8b44ad24a254e0e8` has failed partitions; `dataferry retry --failed ba4e513cd87644dd8b44ad24a254e0e8` re-runs them
select count(*) from accounts;
 count 
-------
     0
(1 row)

Fix the data or the mapping, then start again. A changed flow is a new config hash, so it needs a new run; with the data fixed and the config unchanged, dataferry retry --failed <run-id> re-runs only the failed partitions.

on_error: dead_letter

  # Keep going: a rejected record goes to _ferry_dlq on the sink, until more
  # than 10 have been rejected in the run.
  on_error: dead_letter
  max_errors: 10

A rejected record is written to the table _ferry_dlq on the flow’s sink, in the partition’s own transaction, and the run carries on. The same five signups:

run 68abfa6b37944a45a814e4ce6adee339
  flow `accounts`: TxnWithState
run 68abfa6b37944a45a814e4ce6adee339: complete; 1 partitions; read 5, dead-lettered 2; wrote accounts 3
verification of `warehouse` passed; report: ./.ferry/runs/68abfa6b37944a45a814e4ce6adee339/report.md
select * from accounts order by id;
 id |      email      | plan | seats 
----+-----------------+------+-------
  1 | ana@example.dev | pro  |     5
  2 | bo@example.dev  | free |     1
  5 | di@example.dev  | free |     2
(3 rows)

select source_id, target, error from _ferry_dlq order by source_id;
 source_id |  target  |                             error                             
-----------+----------+---------------------------------------------------------------
 [3]       | accounts | accounts.email can't be null, but `lower(email)` is null
 [4]       | accounts | accounts.seats = `int(seats)`: int(): `ten` is not an integer
(2 rows)

select record from _ferry_dlq where source_id = '[4]';
                                           record                                            
---------------------------------------------------------------------------------------------
 {"_id": 4, "plan": "team", "email": "cy@example.dev", "seats": "ten", "_ferry_missing": []}
(1 row)

The run is complete and exits 0: three accounts are written, and two records are dead-lettered with the reason. The summary line and the report count them.

Only driver records can be dead-lettered. If a record of a joined entity can’t be read, its partition fails, whatever on_error says.

max_errors

max_errors is how many records the whole run may dead-letter. Once committing a partition would bring the run’s total above it, that partition is rolled back (its rows and its dead letters), marked failed, and the run stops with exit 6:

  # Two records are rejected, one more than this run accepts.
  on_error: dead_letter
  max_errors: 1

run cbfe8b46a34b4250ab799b56f725b2a3
  flow `accounts`: TxnWithState
run cbfe8b46a34b4250ab799b56f725b2a3: not complete (0 done, 1 failed, 0 pending, 0 leased); 1 partitions; read 0, dead-lettered 0; wrote nothing
error[F0602]: flow `accounts`: partition `accounts/signups-0` failed: runner: max_errors exceeded: this partition would bring the run to 2 dead-lettered records, more than max_errors (1); it was rolled back and the run stops
verification of `warehouse` incomplete: the run isn't finished; report: ./.ferry/runs/cbfe8b46a34b4250ab799b56f725b2a3/report.md
run `cbfe8b46a34b4250ab799b56f725b2a3` has failed partitions; `dataferry retry --failed cbfe8b46a34b4250ab799b56f725b2a3` re-runs them
select count(*) from accounts;
 count 
-------
     0
(1 row)

• The count is run-wide and includes the partitions committed by earlier attempts of the same run: a resumed run doesn’t get a fresh budget.
• max_errors is in the config hash, so you can’t raise it to resume a run: that is a new run.
• Without max_errors, nothing stops a dead-lettering run, and plan warns about it (W0401).
• In this release, on_error and max_errors are run-wide. A flow may repeat them in its runtime: only with the run’s values; a different value is refused by plan, so it is never silently ignored. Per-flow policies and budgets are planned for a later release.

A run that dead-letters records but stays within max_errors still exits 0. To fail a CI job when any record didn’t land, assert the count you expect: count(accounts) == 5 fails, with exit 7, as soon as one of the five signups is missing (Verification). count(accounts) == source.count() can’t catch it, because source. aggregates cover only the records the run wrote.

The _ferry_dlq table

Created in each sink’s schema for flows with on_error: dead_letter, with the DDL plan shows:

The primary key is (flow, source_id): one row per record and flow. A record dead-lettered again (by a re-run partition, or a later run) replaces its earlier row, so the table never fills up with duplicates.

Inspecting dead letters

It is an ordinary table:

-- what went wrong in a run, grouped
select flow, target, error, count(*)
from _ferry_dlq
where run = '<run-id>'
group by flow, target, error
order by count(*) desc;

-- the records themselves
select source_id, record
from _ferry_dlq
where flow = 'accounts' and error like '%seats%';

The report’s Dead letters section shows the same grouping, with an example source_id per reason, and checks that _ferry_dlq holds exactly the number of records the state store counted for the run.

Three things to know about record:

• For a record the mapping rejected, it includes the column _ferry_missing: the list of fields that were absent from a MongoDB document (as opposed to null). It is how the engine tracks missing values.
• For a MongoDB document the source itself rejected (it doesn’t fit the inferred schema), it is the document as relaxed Extended JSON, without _ferry_missing, and target is null.
• A MongoDB field read as json (because its type varies between documents) holds its value’s JSON text, as a string. In the quickstart’s dead letter, qty: "three" shows up as "qty": "\"three\"".

Re-running dead-lettered records

There is no command to replay dead letters in this release. Fix the records in the source (or the mapping), then run again: a new run reads every record again, writes those that now map, and dead-letters the rest again, replacing their rows in _ferry_dlq. Rows of records that now map stay in _ferry_dlq from the earlier run until you delete them:

delete from _ferry_dlq where run = '<old-run-id>';

Other failures

Not every failure is a record’s fault:

Troubleshooting goes through the common ones.

Verification

After the flows of a run have committed, run (and resume, and retry) verifies every sink it wrote to, and writes a report. There is no separate verify command in this release: to verify a finished run again, resume it, which finds nothing left to run and verifies.

What it proves, and what it doesn’t

The report says it in its first paragraphs, and it is worth reading once:

Verification proves that what the run computed is what landed: every row the runner handed the sink for a done partition is in the target, exactly once, with the same values after normalization, and the target holds no rows the run didn’t write. Counts are what the run read, recorded in the state store at read time, so changes to the source after it was read don’t affect them.

It does not prove the mapping is correct: a wrong expression computes wrong values, and they land faithfully. Only expect: assertions check the mapping.

Verification never queries the sources again. It compares the targets with what the run recorded while it wrote: per partition, a checksum and a count of the rows of each table, kept in the work directory (writes/) and in the state store.

The checks

The checksum is a 128-bit hash per row, over the mapped columns, added up per table, so it doesn’t depend on row order or partitioning. Values are normalized before hashing, so representation differences don’t count: integers and decimals of equal value hash alike, timestamps are compared as instants at the target column’s precision, -0.0 equals 0.0, and JSON is compared with its keys sorted, whitespace aside.

The result is passed when there are no problems, failed when there are any, and incomplete when some partition isn’t done (the run failed or was interrupted): the report is still written, with the counts so far.

run exits 7 when verification failed: the data is written, and the report says what doesn’t match. A run with failed partitions exits 6 instead, which takes precedence.

expect: assertions

Verification shows that the data landed as computed. Assertions check that what was computed is right. They are named, and compare two aggregates:

    expect:
      order_count: count(orders) == source.count()
      order_total: sum(orders.total_cents) == source.sum(int(round(total * 100)))
      item_count: count(order_items) == source.sum(coalesce(cardinality(items), 0))

An assertion is <side> <op> <side>, with ==, !=, <, <=, > or >=. Each side is a number, a text in single quotes, an aggregate, or arithmetic on those (+ - * /, abs(...), round(..., digits), parentheses): abs(sum(orders.total_cents) - source.sum(int(round(total * 100)))) <= 1.

Target side, computed by querying the target after the run:

When a flow has several targets, columns must be qualified with their table (orders.total_cents); with one target the table may be left out.

Source side, accumulated while the run reads:

Source aggregates cover the driver records the run wrote: not those it filtered or rejected. That makes an assertion a check of the mapping (did the cents add up for the orders we wrote?), while the Counts section accounts for filtered and dead-lettered records separately. A partition that is retried counts once.

Aggregates skip nulls: a sum of no values is null, a count is 0. A comparison with a null side holds only for !=. Integers and decimals are compared exactly; / gives a float.

plan checks every assertion against the targets and the source’s fields, so a misspelled column or table fails before the run, with a suggestion. Not allowed, each with its own error: and/or between comparisons, avg (divide a sum by a count), comparing with null, source.count_distinct, and a bare column where an aggregate belongs.

A failing assertion

The dead letters example, with two assertions added:

    expect:
      # Dead-lettered records are not written, so this holds only if none were.
      every_signup_landed: count(accounts) == 5
      seats_add_up: sum(accounts.seats) == source.sum(int(seats))

Two of the five signups are dead-lettered, so the first fails, and the run exits 7 though everything that was written verified:

run 0ab83143cdbb48ebacfb9d898e714368
  flow `accounts`: TxnWithState
run 0ab83143cdbb48ebacfb9d898e714368: complete; 1 partitions; read 5, dead-lettered 2; wrote accounts 3
verification of `warehouse` failed (1 problems); report: ./.ferry/runs/0ab83143cdbb48ebacfb9d898e714368/report.md
error[F0701]: flow `accounts`: assertion `every_signup_landed` failed: 3 == 5 is false

The second holds: 8 seats were written, and the source side adds up the seats of the three records written.

What it assumes

Verification and assertions look at whole tables: the checksum covers every row of a target, and count(orders) counts every row. That is right for the job dataferry is built for, a full load into empty targets, and it is why the report says the target holds no rows the run didn’t write.

Loading into tables that already hold rows the run doesn’t rewrite (an upsert into a live table, a second run after rows were deleted from the source) makes the table checks fail, by design: those rows are rows this run didn’t write. A second run over an unchanged source verifies, since it rewrites every key the first one wrote. Assertions count whole tables too. Scoping verification to the keys a run wrote is planned, not built.

Several sinks

Each sink is verified on its own, against its flows’ share of the run. A run with one sink writes report.md and report.json; a run writing to several writes report-<sink>.md and report-<sink>.json for each.

A sink whose type this binary can’t read back (a third-party connector without a verifier) is reported as not verified (W0701), which doesn’t change the exit code. What finalize couldn’t restore on it still fails the run.

The report files

In the work directory, under runs/<run-id>/:

• report.md, for people: the verdict (# Run <run-id>: verification PASSED, FAILED or INCOMPLETE), what it proves, then the sections Run (config hash, versions, connections by name and type, never their URLs, timings, partition counts), Problems (when there are any), Counts, Tables, Constraints, Assertions, Finalize (when needed) and Dead letters. Values are escaped and cut to 200 characters. The quickstart shows a complete one.
• report.json, for tools, format ferry-report/1: status (passed, failed, incomplete), run (id, config_hash, engine_version, connections), versions, target, timings, partitions (pending, leased, done, failed), counts (read, dead_lettered, written per table, and per flow read, records written/filtered/rejected and unaccounted), tables, constraints, assertions, finalize, dead_letter and problems. Values are kept whole.

Everything in both files goes through the same redaction as the console: no secret appears in a report.

Connectors

A connector is the code that talks to one kind of database. Each connection in connections.yaml names one by its type:. The dataferry binary has two:

Both verify TLS certificates whenever they use TLS, and use it by default for anything that isn’t on the local machine (TLS).

dataferry schema prints each type’s options as JSON Schema, and dataferry -v validate shows how each connection was understood, without secrets (quickstart).

Canonical types

Connectors don’t convert to each other’s types directly. A source maps each of its types to one of dataferry’s canonical types (Expressions); expressions compute on those; a sink maps them to its own. Each connector page has its mapping table. A source type with no mapping fails discovery with a message naming the table and column, before anything is written.

What every source does

• Discovers its entities: names, fields, types, nullability and primary key. plan and run do this on every invocation.
• Partitions an entity by ranges of its primary key, about partition_rows records each, and reads one partition at a time, in key order.
• Co-partitions a joined entity at a driver’s boundaries, when a merge asks for it (Merge and nest).
• Reads with no consistent snapshot across partitions: rows changed while a run is going may or may not be seen. Migrate from a quiet source, or a snapshot of it.

Your own connector

A connector is a Rust crate implementing a small trait, registered in your own build of the command line. See Writing a connector.

PostgreSQL

type: postgres is a source and a sink, and can hold the state store.

Connection options

connections:
  - name: legacy
    type: postgres
    url: ${LEGACY_URL}

  - name: warehouse
    type: postgres
    url: ${WAREHOUSE_URL}       # required
    password: ${WAREHOUSE_PW}   # optional: overrides the password in url
    schema: reporting           # default: public
    drop_indexes: true          # default: false
    foreign_keys: defer         # default: defer

url and password are secret: their values, and a password embedded in the URL, are replaced with *** in everything dataferry prints. plan shows the connection as postgres://user@host:port/db, schema public, without the password.

As a source

What it reads. The tables of the connection’s schema:: ordinary tables and partitioned tables (as a whole), not views, materialized views, foreign tables or the partitions of a partitioned table. A flow names a table as <connection>.<table> or <connection>.<schema>.<table> with the connection’s own schema; app.public.orders and app.orders are the same table on a connection with schema: public.

Every table needs a primary key, and every column a supported type. Discovery reads the whole schema, so one table that breaks either rule fails discovery, even one no flow reads: plan and run stop (exit 5) with a message listing every such table and why (has no primary key, which is needed to partition and re-run it safely). Fix the table, or point schema: at a schema holding only what you migrate.

Partitions. A table with a single-column primary key of type smallint, integer, bigint, text-like or uuid is split into ranges of that key, about partition_rows rows each. Text keys are ordered bytewise (COLLATE "C"). Tables with a composite key or another key type are read as one partition. Planning the ranges reads every key of the table once.

Reading. Each partition is one SELECT ... WHERE key >= $1 AND key < $2 ORDER BY key on its own connection, with binary results. Filters are not pushed down to the database: where: is evaluated by dataferry. There is no snapshot across partitions.

Co-partitioning (for a merge’s many join) works on any column of integer, text or UUID type, not only the key. Rows whose co-partition column is null match no range.

Types

Not supported, and failing discovery: any other type (point, interval, money, enums, every domain, even one over a supported type, …) and numeric without a precision (“declare the column as numeric(p, s) or cast it in a query”), or with a precision above 76. Values that can’t be represented fail the read with the table and column: a numeric NaN or infinity, an infinite date or timestamp, a multi-dimensional array.

As a sink

Tables it creates

Before a flow’s first partition, the sink prepares its targets, in one transaction: CREATE SCHEMA IF NOT EXISTS the connection’s schema, then CREATE TABLE for each target that doesn’t exist, with the columns in columns: order, NOT NULL where the column isn’t nullable (key columns always), and the PRIMARY KEY from key:. Foreign keys are added at the end of the load, NOT VALID and then validated. plan prints this DDL.

Nested values in jsonb keep decimals exact, write dates as YYYY-MM-DD, timestamps as ISO 8601 with microseconds and Z, UUIDs in their usual form and bytes as "\\x..." hex.

Tables that already exist

A target table that exists is used as it is, if it fits. Prepare checks, for every target, and reports every problem before writing anything:

• each column the flow writes exists, with the type the table above gives, or an accepted equivalent (varchar(n) for text, json for jsonb). A different type is refused, for example an integer column for an int64 value (“would truncate”);
• each column the flow doesn’t write is nullable or has a default;
• for mode: upsert, a primary key or unique index covers exactly the key: columns.

Extra nullable columns are fine. A char(n) column can’t be written (the sink writes text), even though the source reads char(n) as utf8.

How a partition is written

All of a partition’s rows, in every target table of the flow, are written in one transaction, with binary COPY through a temporary staging table, then merged into the target:

• mode: insert: rows of the target with the same keys as the partition’s are deleted, then the partition’s rows inserted. A re-run partition therefore replaces its own rows, and a run into a table that already holds those keys overwrites them.
• mode: upsert: INSERT ... ON CONFLICT (<key>) DO UPDATE (or DO NOTHING when every column is in the key).
• In both modes, before a parent’s rows land, the child rows that reference them through a foreign key the plan declared (and their own children) are deleted, so children are replaced with their parent.
• A table without key: is written with plain COPY, which only TxnWithState makes safe to re-run; under any other strategy the sink refuses a keyless target.

The strategies the sink offers are TxnWithState, when the run’s state store is on this connection, then DeletePartitionThenInsert and UpsertByKey. The planner prefers them in that order, so UpsertByKey is never chosen: an upsert flow without the state store gets DeletePartitionThenInsert, and is still written with INSERT ... ON CONFLICT, as above. plan shows which strategy each flow gets.

Existing tables: foreign keys and indexes

Loading into tables that exist could trip over their constraints, or be slowed by their indexes. When it prepares a flow whose targets already exist, the sink:

• with foreign_keys: defer (the default), drops the foreign keys defined on those tables (not foreign keys of other tables pointing at them);
• with drop_indexes: true, drops their secondary indexes (not unique, not primary, not backing a constraint);

and records each definition in _ferry_dropped, in the sink’s schema. When the flow’s partitions are all done, it finalizes: indexes are recreated, foreign keys added back NOT VALID and then validated. A foreign key whose validation fails stays NOT VALID, keeps its record, and the report lists it with the number of orphan rows and some of their keys. Run ALTER TABLE ... VALIDATE CONSTRAINT once the data is fixed. Finalize runs again on every resume or retry, so an interrupted finalize completes later.

With foreign_keys: enforce, foreign keys that already exist on target tables stay in place, every write is checked, and parents must be written before their children: use depends_on: between flows. The setting doesn’t change the foreign keys the plan declares between a flow’s own targets: those are always added at finalize, after the load.

As the state store

runtime:
  # The state store is on the flow's sink connection, so data and the
  # partition's done-record commit in one transaction (TxnWithState, 0004 §8).
  # Any other connection here would make the flow fall back to
  # DeletePartitionThenInsert.
  state:
    backend: postgres
    connection: warehouse

The state store’s tables (_ferry_runs, _ferry_partitions, _ferry_leases and the sequence _ferry_epochs) are created on first use. They are created without a schema name, so they land in the first schema of the connection’s search_path (usually public), whatever the connection’s schema: says. When the state connection is also a flow’s sink, that flow gets TxnWithState.

Connections and load

The sink (with the state store, when they share the connection) and the verifier each keep a pool of connections, of the pool library’s default size: twice the number of CPU cores. The source doesn’t pool: it opens a new connection for each discovery, partition plan and partition read. Leave the server room for both during a run.

MongoDB

type: mongo is a source. It can’t be a sink, or hold the state store, in this release.

Connection options

connections:
  - name: shop
    type: mongo
    url: ${FERRY_GOLDEN_MONGO_URL}
    database: shop

The whole url is secret: plan shows only the hosts and the database. Every value substituted from the environment is also replaced with *** wherever it appears in output, so with url: ${NAME} the description is just ***, database shop, as in the quickstart.

Collections and entities

Every collection of the database is an entity, named as the collection is: shop.orders is the collection orders on connection shop. system.* collections and views are skipped. A time-series collection fails discovery for the whole database (“time-series collections are not supported yet”).

The primary key is always _id.

Schema inference

A collection has no declared schema, so the connector samples it to infer one, every time a run or plan starts:

• A collection of up to 1,000 documents (by MongoDB’s estimated count, which can be stale) is read whole, in _id order. A larger one is sampled: 1,000 random documents ($sample).
• Each field’s type is the union of what the sample holds, by these rules:
  • int32 and int64 together are int64; int32 and double are float64;
  • Decimal128 values become one decimal(38, s), with the largest scale seen; together with integers, they stay decimal;
  • objects merge field by field, arrays element by element;
  • null adds nothing: a field that is always null, or always {}, is json, and one that is always [] is list(json);
  • anything else that disagrees (int64 and double, a number and a string, an ObjectId and a string, an object and a scalar) makes the field json: its values are read as relaxed Extended JSON.
• _id is never nullable; every other top-level field is nullable.
• A field name starting with _ferry_ fails discovery: the prefix is the engine’s.

Then, while reading, every document is checked against the inferred schema. A document that doesn’t fit (a field the sample never saw, at any depth, or a value whose type the field can’t hold) is rejected, with its _id and the path, and on_error decides: fail the run, or dead-letter the document (Errors). Nothing is cast or dropped silently. With a large, irregular collection, a rare field the sample missed shows up this way: run plan, look at the inferred types, and expect such documents in the dead-letter table.

A document whose _id itself doesn’t fit fails the whole read, not just that document.

Types

ObjectId. An ObjectId reads as its hex string, so string(_id) and _id give the same text (650000000000000000000001), it sorts like the ObjectId, and it can be compared with text. The connector still tells an ObjectId field apart from a text field: a document holding a string where the sample saw ObjectIds (or the reverse) doesn’t fit.

json fields. A field that couldn’t get one type holds JSON text. int(x) of such a field parses that text, so the number 5 converts and the string "5" (with its quotes) doesn’t. An unnest: of a json field unnests JSON arrays. Int64 values lose their width in JSON, and decimals appear as {"$numberDecimal": "1.5"}.

Missing and null. The connector records, for each document, which fields were absent, so missing(x) can tell a missing field from an explicit null (Expressions).

Partitions

A collection whose _id values are all integers, all strings, or all ObjectIds is split into ranges of _id, about partition_rows documents each, compared bytewise (collation: simple) whatever the collection’s default collation. Planning the ranges streams every _id of the collection once. Any other _id type (or a mix) gives one partition. Documents whose _id has another type, or none, land in the first range and fail to read there, rather than being skipped.

Each partition is read with find(...).sort({_id: 1}). Filters are not pushed down: where: is evaluated by dataferry. There is no snapshot across partitions.

Co-partitioning (for a merge’s many join) works on a field whose values are integers, strings or ObjectIds.

TLS and connections

The connector uses the MongoDB Rust driver, with one client per source and the driver’s default pool. Remote hosts get TLS by default (TLS).

TLS

Both built-in connectors share one TLS policy:

• Remote hosts use TLS by default, and a connection that asks for TLS never falls back to plaintext.
• Certificates are always verified, chain and hostname. There is no “encrypt but don’t verify” mode, and no way to turn the checks off. To trust a private CA, give its certificate.
• Local connections may use plaintext. When nothing is said, a connection to a remote host requires TLS. One whose hosts are all local (localhost, a loopback address such as 127.0.0.1 or ::1, or a Unix socket path) may use plaintext: PostgreSQL tries TLS and falls back, MongoDB connects to a single local host in plaintext (details below). Private network addresses (10.x, 192.168.x) count as remote.
• For PostgreSQL, TLS is done with rustls, and without a CA file the trusted roots are Mozilla’s. The MongoDB driver does its own TLS.

TLS settings go in each connector’s URL, in the form its ecosystem uses.

PostgreSQL

postgres://app@db.example.com:5432/shop?sslmode=verify-full&sslrootcert=/etc/ssl/corp-ca.pem
host=db.example.com user=app dbname=shop sslmode=require

• Client certificates are not supported yet: sslcert, sslkey and sslpassword are refused (“client certificates are not supported yet”).
• A connection over a Unix socket only is always plaintext, whatever sslmode says.
• With hostaddr, both host (the name the certificate is checked against) and hostaddr must be local for the plaintext fallback.
• An invalid sslmode is an error that lists the valid ones. Errors never echo the connection string.

MongoDB

mongodb://app@mongo-1.example.com:27017,mongo-2.example.com:27017/?replicaSet=rs0&tlsCAFile=/etc/ssl/corp-ca.pem
mongodb+srv://app@cluster0.example.mongodb.net/

A mongodb+srv:// name is a remote host, so it gets TLS. The MongoDB driver does the TLS handshake itself, with its own certificate handling; dataferry decides whether TLS is used and refuses the options that would disable verification. Client certificate options are passed to the driver unchecked.

In tests and local development

Containers on localhost need no TLS settings: the local fallback applies, as in the quickstart. For a database on another host without TLS (a CI service container reached by name, for example), say so explicitly: sslmode=disable for Postgres, tls=false for MongoDB.

Writing a connector

dataferry has no plugin loading. A connector is a Rust crate, and you ship it by building your own binary: the dataferry crate is a library as well as a command, and a three-line main gives your binary every subcommand, the built-in connectors, and yours. The built-in postgres and mongo connectors are registered exactly this way, through the same public calls.

The design is decision 0004. This chapter walks through it with code from the repository that its tests compile and run.

The pieces

A connector crate needs only the first two (plus serde and schemars for its options, and usually async-trait), so it doesn’t pull in the planner or DataFusion.

The Connector trait

A connector declares its type: name, the roles it provides, and an options struct for its fields in connections.yaml, and builds endpoints from them. The repository’s README example, which a test compiles and registers:

use dataferry_registry::{BuildContext, BuildError, Connector, Endpoints, Roles, Secret};
use schemars::JsonSchema;
use serde::Deserialize;

pub struct ClickHouse;

/// A ClickHouse database. Doc comments become editor hover text.
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ClickHouseOptions {
    /// `clickhouse://` URL. A `Secret` is redacted from everything printed.
    url: Secret,
    /// The database written to.
    database: String,
}

impl Connector for ClickHouse {
    const TYPE: &'static str = "clickhouse";
    const ROLES: Roles = Roles::BOTH;
    type Options = ClickHouseOptions;

    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: ClickHouseOptions,
        wants: Roles,
    ) -> Result<Endpoints, BuildError> {
        // No I/O here: `dataferry validate` calls this. Keep the URL for the
        // endpoints, which connect on their first trait call.
        let ClickHouseOptions { url, database } = options;
        let _ = (ctx, url, database);
        todo!("the Source and/or Sink that {wants} asks for")
    }

    /// One line for `dataferry plan` and the run report. No secrets.
    fn describe(&self, options: &ClickHouseOptions) -> String {
        format!("clickhouse, database {}", options.database)
    }
}

• TYPE is the type: value. ROLES says whether it can be a source, a sink or both (Roles::SOURCE, Roles::SINK, Roles::BOTH): validate refuses a flow that uses a connection in another role.
• Options is your connection fields, minus name and type. Derive Deserialize with #[serde(deny_unknown_fields)], so a typo is an error, and JsonSchema, so dataferry schema describes it: doc comments become editor hover text, and defaults and enums carry through. Don’t use the reserved keys name, type or description. Options are validated by deserializing them: the Rust type is the source of truth, and the schema is derived from it.
• Secret for every credential-bearing field. It prints as ***, its schema is marked x-ferry-secret, validate warns when it holds a literal instead of ${ENV}, and its value is redacted from everything the CLI prints. Secrets inside another value (a password in a URL) are returned by derived_secrets (it defaults to none) and redacted too.
• build returns the Source and/or Sink that wants asks for (always a subset of ROLES), in an Endpoints. It must not do I/O: validate calls it to catch bad options without connecting. Endpoints connect on their first trait call. Report bad options as BuildError::option("<field>", "<message>"); the CLI turns the field into a file, line and column.
• describe is the one line plan and validate -v show for a connection. Never put a secret in it.

Your binary

fn main() -> std::process::ExitCode {
    dataferry::Ferry::with_builtins()
        .register(ClickHouse)
        .name("acme-dataferry")
        .main()
}

Ferry is a builder:

Registering a type twice, or a connector whose options use a reserved key, is a bug in the binary: every command then fails at start-up with exit 1 and a diagnostic (F0002) naming the type.

The subcommands and these builder methods are the library’s whole stable surface in this release.

The acme test connector

crates/dataferry/tests/acme-dataferry is a complete third-party binary, written against the public API only and used by the tests: it registers its own postgres and mongo (with the built-ins’ options, and endpoints that panic when used, which proves validate never connects) and a fake type the stock binary doesn’t have.

Its command line:

/// The binary's command line: what `main` runs, and what tests build on.
pub fn cli() -> Ferry {
    Ferry::new()
        .register_state_capable(Postgres, Backend)
        .register(Mongo)
        .register(Fake)
        .name("acme-dataferry")
}

A source-only connector, with a URL check in build that stands in for parsing a connection string, and no I/O:

// ---- mongo: a source only ----

pub struct Mongo;

/// A MongoDB database.
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct MongoOptions {
    /// `mongodb://` or `mongodb+srv://`.
    url: Secret,
    database: String,
}

impl Connector for Mongo {
    const TYPE: &'static str = "mongo";
    const ROLES: Roles = Roles::SOURCE;
    type Options = MongoOptions;

    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: MongoOptions,
        wants: Roles,
    ) -> std::result::Result<Endpoints, BuildError> {
        check_url(&options.url, &["mongodb", "mongodb+srv"])?;
        Ok(endpoints(ctx, wants))
    }

    fn describe(&self, options: &MongoOptions) -> String {
        format!("mongo database {}", options.database)
    }
}

endpoints builds whatever wants asks for. Endpoints also has two optional fields for sinks: a finalizer (restores what prepare loosened for the load, such as dropped foreign keys) and ddl (renders the DDL prepare would run, which plan prints):

fn endpoints(ctx: &BuildContext<'_>, wants: Roles) -> Endpoints {
    let endpoint = Arc::new(Unreachable {
        connection: ctx.connection.to_owned(),
    });
    Endpoints {
        source: wants.source.then(|| endpoint.clone() as Arc<dyn Source>),
        sink: wants.sink.then_some(endpoint as Arc<dyn Sink>),
        finalizer: None,
        ddl: None,
    }
}

Its tests run the binary as a real process (validate against the repository’s example configs and its bad-config set, with every secret checked to never appear in the output), and in-process through run_with (registration mistakes, and the JSON Schema its registered types compose).

Implementing the source and sink

The traits are in dataferry-core. All are async_trait, Send + Sync, and the runner holds them as Arc<dyn Source> / Arc<dyn Sink>.

Source:

• discover() returns the entities: name, fields with canonical types and nullability, and the primary key. For a schemaless store, infer the schema; records that don’t fit it are surfaced at read time as rejected rows (dataferry_registry::rejected), not errors.
• plan_partitions(entity, hint) splits an entity into partitions that together cover every row exactly once, sized by hint.target_rows. Called once per run; the result is stored, never re-planned. When the hint carries a co-partition (a merge’s joined entity), produce exactly its ranges on its column, or return CoPartitionUnsupported (decision 0002).
• read(entity, partition) streams the partition’s rows as Arrow batches matching the entity’s schema. Every source must also accept a Whole partition.
• capabilities() says what the source can do (range scans, predicate pushdown, snapshots).

Sink:

• prepare(target_schema, mode) creates or checks the targets, once per flow.
• begin(partition, lease, strategy) opens a SinkTxn for one partition: write(target, batch) any number of times, then commit() once, or abort(). Everything a partition writes commits together.
• capabilities() lists the idempotency strategies the sink supports. The planner takes the first one compatible with the flow’s mode, in the order TxnWithState, DeletePartitionThenInsert, UpsertByKey, StagingSwap; a flow with none is refused. Whatever you offer must make re-running a partition safe (decision 0001).

Exactly-once: state-capable connectors

TxnWithState commits a partition’s rows and its done-record in one transaction. A connector offers it by implementing StateCapable (a pool() that connects lazily, and build_with_ledger, which builds a sink that calls the given Ledger on its own transaction just before COMMIT), paired with a StateBackend that keeps the state store in the same database, and is registered with register_state_capable. The compiler checks that the connector’s and the backend’s pool and connection types are the same. A sink built with plain build must not list TxnWithState. The acme connector’s version, which does nothing:

/// The driver's pool, connecting lazily: there is nothing to connect to.
#[derive(Clone)]
pub struct Pool;

impl StateCapable for Postgres {
    type Pool = Pool;
    type Conn = ();

    fn pool(
        &self,
        _: &BuildContext<'_>,
        _: &PostgresOptions,
    ) -> std::result::Result<Pool, BuildError> {
        Ok(Pool)
    }

    fn build_with_ledger(
        &self,
        ctx: &BuildContext<'_>,
        options: PostgresOptions,
        wants: Roles,
        _: Pool,
        _: Arc<dyn Ledger<()>>,
    ) -> std::result::Result<Endpoints, BuildError> {
        self.build(ctx, options, wants)
    }
}

/// A state store that panics on first use, like the endpoints.
pub struct Backend;

impl StateBackend for Backend {
    type Pool = Pool;
    type Conn = ();

    fn open(&self, _: Pool) -> (Arc<dyn StateStore>, Arc<dyn Ledger<()>>) {
        (Arc::new(Store), Arc::new(Store))
    }
}

Testing a connector

With the testkit features:

The contract suites return every violation they find, so a test asserts the list is empty; certify returns a certificate, or the first reason it refused one. The built-in connectors pass all four.

What a third-party connector can’t do yet

• Be verified. After a run, dataferry reads a sink back to verify it only for the types it knows how to read: postgres. A third-party sink’s flows are reported as not verified (W0701), which doesn’t change the exit code; what its finalizer couldn’t restore still fails the run.
• Qualify entity names with a namespace, as app.public.orders does for Postgres: that is also registered only for the built-ins.
• Be loaded at run time. Connectors are compiled in. A language-agnostic subprocess protocol is designed to fit later (decision 0004 §7), not built.

Using dataferry as a library

Beyond Ferry, the command functions (plan, run, …) are private in this release: embedding dataferry means running its commands through Ferry::run_with, with your own arguments and output buffers, and reading the exit code and the run report. The crates below the CLI (the DSL, the runner, the state store, the verifier) are published and documented on docs.rs, but their APIs are not stable yet.

Architecture

dataferry is a Cargo workspace of small crates with one-way dependencies. This page maps them, follows a record through a run, and points to the decision records behind the design.

Crates

How they depend on each other:

dataferry                     ─► dataferry-dsl, dataferry-runtime, dataferry-state,
                                 dataferry-verify, dataferry-connector-postgres,
                                 dataferry-connector-mongo
dataferry-dsl                 ─► dataferry-runtime, dataferry-verify
dataferry-verify              ─► dataferry-runtime, dataferry-connector-postgres
dataferry-runtime             ─► dataferry-state
dataferry-connector-postgres,
dataferry-connector-mongo     ─► dataferry-connector-tls

every crate except
dataferry-connector-tls       ─► dataferry-registry ─► dataferry-core

The constraints that matter: a connector crate needs only dataferry-core and dataferry-registry (plus the TLS policy), so a third-party connector doesn’t depend on the planner, the runner or DataFusion; and the built-in connectors register through the same public API a third party uses.

A run, end to end

 ferry/*.yaml
     │  dataferry-dsl: load ─► validate ─► hash ─► ${ENV} substitution
     ▼
 Config ──► dataferry-registry: build each connection (no I/O) ─► Source / Sink
     │
     │  dataferry (CLI): Source::discover, Sink::capabilities
     │  dataferry-dsl: type-check expressions ─► compile flows into
     ▼          transform plans ─► order flows ─► pick strategies
 Plan
     │  dataferry-state: create the run, pin hash + engine version
     ▼
 dataferry-runtime:
     every flow, up front: Sink::prepare ─► plan + store partitions
     workers, each flow once the flows it waits for are done:
        lease (epoch) ─► Source::read (Arrow batches)
                     ─► transform plan: rows per target, or rejected records
                     ─► SinkTxn::write per target ─► dead letters
                     ─► record counts + write-time checksums
                     ─► SinkTxn::commit (with the done-record, under TxnWithState)
     every flow done, after all have stopped: finalize (restore
        constraints and indexes, validate foreign keys)
     │
     ▼
 dataferry-verify: read targets back ─► compare checksums, counts,
     constraints, dead letters ─► evaluate expect: ─► report.md / report.json

Records travel as Arrow RecordBatches from the source to the sink. Expressions are compiled once per flow into DataFusion physical expressions and evaluated over whole batches. A partition’s rows for all its targets go to the sink in one transaction.

Decision records

The design decisions are written down, with their context and the alternatives rejected. They are the authority when this guide and the code seem to disagree on intent:

• 0001: Commit once per partition: one sink transaction per partition, no checkpoints inside one; every sink declares an idempotency strategy.
• 0002: Co-partitioning: how a merge’s joined entity is cut at the driver’s partition boundaries, and the fallback to staging.
• 0003: Config shape: the ferry/ directory, the flow grammar, scopes and sigils, where: propagation, ${ENV}, the config hash.
• 0004: Connector registry: the Connector trait, options and their schemas, secrets and redaction, the CLI layer and exit codes, state-capable connectors.

They were written while the project was called ferry, and use its crate and command names (ferry-core, ferry plan): read them as dataferry-core and dataferry plan.

Troubleshooting and FAQ

Start with the exit code (CLI) and the diagnostic’s code: every message names the file and line it is about, and most say what to do in their help: line.

The config

no config directory at ..., ... is a file; --config expects the ferry/ directory (F0001), no ferry.yaml in ... (F0103). --config points at the directory holding ferry.yaml: not at the file, and not at its parent. Without --config, commands look for ./ferry.

environment variable not set: NAME (F0108). Export it in the shell that runs dataferry. Variables are read only from the environment, and only in connections and runtime.state.path.

`${` in a flow (F0108). Flows can’t use environment variables: write the value into the flow. It then becomes part of the config hash, which is what keeps a resumed run identical to the first attempt.

unknown key (F0105). A typo, or a key in the wrong file or at the wrong level. The help line suggests the closest key.

expected an expression (a string) ... found boolean. An unquoted true, null or number where an expression belongs: quote it.

`index` is a reserved name (F0110). Write $index for the position, or "index" for a field of that name. The same goes for parent, item and key.

Planning

could not discover the entities of ... (exit 5). Either the database can’t be reached (check the URL, the network, credentials and TLS), or the source’s schema can’t be read. For PostgreSQL, every table of the connection’s schema: must have a primary key and supported column types, even tables no flow reads; the message lists each offending table. Point schema: at a schema holding only what you migrate, or fix the tables. For MongoDB, a time-series collection or a field named _ferry_... in the database blocks discovery.

A remote database without TLS fails to connect. Remote hosts require TLS unless you turn it off. For a server reached by name without TLS (a CI service container), add sslmode=disable (Postgres) or tls=false (MongoDB) to the URL.

A certificate isn’t trusted. Certificates are always verified. For a private CA, give its PEM file: sslrootcert= (Postgres) or tlsCAFile= (MongoDB).

no field ... in entity ... (F0110) for a field that exists. Names are case-sensitive. For MongoDB, the schema comes from a sample of 1,000 documents: a field none of them has isn’t in the schema. Check what plan reports for the entity, and see schema inference.

unconstrained numeric has no fixed precision. Declare the Postgres column as numeric(p, s). The message also suggests casting it in a query, but the source reads tables only, not views or queries, so the column type itself must change (or migrate from a copy of the table that has one).

A type error (F0111) on a typed column. The conversion from the expression’s type to type: can lose information. Convert explicitly in the expression (int(x), int(round(x))), or add allow_lossy: true if you accept what the lossy table says will happen.

flow ... sets on_error ... but this runner applies on_error to the whole run (F0112). Per-flow on_error and max_errors aren’t supported yet: set them in runtime.yaml.

Running

The run stopped with failed partitions (exit 6). The stderr lines F0602 say which partition and why. Fix the cause, then dataferry retry --failed <run-id>. If the fix is a change to the config, the hash changes: start a new run instead.

Records are rejected that look fine. Read the reason in the dead-letter table or in the error. Common ones: int() of text that isn’t exactly an integer (' 3', '3.0'); a MongoDB field with mixed types, read as json, so int() sees "7" with its quotes; a NOT NULL key column with a null; a MongoDB document with a field the sample didn’t see.

prepare fails on a table that exists. The existing table doesn’t fit what the flow writes: a column type that differs (an integer column for an int64 value), a NOT NULL column the flow doesn’t write, or, for upsert, no unique index on exactly the key columns. The message lists every problem. Change the table, or the flow’s types.

has no key, so re-running a partition can't replace the rows it wrote before. Give the target a key:, or keep the state store on the sink (runtime.state.connection: <sink>), the only case where a keyless target can be written safely.

resume is refused (exit 8). The config’s meaning or the dataferry version changed since the run started. Tuning settings (workers, memory_budget, partition_rows, the source limits) don’t count, so you can change those. Anything else: restore it, or start a new run.

resume waits before doing anything. After a crash or kill -9, the dead process’s leases have to expire first, at most 30 seconds. After a Ctrl-C there is no wait.

A run uses too much memory. Lower memory_budget (it sizes batches and read-ahead) or workers, and resume: both are tuning settings. A merge that stages a large joined entity sorts it in a quarter of the budget, spilling the rest to the work directory.

A run is too slow, or too hard on the source. More workers read more partitions at once. source_max_connections and source_rows_per_sec cap what the run asks of the sources. Keep the state store on the sink so partitions aren’t written twice after a crash.

Verification

Verification fails on tables that had rows before the run. It compares whole tables with what the run wrote, so rows it didn’t write count as differences. That is by design: it is meant for full loads into empty tables. Load into empty tables, or read the report’s Tables section knowing that.

done partitions have no write-time record. The work directory of the run was lost or changed (a different --work-dir, a new container). Verification needs the records each partition wrote there: keep the work directory for the life of a run, including across resume.

An assertion fails. The report shows both values. If the mapping is right and the assertion is wrong, fix the assertion: that changes the config hash, so start a new run.

FAQ

Which databases does it support? MongoDB and PostgreSQL as sources, PostgreSQL as the sink, in this release. Others can be added as connectors in your own build.

Is it safe to stop a run? Yes. Ctrl-C, a crash or a lost connection loses at most the partitions in flight; resume continues from the committed ones and writes nothing twice (Running migrations).

Is it exactly-once? For flows whose sink is the state store’s connection, yes: a partition’s rows and its done-record commit in one transaction. Other flows may write a partition twice after a crash, and the second write replaces the first.

Does it keep the target in sync with the source? No. A run copies what the source holds when it reads it. There is no change data capture, and upserts never delete.

Can several machines share a run? No: one process runs a run. Use workers for parallelism within it.

Can I run it in CI or a container? Yes. validate needs no database. For run, keep the state store in Postgres (so it survives the container) and the work directory on a persistent volume (verification needs it).

Where are the logs? There are none beyond what the commands print: stdout for output, stderr for diagnostics. The report in .ferry/runs/<run-id>/ is the record of a run.

How do I see what a run will do without running it? dataferry plan. It shows the DDL, the strategies and every type, and writes nothing.

Limitations and roadmap

0.1.0-beta.1 is a beta. What it does, it does carefully: every partition commits once, every run is verified, every record is accounted for, and the connectors pass a conformance suite of 50 runs killed mid-partition and resumed. But it does fewer things than a 1.0 will, and some behaviour may change before then. The config grammar (version: 1) and the exit codes are meant to stay.

What this release doesn’t do

Connectors and sources

• Sources are MongoDB and PostgreSQL; the only sink is PostgreSQL. There is no MongoDB sink.
• PostgreSQL sources read tables, not views or queries, and every table of the connection’s schema needs a primary key and supported types: one table with an enum, a domain, an unconstrained numeric or no primary key blocks discovery of the whole schema. Composite and non-integer, non-text, non-UUID keys are read as one partition.
• MongoDB schemas are inferred from a sample of 1,000 documents.
• Reads have no consistent snapshot: migrate from a quiet source or a snapshot.
• Client TLS certificates aren’t supported for PostgreSQL.
• Connectors are compiled in: there is no plugin loading, and a third-party sink can’t be verified after a run.

Runs

• One process runs a run. There is no distributed mode.
• on_error and max_errors apply to the whole run, not per flow.
• A partition commits once, at its end: there are no checkpoints inside a partition, so a crash redoes the partitions in flight.
• Flows are ordered by depends_on: and by the tables they read; foreign keys between flows aren’t discovered.
• No progress display, and no command to replay dead letters.
• resume needs the same dataferry version that started the run.
• Every flow is prepared and partitioned when the run starts, so a flow reading a table another flow writes needs that table to exist already.
• A run that stops before finalizing leaves the foreign keys and indexes it dropped for the load dropped, until a resume or retry finalizes it.
• The state store’s tables are created in the first schema of the state connection’s search_path, not in its schema:.

Mapping

• Expressions have no date and time functions, and missing() works on the current record only, not on joined entries.
• Integer arithmetic isn’t checked for overflow: it wraps around (Expressions).
• columns: "*" can’t be combined with other columns, and a nest column can’t contain another nest.
• Flows can’t be generated from a source schema yet: every flow is written by hand.

Verification

• It assumes full loads into empty targets: rows already in a table that the run doesn’t rewrite fail the checks.
• The write-time checksums it compares against are kept in the work directory only: a run resumed on another machine, or with another --work-dir, can’t be verified.
• There is no standalone verify command; resume on a finished run verifies it again.

The CLI

• No shell completions, no progress output, no log files.

Roadmap

The next release, v0.2, is planned in four themes. They are plans, not promises, and their order may change:

• Hardening: correctness first, such as more precise redaction, the verifier’s records in the state store instead of the work directory, and lease and dead-letter hygiene.
• Features for real-world migrations: foreign-key discovery with automatic flow ordering, generating flows from a source schema, per-flow on_error and max_errors, partitions on composite keys, date and time functions, missing() on joins, and a MongoDB sink.
• Performance: a benchmark harness first, then faster paths where it shows they matter.
• Release: regular tagged releases, and editor completion from the JSON Schemas out of the box.

The issue tracker is the place to ask for what you need, or to report what doesn’t work.

CLI

dataferry [OPTIONS] <COMMAND>

The help texts below are the binary’s own (dataferry <command> --help); a test checks them against it.

Global options

The options are the same for every command, and may come before or after it (dataferry --config ferry plan and dataferry plan --config ferry).

Move data between databases, one committed partition at a time.

Usage: dataferry [OPTIONS] <COMMAND>

Commands:
  validate  Check the config without connecting to anything
  plan      Print what a run would do: discovered entities, types, DDL and strategies. Reads from sources, writes nothing
  run       Plan, then run every flow. Prints the run id first
  resume    Continue an unfinished run
  retry     Re-run the failed partitions of a finished run
  schema    Print the JSON Schemas for this binary's config files, including the connection types it has registered
  help      Print this message or the help of the given subcommand(s)

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

  -V, --version
          Print version

Output

• stdout carries the command’s output: validate’s summary, the schemas, the plan, the run id and run summaries. dataferry plan > plan.txt captures only the plan.
• stderr carries diagnostics and notes.
• Every value substituted from the environment, and every value of a secret connection field, is replaced with *** in both, and in every file a run writes.

A diagnostic has a severity, a stable code, a message, the file, line and column it points at (after includes), the field path, and often a help line:

error[F0303]: flows[products].sink: flow `products` writes to connection `shop`, but the `mongo` connector in this build has no sink
 --> ferry/flows/products.yaml:4:11
  |
4 |     sink: shop
  |           ^^^^
  = help: `mongo` provides: source
config is invalid: 1 error

With --format json, each diagnostic is one JSON object per line on stderr, with the keys severity, code, message, file, line, column, path and help. validate --format json also prints a summary object on stdout:

{"severity":"error","code":"F0105","message":"unknown key `wrokers`","file":"ferry/runtime.yaml","line":7,"column":3,"path":"runtime.wrokers","help":"did you mean `workers`?"}
{"valid":false,"errors":1,"warnings":0}

validate

Check the config without connecting to anything

Usage: dataferry validate [OPTIONS]

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

Loads every file of the config, substitutes ${ENV}, builds every connection through its connector with the roles its flows use (without connecting), and checks: the files, keys and values; names and references; depends_on:; connection options; that each connection is used in a role its connector has; that runtime.state.connection is a connection of the backend: type that can hold the state store; and the expressions’ syntax and reserved names. Reports every problem, then `<name>` is valid: <n> connections, <n> flows (exit 0) or config is invalid: <n> errors (exit 3).

What it can’t check without connecting (entities, fields, types) is plan’s.

schema

Print the JSON Schemas for this binary's config files, including the connection types it has registered

Usage: dataferry schema [OPTIONS]

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --out <DIR>
          Write one `<file>.schema.json` per config file into DIR instead

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

Prints one JSON object keyed by file name (ferry.schema.json, connections.schema.json, runtime.schema.json, flows.schema.json), or with --out <dir> writes those four files there. The connections schema has one branch per connection type registered in the binary. See JSON Schemas.

plan

Print what a run would do: discovered entities, types, DDL and strategies. Reads from sources, writes nothing

Usage: dataferry plan [OPTIONS]

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

validate, then discovers every source the flows read, asks every sink for its capabilities, type-checks and compiles every flow and every expect: assertion, and prints: the config hash, the state store and the error policy; the connections; the run order; and per flow its driver and key, sink and mode, strategy (and why, when it isn’t TxnWithState), partition size, joins and their strategies, nest orderings, filters, write order, targets with their columns and types, assertions, and the DDL the sink would run. Runs no DDL and writes nothing.

Exit 0; 3 for a config error; 4 when the plan is refused (an unknown entity or field, a type error, a flow with no idempotency strategy, a cycle that only appears through the tables flows read, a per-flow setting the runner can’t apply; a depends_on: cycle is a config error, exit 3); 5 when a source or sink can’t be reached or a source’s schema can’t be read.

run

Plan, then run every flow. Prints the run id first

Usage: dataferry run [OPTIONS]

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

Plans, creates the run, prints run <run-id> and each flow’s strategy, runs every flow, verifies each sink, writes the report and prints the summary (Running migrations).

resume

Continue an unfinished run

Usage: dataferry resume [OPTIONS] <RUN_ID>

Arguments:
  <RUN_ID>
          The run to continue

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

Plans again, checks that the config hash and the engine version are the ones the run pinned (exit 8 if not), and runs every partition that isn’t done or failed. Then verifies and reports, like run.

retry

Re-run the failed partitions of a finished run

Usage: dataferry retry [OPTIONS] --failed <RUN_ID>

Options:
  -c, --config <DIR>
          The `ferry/` config directory
          
          [env: FERRY_CONFIG=]
          [default: ferry]

      --failed <RUN_ID>
          The run whose failed partitions to re-run

      --color <WHEN>
          Colour diagnostics: `auto` colours a terminal unless NO_COLOR is set
          
          [default: auto]
          [possible values: auto, always, never]

      --format <FORMAT>
          `json` prints diagnostics as JSON lines, for CI and editors

          Possible values:
          - text
          - json: One JSON object per line
          
          [default: text]

  -v, --verbose...
          More detail, e.g. how each connection was understood

  -q, --quiet
          Only errors: no warnings, no summary

      --work-dir <DIR>
          Where `run` keeps write-time records, reports and file-backed state. Default: `.ferry` next to the config directory
          
          [env: FERRY_WORK_DIR=]

  -h, --help
          Print help (see a summary with '-h')

Like resume, after putting the run’s failed partitions back to pending.

Environment variables

Exit codes

Scripts and CI can rely on these: a code never changes meaning once released.

Diagnostic codes

Config reference

Every key of every file in a ferry/ directory, for grammar version: 1. The grammar and the reasons behind it are in decision 0003, and the connector fields in decision 0004. The JSON Schemas are in schemas/v1, and dataferry schema prints them for a binary, connection types included.

Unknown keys are errors at every level, so a typo fails dataferry validate instead of being ignored.

Paths read from the top of the file: runtime.state.backend is backend: under state: under runtime:, flows[] is each entry of the flows: list, and <table> stands for a name you choose.

ferry.yaml

Flows are not written here: they live under flows/.

connections.yaml

description: is reserved and rejected in v0.1. Every other key belongs to the connection’s type. A field marked secret should be a ${ENV} reference: a literal value is warned about (W0301), and its value is shown as *** in everything dataferry prints.

type: postgres

Source and sink, and can hold the state store (runtime.state.backend: postgres).

type: mongo

Source only.

runtime.yaml

Every key is optional, and runtime: {} is valid.

flows/**/*.yaml

Every .yaml and .yml file under flows/, at any depth, holds flows: (flows: [] keeps a placeholder file). Names are unique across every file, and the order flows are written in doesn’t matter.

A flow

A merge: from: as a list

A target: to.<table>

A column: to.<table>.columns.<column>

A column spec is an expression string (status: lower(status)), a typed column or a nest column. Values must be YAML strings: quote "null", "true" or "5".

Expressions and sigils

Expressions are SQL-like (lower(status), int(round(total * 100)), deleted_at == null). A bare name is a field of the current scope: the driver record, the current array element under unnest: <path>, the current joined row under unnest: <alias>, or the current nested row.

${ENV}

${NAME} in a string value is replaced from the environment, in connections.yaml entries (included connections: too) and in runtime.state only. A variable that isn’t set is an error (exit 3) that names it, and substituted values are shown as *** in everything dataferry prints. ${NAME} is the only form: ${file:...} is an error. A ${ in a flow is an error too, so what a flow writes can’t change between a run and its resume.

The config hash

A run pins a hash of the config, and dataferry resume refuses (exit 8) when it changed. It is taken before ${ENV} substitution, with defaults filled in, and doesn’t depend on file layout or the order flows are written in. The settings that only change how fast a run goes are left out, so a run can be resumed with more memory: workers, memory_budget, partition_rows, source_max_connections and source_rows_per_sec, in runtime.yaml and in a flow’s runtime: (0003 §10).

Test only: FERRY_LEASE_TTL_MS

Not a config key, and not for production. Leases last 30 seconds without a heartbeat (every 10), set by the binary, not the config. The FERRY_LEASE_TTL_MS environment variable overrides that, in milliseconds from 1000 to 600000, with the heartbeat at a third of it, so the chaos and conformance suites can resume at once after a kill. Any other value is a usage error (exit 2, F0601), and run, resume and retry warn (W0601) whenever it is set. A short lease expires under a live worker on a slow commit, a database pause or a network stall: the worker loses its partition (fenced, so nothing is written twice) and the run redoes or fails work it had done.

JSON Schemas

Every kind of config file has a JSON Schema (draft 2020-12), so a YAML editor can complete keys, show their documentation, and flag unknown keys and wrong values as you type.

Getting them

The schemas depend on the binary: the connections.yaml schema has one branch per connection type the binary registered, with that connector’s options. So take them from the binary you run:

dataferry schema --out schemas

That writes schemas/ferry.schema.json, connections.schema.json, runtime.schema.json and flows.schema.json. Without --out, dataferry schema prints all four in one JSON object, keyed by file name.

The repository also keeps the grammar’s schemas in schemas/v1/, generated from the config model and checked by its tests. They leave connection entries open (only name and type are checked), because the connector fields depend on the binary.

Editor setup

VS Code with the YAML extension (redhat.vscode-yaml), in .vscode/settings.json, with the schemas written to schemas/ next to ferry/:

{
  "yaml.schemas": {
    "schemas/ferry.schema.json": "ferry/ferry.{yaml,yml}",
    "schemas/connections.schema.json": "ferry/connections.{yaml,yml}",
    "schemas/runtime.schema.json": "ferry/runtime.{yaml,yml}",
    "schemas/flows.schema.json": "ferry/flows/**/*.{yaml,yml}"
  }
}

Or per file, with a modeline on its first line (any editor using yaml-language-server, such as Neovim, Helix or Zed):

# yaml-language-server: $schema=../../schemas/flows.schema.json

JetBrains IDEs map schemas to files under Settings → Languages & Frameworks → Schemas and DTDs → JSON Schema Mappings.

A file reached through include: holds flows:, connections: or both: map it to the schema of what it holds.

What they cover

• Every key of every file, with its description, default and allowed values, and unknown keys as errors.
• The type: of a connection, and each type’s own fields, with secret fields marked (writeOnly, x-ferry-secret).
• ${NAME} where a connection field expects a number or a boolean: those fields also accept a ${NAME} string, since substitution happens before the value is read.

What they can’t cover, and dataferry validate does: names that refer to something defined elsewhere (the connection in from: and sink:, flows in depends_on:), ${ENV} rules, and expressions. Entity and field names need the databases, so they are dataferry plan’s.

The config reference lists the same keys as these schemas, and a test checks that the two agree.

Decision records

The architecture decisions behind dataferry, each with its context, the decision, the alternatives rejected and the consequences. They are kept in the repository under docs/decisions/ and included here as written, with their amendments.

They were written while the project was named ferry, so they use its crate and command names: ferry-core is dataferry-core, and ferry plan is dataferry plan. Names that didn’t change: the ferry/ config directory, the .ferry/ work directory, the _ferry_* tables and the FERRY_* variables.

Where a record and this guide disagree, the guide describes what this release does. The one point where the release does less than the records describe: per-flow on_error and max_errors (0003 §6) are accepted only with the run-wide values, as the amendment in §6 explains.

0001: Commit once per partition in v0.1

Status: accepted (Phase 0, ITS-109)

Context

The spec’s Source::read(partition, from: Option<Checkpoint>) implies checkpoints inside a partition, while Sink::commit happens per partition. Mixing the two means either long sink transactions (a partition of several minutes held open in Postgres causes bloat and lock pressure) or commits inside a partition, which breaks the “delete the partition’s rows, then insert” idempotency strategy because each chunk would delete the previous chunks’ rows.

Decision

• The runner opens one SinkTxn per leased partition, writes every target batch for it, and commits once.
• Transaction length is bounded by partition size: the planner passes PartitionHint::target_rows, and sources split entities accordingly.
• Source::read takes no resume checkpoint. A partition that didn’t commit is re-read from the start.
• Checkpoint stays in ferry-core but is unused in v0.1. It is reserved for chunked commits, which v0.2 can add for sinks whose strategy tolerates them (UpsertByKey, TxnWithState).

Consequences

• Resume granularity is one partition: a crash loses at most one partition’s work per worker.
• Every sink must declare an IdempotencyStrategy compatible with the write mode; SinkCaps::strategy_for returns None otherwise and the planner refuses the flow.
• Under TxnWithState, the sink checks the lease epoch inside the commit, so a worker whose lease was superseded cannot commit.

0002: Co-partitioning handshake for streaming merges

Status: accepted (Phase 0, ITS-112). Needs review from both tracks.

Context

A merge like orders + order_items → documents can stream partition by partition only if every child row is read in the same unit of work as its parent. That needs the planner (Track B) to tell the source partitioner (Track A) where to cut. Agreeing on this late would force both sides to be rewritten.

Decision

1. The planner plans the driver entity normally (e.g. orders on id).
2. It derives CoPartition::from_partitions("order_id", &driver): the driver’s range boundaries, applied to the child’s join column. This only works if the driver’s partitions are contiguous key ranges on one column that are unbounded at both ends; otherwise it returns CoPartitionUnsupported.
3. It plans the child with PartitionHint { co_partition: Some(..), .. }. A source must then produce exactly CoPartition::ranges(), in order, on that column, ignoring its own sizing. A source that can’t (no such column, no efficient range scan on it) returns CoPartitionUnsupported.
4. align(&driver, &child) pairs partition i with partition i and refuses lists whose bounds differ. The runner leases a pair as one unit of work and feeds both streams to the merge’s TransformPlan.
5. On CoPartitionUnsupported, the planner falls back to staging the smaller side and hash-joining. It shows the chosen strategy in ferry plan.

Consequences

• Child partitions can be very uneven (one order with 100k items). That’s accepted for v0.1. The child’s target_rows is ignored when co-partitioning.
• Boundaries are KeyValues, so driver key and child join column must compare as the same kind (both integers, both strings, both bytes).
• Track A (A1, A2) implements step 3 for Postgres and Mongo. Track B (B6) implements steps 1, 2, 4 and 5.

0003: Config shape

Status: proposed. The points that were open were decided on 2026-10-16 (see Open questions); the record still needs review from both tracks. Blocks B1 (0XT-254). Once accepted, B2, B4, B5, B6, B7 and B8 are edited to use its words (see §11). The shape and validation of each connection type’s own fields belong to 0004.

Context

The config layout is already agreed: a ferry/ directory holding ferry.yaml (version and name), connections.yaml, runtime.yaml, and every YAML file under flows/, with include: as an escape hatch. from: is one entity or a list of joined entities (merge), and to: can name several target tables (split).

What the agreement leaves open is the grammar inside a flow, and the tickets describe the same ideas in a second vocabulary: B4 has columns:, split:, where: and key:, B5 has unnest: path:, B6 has merge: with a driver entity, and B8 has expect:. Our draft from: list also lacks two things B6 needs: which entity is the driver, and the cardinality of each join. B1 can’t freeze a JSON Schema, and the golden config (B1, B7, S2 and S3 test against it) can’t be written, until there is one vocabulary.

The grammar has to fit the ferry-core contracts as they stand:

• TransformPlan::inputs() lists the driver first. Every output row is attributed to one driver record, and a record is written in full or rejected in full.
• TransformPlan::write_order() puts parents before children, and TransformOutput::validate enforces it.
• One SinkTxn per partition covers every target of a flow (0001), so a flow writes to exactly one sink.
• Sink::prepare and SinkCaps::strategy_for take one WriteMode per flow.
• Co-partitioning (0002) needs the driver’s partition column and the child’s join column.

Decision

1. Grammar at a glance

Each line names a key, its type, and its default. ? means optional, and | separates alternatives. This is a summary, not literal YAML.

# ferry.yaml
version: 1                        # int, required. Version of this grammar
name: <ident>                     # required. Names the migration in state and reports
include: [<path-or-glob>, ...]    # ? escape hatch, see §8

# connections.yaml
connections:
  - name: <ident>                 # required, unique, no dots
    type: <connector type>        # required, a key in the ConnectorRegistry
    # ...type-specific fields: owned by 0004. Reserved keys: see §3

# runtime.yaml (every key optional)
runtime:
  state: {backend: file, path: <dir>}    # default backend: file
       | {backend: postgres, connection: <conn>}   # a state-capable type, see §4
  workers: auto | <int>           # default auto (one per core)
  memory_budget: <size>           # e.g. 512MiB, default 1GiB
  partition_rows: <int>           # PartitionHint::target_rows, default 100000
  on_error: fail | dead_letter    # default fail
  max_errors: <int>               # ? default unbounded (plan warns)
  source_max_connections: <int>   # ? A5 throttle
  source_rows_per_sec: <int>      # ? A5 throttle

# flows/*.yaml (any file name, any depth)
flows:
  - name: <ident>                 # required, unique across every file
    from: <conn>.<entity>         # one entity
       |  [<from entry>, ...]     # or a merge, see §5.2
    sink: <conn>                  # required, the one connection this flow writes to
    mode: insert | upsert         # default insert
    where: <expr> | {<name>: <expr>, ...}   # ? filter on the driver record
    depends_on: [<flow name>, ...]          # ? see §7
    runtime: {<overridable key>: ...}       # ? see §6
    to:                           # required, one or more targets (split)
      <table>:
        parent: <table>           # ? another target of this flow
        unnest: <path> | <many alias>       # ? one row per element
        key: [<column>, ...]      # required if mode is upsert
        where: <expr> | {<name>: <expr>}    # ? filter on this target's rows
        columns: "*" | {<column>: <column spec>, ...}   # required
    expect: {<name>: <assertion>, ...}      # ? see §5.9

Unknown keys are errors at every level. A typo must fail at load time, not be ignored.

2. The ferry/ directory and loading

• Finding it. --config <dir> if given, otherwise ./ferry. The directory must contain ferry.yaml. That file is what marks it as a ferry config.
• Well-known files. ferry.yaml is required. connections.yaml is required once any flow exists. runtime.yaml is optional, and every setting has a default. Each file may contain only its own top-level key: connections.yaml only connections:, runtime.yaml only runtime:, a flow file only flows:. Both .yaml and .yml are accepted. Having connections.yaml and connections.yml side by side is an error.
• Flow files. Every *.yaml or *.yml file under flows/, at any depth, is loaded. Other files (a README.md, say) are ignored, and so are files and directories whose names start with . (editor swap files). An empty flows: list is allowed. A config with no flows at all is an error.
• Order. Files are loaded sorted by their path relative to ferry/, compared bytewise as UTF-8. The order only affects which file an error points at first. The model itself does not depend on it (§10).
• Single document. A file with several YAML documents (---) is an error, which keeps every error’s file and line exact.
• Names. Connection and flow names match [a-z][a-z0-9_]*. A flow name: that appears twice, in one file or in two, is an error naming both files and lines.
• Error locations. Every error after loading, merging, including and type-checking carries the original file, line and field path (flows/orders.yaml:14 flows[orders].to.order_items.key).
• YAML 1.2. The loader uses the YAML 1.2 core schema, so on, yes and no are strings. Editors and other tools often still resolve YAML 1.1, where those words are booleans, so the grammar never uses them as keys. That is why joins say keys: rather than on:.

3. Connections and how flows reference them

connections.yaml holds a list whose entries are {name, type, ...}. name and type are owned here. Everything else in an entry is type-specific and belongs to 0004. type is looked up in the ConnectorRegistry. The built-in postgres and mongo types register the same way a third-party connector would, and an unknown type is an error that lists the registered ones.

Reserved keys. name, type and description are reserved for this record in every connection entry, so a key 0003 adds later can’t collide with a connector’s own field. ConnectorRegistry::register rejects a connector whose Options schema uses one (0004 §2). description has no meaning in v0.1 and is rejected as an unknown key until this record defines it.

Flows reference connections in exactly two ways:

• An entity, as <conn>.<entity>, in from: and in merge entries. The string is split at the first dot: what comes before it is the connection name (which can’t contain a dot), and the rest is the entity name exactly as the connector’s discover() reports it (Entity::name). So app.public.orders is entity public.orders on connection app. Amended in S3 (2026-10-17): a connector type may also accept the entity qualified by the namespace its connection reads. For postgres that is the connection’s schema:, so on a connection with schema: public, app.public.orders names the table orders that discover() reports, as does app.orders. An entity discovered under the full dotted name still wins. Which types have a namespace is set where the binary registers its connectors, next to how their sinks are verified.
• A connection, by bare name: sink:, and runtime.state.connection.

Target tables (the keys under to:) are plain identifiers with no dots. Which schema or database they land in is a property of the sink connection (0004). A given (sink, table) pair may be written by only one flow. Two flows inserting into one table would undo each other’s idempotency strategy.

4. runtime.yaml

runtime: holds process-wide settings, listed in §1. state picks the A4 backend. backend: file keeps state in a local directory. Any other backend is the type of a state-capable connector (postgres in the stock binary), and connection: must name a connection of that type. Both are checked by ferry validate through the registry (0004 §8).

A flow gets TxnWithState (data and the partition’s done-record in one transaction) only when runtime.state.connection is the flow’s sink:. Every other flow, including every flow under backend: file, gets the next strategy SinkCaps::strategy_for offers, and ferry plan says why (0004 §8). Batch size is not a setting: the runner derives it from memory_budget (A5).

on_error: fail stops the run at the first error row. on_error: dead_letter writes the rejected source record to _ferry_dlq on the flow’s sink (A7) and continues until more than max_errors records have been rejected. There is no skip: a record is never dropped without a trace.

5. Flows

5.1 Shapes

There are no split: or merge: keywords. They are shapes the grammar already expresses:

5.2 from: and merges

The single form is a string: from: shop.orders.

The merge form is a list of entries:

from:
  - entity: app.public.orders          # the driver: first, and no join
    as: o
  - entity: app.public.customers
    as: c
    join:
      cardinality: one
      keys: {id: o.customer_id}        # c.id = o.customer_id
  - entity: app.public.order_items
    as: i
    join:
      cardinality: many
      keys: {order_id: o.id}
      strategy: co_partition           # ? auto (default) | co_partition | stage

• The driver is the first entry, and it must not have join:. Every other entry must have one. Position and structure agree, so the driver can’t be ambiguous, and it maps directly onto TransformPlan::inputs()[0]. Output rows are attributed to driver records, and ferry plan prints driver: app.public.orders.
• as: is the alias that expressions use. It defaults to the last dot-separated part of the entity name, and aliases must be unique within a flow. In the flow’s columns, a bare name refers to a driver field, so status and o.status mean the same thing there.
• join.keys: is required. It maps a column of this entry to a qualified column of an earlier entry, each pair being an equality, and several pairs make a composite key. Only equi-joins on columns are allowed, which is exactly what co-partitioning (0002) and hash joins need, and it lets the planner refuse a join with no keys. A join may reference the driver or an earlier one entry (lookup chains such as orders → customers → regions). It may not reference a many entry.
• join.cardinality: is required and has no default:
  • one: at most one match per driver record. No match gives null joined columns. A second match found at run time rejects the affected driver record (error row, then on_error), and never fans out. Sampling at plan time is advisory only.
  • many: zero or more matches. A many alias can be used only as the source of nest: (§5.8) or of unnest: (§5.7). Anywhere else it would fan out the driver, and the plan fails.
• Every join is a left join from the driver. To drop drivers with no match, use a where: on the joined column.
• join.strategy: defaults to auto, where the planner picks between co-partitioned streaming and staging (0002 step 5) and shows the choice in ferry plan. co_partition pins streaming and makes the plan fail, instead of falling back, if co-partitioning is impossible. stage forces staging.
• Merge entries may come from different connections. The planner names inputs <conn>.<entity>, so two entities with the same name don’t collide in PartitionInputs. A cross-connection join can only be staged.

5.3 sink:, to:, and write order

sink: names the single connection the flow writes to, which follows from 0001 (one SinkTxn per partition). Each key under to: is a target table. parent: <table> makes one target the child of another in the same flow. That one declaration drives three things:

• Write order. Parents come before children, ordered topologically on parent: with ties broken by table name. Nothing else affects it, so the order doesn’t depend on the order of keys in the YAML file. This becomes TransformPlan::write_order().
• Filter propagation (§5.6).
• Foreign keys. If a child maps every one of its parent’s key: columns from $parent.<col>, the planner emits a ForeignKey from those child columns to the parent’s key. A6 decides whether that FK is enforced during the load.

A parent: cycle, or a parent that isn’t a target of this flow, is an error.

5.4 columns:, scopes and sigils

columns: is an ordered map from target column to column spec, and its order is the column order in the target DDL. A spec is one of:

• a string, which is an expression (B2): status: lower(status)
• an object, for when the type matters: {expr: <expr>, type: <canonical type>, nullable: <bool>, allow_lossy: <bool>}. Type names follow CanonicalType (int64, decimal(12,2), utf8, timestamp(UTC), uuid, json, …). B3 owns the exact list. allow_lossy: true is the explicit opt-in B3 requires for a lossy conversion to type:. A cast written in the expression itself (int(qty)) also counts as explicit. A value that doesn’t fit at run time is an error row. See the amendment below for the exact rule and table.
• a nest object (§5.8).

columns: "*" copies every declared field of the entity unchanged (IdentityPlan). It is allowed only on a target with no parent: and no unnest: in a single-entity flow, and can’t be combined with other columns in v0.1. It keeps the ~190 plain copies in a 200-table migration to about six lines each. Generating whole flows (wildcards, ferry infer) is out of scope for v0.1.

Values must be YAML strings. A bare null, true or 5 is an error telling you to quote it ("null"), so YAML’s type guessing never silently changes an expression. Quote any expression that starts with *, [, {, &, !, |, >, ', ", % or @, or that contains : .

Scope. Every target evaluates its expressions in one scope:

Sigils (B2 reserves the $ prefix):

A child doesn’t see the driver record’s raw fields directly. It sees them through $parent, so values it shares with the parent are mapped once, in one place. Missing and null keep the meaning B2 gives them (read through _ferry_missing).

5.5 mode: and key:

mode: is set per flow, because Sink::prepare and strategy_for take one WriteMode per flow:

• insert (default). This run creates the target rows. Re-runs are made safe by the sink’s strategy (TxnWithState when the state store is on the flow’s sink connection, see §4, otherwise DeletePartitionThenInsert or StagingSwap).
• upsert. Rows may already exist and are replaced by key. Every target must declare key:, and the plan error names any that doesn’t (B4). Children with a parent: get replace-children semantics (A3): the children of each parent in the batch are deleted and then rewritten. Rows for records that are filtered out are never deleted from the target. Upsert doesn’t propagate deletes.

key: lists target columns, and they become the target’s primary key in TargetTable. Key columns are implicitly nullable: false. In insert mode key: is optional but recommended. ferry plan warns about any child target without one. There is no implicit generated key. A child that needs a surrogate key maps it explicitly: id: $key.

The config never names an idempotency strategy. The planner derives it from mode and SinkCaps::strategy_for, shows it in ferry plan, and refuses the flow when the result is None (B7).

5.6 where: and the propagation rule

where: is either one expression or an ordered map of named predicates that must all hold:

where: deleted_at == null
# or
where:
  not_deleted: deleted_at == null
  not_test: customer.id != 'test'

The name of the first predicate that fails is what reports and filtered counts show. An unnamed where: is reported by its expression text. A predicate that evaluates to null counts as false. A predicate that errors doesn’t filter: it is an error row, and the record is rejected.

There are two levels, and one rule connects them.

• Flow where: applies to the driver record, in driver scope (joined one columns included). When it fails, the record is filtered: it has no rows in any target and is counted once. A flow where: may be pushed down to the source (predicate_pushdown) only if the pushed form keeps the missing-versus-null meaning exactly. Otherwise it is evaluated in the transform.
• Target where: applies to that target’s rows, in that target’s scope (§5.4). It is evaluated against source values, not against mapped output.

Propagation rule. When a row is filtered out of a target, every row that reaches it through parent: (children, grandchildren, and so on) is dropped with it. Filters never travel upward or sideways: a sibling target with no parent: link keeps its rows. Filtering is not an error, so the record isn’t rejected and the rows that remain are written. Errors keep their own rule from 0.4: any error in any row rejects the whole driver record.

In the golden dataset this is why document …0004 writes neither the order nor its item (a flow where:), and why a target where: on orders would also drop that order’s order_items.

5.7 unnest:

unnest: <path> writes one row per element of an array. The path is resolved in the scope of the target’s parent: the parent’s element if the parent is itself unnested, otherwise the driver record. A target without parent: unnests from the driver record and has no $parent. unnest: <many alias> writes one row per joined row of a many join.

• An empty array, a missing field and an explicit null all produce zero children, and the parent is still written. The golden dataset pins the first two, and treating null the same way matches how B2 treats null and missing elsewhere. It is not configurable.
• A value that isn’t an array (items: "x") is an error row, so the record is rejected.
• Unnests can nest (order_items → item_options with parent: order_items, unnest: options). $index is always the innermost index.

5.8 nest:

A nest builds an array column from a many join:

items:
  nest: i                 # a many alias
  order_by: [line_no]     # ? in the nested row's scope; "<expr> desc" sorts descending
  where: qty > 0          # ? drops nested rows from the array
  columns:                # same column specs as a target
    sku: sku
    qty: qty

• Deterministic order. The planner always appends the nested entity’s primary key (the columns not already listed) to order_by as the final tie-breaker. Without order_by, the order is that primary key. ferry plan shows the full effective ordering. The same input under any partitioning gives the same array (B6).
• A driver with no matches gets [], not null.
• The column’s canonical type is List(Struct(...)). How a sink stores it (Postgres jsonb, say) is the sink’s decision.

5.9 expect:

expect: is an ordered map of named assertions. The names appear in the run report.

expect:
  order_count: count(orders) == source.count()
  order_total: sum(orders.total_cents) == source.sum(round(total * 100))

• An assertion compares two aggregates with ==, !=, <, <=, > or >=. Arithmetic is allowed on either side (abs(a - b) <= 1).
• Target side. The aggregates count(<table>), count, sum, min, max and count_distinct, over <table>.<column>. The table qualifier may be omitted when the flow has one target. They are computed by querying the target after the run.
• Source side. source.<agg>(<expr>) is evaluated in driver scope and accumulated at read time (B8, A8) over the driver records the run wrote: not filtered and not rejected. That makes an assertion a check on the mapping, while A8’s counts account for filtered and dead-lettered records separately.
• A column or path an assertion references that doesn’t exist fails at plan time (B8).
• expect: is meant for full loads into empty targets. Target-side aggregates count every row in the table, so under mode: upsert into a table that already has rows they include rows this run didn’t write. The user docs say so. Scoping assertions to the keys a run wrote is deferred.

5.10 The flow vocabulary is closed

from, sink, to, columns, key, mode, where, parent, unnest, nest, order_by, join, keys, cardinality, strategy, expect, depends_on and runtime are the whole flow vocabulary. A target or column named after one of them is fine, because they appear only as keys at fixed places in the grammar.

6. Per-flow overrides of runtime settings

Rule: a flow may override a runtime setting only if the setting’s effect is limited to that flow’s own work. Settings that size or share a process-wide resource can’t be overridden per flow.

Overrides use the same key names under the flow’s runtime:. Precedence is the flow value, then runtime.yaml, then the built-in default. There are no per-file defaults. Setting a non-overridable key on a flow is an error that names the key and gives the reason above.

v0.1 limit, v0.2 item (S3, 2026-10-17). The v0.1 runner has run-wide on_error and max_errors only, so ferry plan refuses a flow whose on_error or max_errors differs from runtime.yaml’s (exit 4, naming the key), rather than silently applying the run-wide value. A flow value equal to the run-wide one is accepted. Per-flow budgets (a flow’s own error count, checked alongside the run-wide one, and per-flow fail vs dead_letter) are planned for v0.2, together with the runner support they need. The grammar above doesn’t change, so configs written for v0.2 keep their meaning; partition_rows is honoured per flow already.

7. depends_on: for ordering flows

In v0.1, flows are ordered across files only by an explicit depends_on: list per flow. FK discovery, which needs a ferry-core change, is deferred until after S2.

depends_on: [<flow>, ...] makes a flow start only after each named flow has finished all of its partitions. Every name must exist, a name may appear only once in the list, a flow may not name itself, and a cycle is an error that names the cycle. All of these are checked at load time, so ferry validate catches them without connecting (B1d). Flows with no path between them run concurrently. Within a flow, order comes from parent: (§5.3).

Why this over FK discovery (A1, B7):

• For depends_on:. It needs no ferry-core change, so S2 is unblocked now. It is visible in the config and in the hash. It works for Mongo sources, which have no foreign keys to discover. And it can express dependencies that exist only in the target, created by the mapping, which source FK discovery could never find.
• Against. It is manual and can drift from the real schema, and a missing edge shows up only as an FK violation, or as a slower run. It is also a flow-level barrier, which costs parallelism compared with finer ordering.
• Why the drift is tolerable in v0.1. A6 drops or defers FKs during the load and validates them at finalize (“write order is an optimization, not the integrity mechanism”). A missing depends_on: therefore produces a finalize report, never silent corruption.
• After S2. Once Source can return FKs, discovered edges are added to the explicit ones (union), and ferry plan labels each edge declared or discovered. depends_on: stays the way to state edges discovery can’t see.

Cross-flow foreign keys in the target DDL are not part of v0.1. TargetSchema FKs stay within one flow.

8. include:

include: lives only in ferry.yaml. It is a list of paths or globs (*, **), resolved relative to the ferry/ directory and allowed to point outside it (../shared/connections.yaml).

• An included file has the same form as a well-known file: it contains flows:, or connections:, or both. runtime: may only come from runtime.yaml, so that one file holds every runtime setting.
• Entries from included files merge with the conventional ones under the same rules: duplicate flow or connection names are errors naming both files.
• A literal path that doesn’t exist, or a glob that matches nothing, is an error. That catches typos.
• A file reached twice (by convention and by include:, or by two overlapping globs) is loaded once. Files are identified by canonical path.
• An include: key anywhere except ferry.yaml is an error. With no nesting there can be no cycles. If nested includes are ever allowed, they will need cycle detection.

9. ${ENV}

${NAME} is substituted in string values only in connections.yaml entries (including included connections:) and in runtime.state. A ${ anywhere in a flow is a config error naming the file and line. NAME matches [A-Za-z_][A-Za-z0-9_]* and is read from the environment. There is no other form in v0.1: ${file:/run/secrets/pg} and the like are errors, and secrets mounted as files are exported to the environment by the caller. A missing variable is an error that names it without printing any other value, and substituted values never appear in errors or in ferry plan (B1).

Keeping ${ENV} out of flows follows from §10. The hash is computed before substitution, so an environment-driven where: created_at < '${CUTOFF}' could change between a run and its resume without changing the hash, and a resumed run would then write different rows than the attempt it is resuming. Parameterized flows are deferred.

10. The config hash

The run pins this hash (A4), and resume refuses to continue if it changes.

1. It is computed on the parsed, merged model, after defaults are filled in and before ${ENV} substitution. url: ${PG_URL} is hashed as that literal text. Rotating a password doesn’t block resume, and secret values never reach the hash.
2. File layout doesn’t count. Moving a flow to another file, renaming a flow file, editing comments or whitespace, or changing include: leaves the hash unchanged as long as the model is the same.
3. Writing a default value out explicitly (mode: insert) doesn’t change the hash, because defaults are filled in before hashing.
4. Canonical form: connections sorted by name, flows sorted by name, targets under to: sorted by name (write order is derived from parent:, not from key order). Everything whose order means something stays as written and is serialized as a list, not an object: columns (DDL column order), from entries (the driver comes first), named where predicates (which one is reported first) and expect. The hash is SHA-256 over canonical JSON with sorted object keys, written sha256:<hex>. depends_on has no meaningful order, so it is sorted before hashing.
5. Settings that change only how fast or how much at a time, not what gets written, are left out of the hash: workers, memory_budget, partition_rows, source_max_connections and source_rows_per_sec, both in runtime.yaml and in a flow’s runtime:. Partitions are stored at plan time, so changing partition_rows has no effect on a resumed run anyway. This lets you resume a run that ran out of memory with a bigger memory_budget. Behavioural settings are hashed: on_error and max_errors (in runtime.yaml and per flow), and state, which decides whether a flow gets TxnWithState (§4).

The engine version is pinned separately (A4).

11. Vocabulary: what changes in the tickets

12. Interface with 0004

0003 owns name and type in each connection entry, the reserved keys and the reference syntax (§3), and the hash and ${ENV} rules (§9, §10). 0004 owns the other fields of each connector type, how they are validated and described in the JSON Schema, and how runtime.state.connection is resolved and shares the sink’s transaction (0004 §8). What 0003 needs from 0004:

• Which fields are secret, so ferry plan and validate can warn when a secret is written literally instead of through ${ENV}. Answered: fields of type Secret, marked x-ferry-secret (0004 §4).
• Where a Postgres sink’s default schema is set, since target keys under to: carry no schema (§3). Answered: the connection’s schema: field, default public (0004 §2).
• Whether per-connection throttles replace or add to the run-wide source_max_connections and source_rows_per_sec (§6). Still open.

Consequences

• B1 can freeze serde structs and the JSON Schema from §1. The bad-config set follows from the error rules in §2, §8 and §9.
• B4, B5, B6 and B8 are edited to the wording in §11 before work starts. The golden config at examples/golden/ferry/ and the merge example at examples/pg-merge/ferry/ are written in this grammar and are what B1, B7, S2 and S3 test against.
• Every flow has exactly one sink, which rules out fan-out to two databases in one flow. Two flows reading the same source do it instead.
• Upsert never deletes target rows for records that are later filtered out. Anyone who needs that needs a sync tool, not v0.1.
• ${ENV} can’t parameterize flows. A cutoff date or tenant id has to be written into the flow, which also puts it in the hash.
• B5 (unnest) is part of S2, not S3: the first end-to-end run produces both orders and order_items from the golden data, so unnest:, $parent and $index (§5.4, §5.7) have to work by S2. nest: (B6) stays in S3.
• Exactly-once (TxnWithState) needs the state store on the flow’s sink connection. A run whose state store is elsewhere still re-runs partitions safely, through the sink’s other strategy (§4, 0004 §8).
• A connection’s real location can change behind an unchanged url: ${PG_URL} without changing the hash. A4 should also record the identity of the connected server (0004 decides what that is) if pointing a resume at a different database has to be refused.
• golden/expected/filtered.json gives the reason as prose (“deleted_at is set”). Under §5.6 the reason is the predicate’s name or expression text, so the fixture and its README need to change together (Track B owns them).

Amendment 2026-10-17: lossy conversions and expression names (B2, B3)

Opt-in syntax (decided). A lossy conversion is allowed per column, by allow_lossy: true on a typed column object:

total_cents: {expr: "int(round(total * 100))", type: int32, allow_lossy: true}

• allow_lossy needs type: in the same object; on its own it is a model error (the JSON Schema says so with dependentRequired). There is no flow-wide or global switch: each lossy column says so where it is written, and the opt-in is part of the config hash.
• Only the conversion from the expression’s inferred type to type: is checked. Conversions written in the expression (int(x), CAST) are explicit already.
• Conversions fall in three classes. Safe (every value converts exactly: integer widening, decimal(10,2) → decimal(12,4), anything → utf8) needs nothing. Lossy needs allow_lossy: true; values that change are changed as the table says, values that can’t convert are error rows. Impossible (bool → timestamp) is an error even with the opt-in.

The canonical lossy table (ferry_dsl::typecheck::cast::LOSSY, each row tested both ways):

Type names for type: are int8 … int64, uint8 … uint64, float32, float64, decimal(p,s), bool, utf8, binary, date, timestamp, timestamp(<zone>), uuid, json and list(<type>).

Expression names (B2, confirms §5.4). The sigils are $parent, $index, $item and $key. Their bare words parent, index, item and key are reserved at the start of a path, so line_no: index is a load error that suggests $index; a field with one of those names is written quoted ("index"). Names are case-sensitive. x == null (and SQL’s x = null) is true for null and missing; missing(x) is true only for missing.

Open questions

Decided 2026-10-16:

• §7: flows are ordered only by explicit depends_on: in v0.1. The ferry-core FK-discovery change is deferred until after S2.
• §10.5: the hash leaves out runtime tuning settings (workers, memory_budget, partition_rows, the two throttles). on_error, max_errors and state stay in it.
• §9: no ${ENV} in flows, only in connections and runtime.state. Only ${NAME} from the environment; no ${file:...}.
• §5.3: each flow names its sink once with sink: <conn>, and keys under to: are plain table names.
• §5.9: expect: is documented as meant for full loads.
• §3: name, type and description are reserved connection keys.
• §4: TxnWithState goes through the registry’s state-capable extension (0004 §8). A flow whose sink isn’t the state connection falls back to the sink’s next strategy.
• B5 (unnest) moves into S2.

Still open:

1. §5.4: a child sees the driver record only through $parent. A $record sigil may be needed later.
2. §9: how to write a literal ${ in a connection value, and whether a variable set to the empty string counts as missing. B1c needs both.

0004: Connector registry, connection options and the CLI layer

Status: proposed. The points that were open were decided on 2026-10-16 (see Open questions). Ticket: [D4] (drafted, gap analysis §3 and §6). Needs review from both tracks: Track A owns the connector crates and the state store that implement it, and Track B owns the loader that feeds it.

Context

Already decided, and not reopened here:

• The config is a ferry/ directory. connections.yaml holds connections:, and each entry is {name, type, ...type-specific fields}. 0003 owns name, type and everything outside connections. This record owns the type-specific fields.
• connections[].type maps to a constructor through a ConnectorRegistry. Built-in postgres and mongo register exactly the way a third-party connector would.
• ferry-cli becomes lib+bin. No unsafe, and no dynamic plugin loading in v0.1.

What exists today:

• Source, Sink and SinkTxn in ferry-core are object-safe (async_trait, Send + Sync). The runner can hold Arc<dyn Source> and Arc<dyn Sink> without knowing the concrete connector.
• PostgresSource::new(url, schema) -> Result<Self> is synchronous and does no I/O. It parses the connection string, and the first trait call connects. TLS policy is read from the parsed config (sslmode inside the URL).
• Connector failures already have a home: FerryError::Connector { connector, message }.
• ferry-core is frozen. Its dependencies are Arrow, serde, futures and thiserror, with no JSON or schema crates.
• Nobody owns the CLI layer: argument parsing, the config path, exit codes, error rendering, or ferry validate. B7 owns the plan handler, S2 wires run, and A5 owns resume/retry entry points (“wiring can be thin”).

Decision

1. The registry lives in a new crate, ferry-registry

ferry-core  ←  ferry-registry  ←  ferry-connector-postgres, ferry-connector-mongo, third-party connectors
     ↑               ↑                          ↑
 ferry-dsl      ferry-state                     │
     ↑               ↑                          │
     └──────── ferry-cli (lib + bin) ───────────┘

ferry-registry holds the connector author’s API: the Connector trait, the ConnectorRegistry, Secret, Roles, the schema composition, the state-capable extension (§8), and a testkit contract suite. It depends on ferry-core, serde, serde_json, schemars and async-trait. ferry-state depends on it for the StateStore and StateBackend traits (§8), which is the edge drawn above.

Rejected:

• ferry-core. Needs a cross-track PR against a frozen crate, and it would push schemars and serde_json into the crate every other crate depends on. The registry deals with configuration and wiring, not the data contracts ferry-core exists for. Nothing in it needs to change a core type.
• ferry-cli. The CLI depends on the built-in connectors to register them. If the trait lived in the CLI, ferry-connector-postgres would depend on ferry-cli: a cycle. The only way around the cycle is to write the built-in impls inside the CLI, which breaks “built-ins register the way a third party would”. It would also make every third-party connector crate depend on clap and the whole engine.
• ferry-dsl. Connector crates would then depend on the planner and DataFusion.

ferry-dsl does not depend on ferry-registry, and the reverse is also true. The loader (B1) parses each connection as a raw entry (name, type, the remaining fields as a serde_json::Value, plus source spans). The CLI passes those entries to the registry and hands the Source/Sink it gets back to the planner. JSON is the interchange format because JSON Schema already speaks it, a future subprocess protocol would too (§7), and YAML-only constructs (tags, non-string keys) are rejected by the loader anyway.

Ownership: ferry-registry is shared, like ferry-cli. Once merged, changes need review from both tracks, because every connector depends on it. Each connector’s impl Connector lives in that connector’s own crate (Track A), exactly where a third party’s would.

2. Registration API

Connector authors implement one typed trait:

#![allow(unused)]
fn main() {
// ferry-registry

pub trait Connector: Send + Sync + 'static {
    /// The `type:` value in `connections.yaml`, e.g. "postgres".
    const TYPE: &'static str;

    /// What this connector can be used as. Checked by `ferry validate`
    /// without building anything: a connection named in `from:` must be a
    /// source, one named in `sink:` must be a sink.
    const ROLES: Roles;

    /// The type-specific fields: the connection entry minus `name` and
    /// `type`. Must use `#[serde(deny_unknown_fields)]`, and must not use a
    /// reserved key (`name`, `type`, `description`, 0003 §3); the contract
    /// suite checks both. Doc comments become editor hover text.
    type Options: DeserializeOwned + JsonSchema;

    /// Builds the endpoints `wants` asks for (always a subset of `ROLES`).
    /// Must not do I/O: `ferry validate` calls it to catch malformed
    /// options, and connecting happens on the first trait call.
    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: Self::Options,
        wants: Roles,
    ) -> Result<Endpoints, BuildError>;

    /// One line for `ferry plan` and the run report, with no secrets,
    /// e.g. `postgres://app@db.example.com:5432/shop, schema public`.
    fn describe(&self, options: &Self::Options) -> String;

    /// Secret values that are not whole `Secret` fields, such as a password
    /// embedded in a URL. They are added to the redaction set (§4).
    fn derived_secrets(&self, options: &Self::Options) -> Vec<String> {
        Vec::new()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Roles {
    pub source: bool,
    pub sink: bool,
}

impl Roles {
    /// Only for the state connection when no flow reads or writes it (§8).
    pub const NONE: Roles = Roles { source: false, sink: false };
    pub const SOURCE: Roles = Roles { source: true, sink: false };
    pub const SINK: Roles = Roles { source: false, sink: true };
    pub const BOTH: Roles = Roles { source: true, sink: true };
}

/// What `build` returns. The registry checks it matches `wants`, and treats
/// a mismatch as a connector bug, not a user error.
#[derive(Default)]
pub struct Endpoints {
    pub source: Option<Arc<dyn Source>>,
    pub sink: Option<Arc<dyn Sink>>,
}

#[non_exhaustive]
pub struct BuildContext<'a> {
    /// The connection's `name:`, for error messages.
    pub connection: &'a str,
    // Room for runtime hints later (e.g. a pool size derived from A5's
    // `source_max_connections`) without breaking connectors.
}

pub enum BuildError {
    /// Bad options, each with a path relative to the connection entry
    /// (e.g. `url`, `tls.ca_file`). The CLI maps the path to file:line.
    Options(Vec<OptionsError>),
    Other(FerryError),
}

pub struct OptionsError {
    pub path: String,
    pub message: String,
}
}

How one type provides a source, a sink or both: ROLES declares it, and build is called once per connection with the roles the config actually uses. A Postgres connection that is both read from and written to gets one build(.., Roles::BOTH), so the connector may share a client or pool between its two halves if it wants to. A connection used only in from: never builds a sink. Mongo declares Roles::SOURCE until the Mongo sink decision (S3) says otherwise, and Postgres declares BOTH once A3 lands. Until then, sink: <pg connection> fails validation with “the postgres connector in this build has no sink”, which is the same message a third-party read-only connector would produce.

The registry stores an object-safe, type-erased form of the trait, so connectors with different Options types can live in one map:

#![allow(unused)]
fn main() {
pub trait DynConnector: Send + Sync {
    fn type_name(&self) -> &str;
    fn roles(&self) -> Roles;
    fn options_schema(&self, generator: &mut SchemaGenerator) -> Schema;
    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: serde_json::Value,
        wants: Roles,
    ) -> Result<Built, BuildError>;

    /// §8. Only connectors registered with `register_state_capable`
    /// override these.
    fn state_capable(&self) -> bool {
        false
    }
    fn build_with_state(
        &self,
        ctx: &BuildContext<'_>,
        options: serde_json::Value,
        wants: Roles,
    ) -> Result<(Built, Arc<dyn StateStore>), BuildError> {
        Err(not_state_capable(self.type_name()))
    }
}

/// Endpoints plus everything the CLI needs to display and redact them.
pub struct Built {
    pub endpoints: Endpoints,
    pub description: String,
    pub secrets: Vec<String>,
}

pub struct ConnectorRegistry { /* BTreeMap<String, Box<dyn DynConnector>> */ }

impl ConnectorRegistry {
    pub fn new() -> Self;
    /// Panics if `C::TYPE` is already registered, or if `C::Options`'s
    /// schema has a property named after a reserved key (0003 §3): both are
    /// wiring bugs in the binary, found the first time it runs.
    pub fn register<C: Connector>(&mut self, connector: C) -> &mut Self;
    /// Replaces a registered type on purpose, e.g. a patched postgres.
    pub fn replace<C: Connector>(&mut self, connector: C) -> &mut Self;
    /// For connectors that aren't a Rust `Options` type (see §7). Same
    /// panics as `register`.
    pub fn register_dyn(&mut self, connector: Box<dyn DynConnector>) -> &mut Self;
    /// Registers a connector that can host the state store, paired with
    /// its backend (§8). Same panics as `register`.
    pub fn register_state_capable<C, B>(&mut self, connector: C, backend: B) -> &mut Self
    where
        C: StateCapable,
        B: StateBackend<Pool = C::Pool, Conn = C::Conn>;

    pub fn types(&self) -> impl Iterator<Item = (&str, Roles)>;
    pub fn build(&self, conn: &RawConnection, wants: Roles) -> Result<Built, BuildError>;
    /// The state connection: `wants` plus the state store, on one pool (§8).
    pub fn build_with_state(
        &self,
        conn: &RawConnection,
        wants: Roles,
    ) -> Result<(Built, Arc<dyn StateStore>), BuildError>;
    /// The `connections[]` item schema for every registered type (§3).
    pub fn connections_schema(&self, generator: &mut SchemaGenerator) -> Schema;
}
}

register wraps C in an internal Typed<C>, which implements DynConnector. Deserialization goes through serde_path_to_error, so a wrong type or unknown field comes back as an OptionsError with a path instead of a bare serde message.

RawConnection (name, type, fields as serde_json::Value) is the ferry-registry view of a connection entry. The CLI converts B1’s parsed entry into it and keeps the span map, so it can turn connections[1].schema back into ferry/connections.yaml:14:5.

Built-in options, as an example of the shape this record owns:

connections:
  - name: legacy
    type: postgres
    url: ${LEGACY_PG_URL}      # libpq URL or key=value string; sslmode goes here
    password: ${LEGACY_PG_PW}  # optional; overrides a password in `url`
    schema: public             # default: public. Read from as a source, written to as a sink.

  - name: shop
    type: mongo
    url: ${SHOP_MONGO_URL}     # mongodb:// or mongodb+srv://
    database: shop

• Both types use url, not uri for Mongo: one word across types is easier to remember than each ecosystem’s own.
• sslmode stays inside the Postgres URL, where Track A’s TLS policy already reads it. It is not duplicated as a separate field.
• Batch size is a runtime concern (A5’s memory budget), not a connection field.
• A2 may add fields, such as a schema-inference sample size, as fields on MongoOptions. That needs no registry change.
• Because build must not do I/O, mongodb+srv:// DNS resolution happens on first connect, not in build.

3. Type-specific fields: per-connector schema fragments, checked by serde

Each connector’s Options derives JsonSchema. connections_schema() composes one complete object schema per registered type and puts them in a oneOf under connections.items:

{ "oneOf": [
  { "type": "object",
    "properties": {
      "name": { "type": "string" },
      "type": { "const": "postgres" },
      "url":  { "type": "string", "writeOnly": true, "x-ferry-secret": true,
                "description": "libpq URL or key=value string. ..." },
      "password": { "...": "..." },
      "schema": { "type": "string", "default": "public" } },
    "required": ["name", "type", "url"],
    "additionalProperties": false },
  { "...": "mongo, then each third-party type" }
] }

Every branch repeats name and type, because additionalProperties: false doesn’t see properties declared in sibling subschemas. B1 generates the rest of the schema with connections.items left open, and the registry fills it in. The JSON Schema therefore depends on which connectors a binary has, so it is generated by the binary (ferry schema, §6), not shipped as a static file. The schema published for the stock ferry is just that command’s output.

${ENV} vs typed fields: ${VAR} is substituted before deserialization, so port: ${PG_PORT} is valid config, but the plain derived schema would flag it in the editor as “expected integer”. The composition step therefore widens every non-string scalar leaf in connection fragments to anyOf: [<original>, {type: string, pattern: "^\\$\\{[A-Za-z_][A-Za-z0-9_]*\\}$"}].

Validation itself never goes through JSON Schema. After ${ENV} substitution, the entry deserializes into the connector’s Options and then build runs. The Rust type is the single source of truth, and the schema is derived from it, so the two can’t drift.

Editors (VS Code YAML / yaml-language-server, JetBrains) with this schema can autocomplete:

• type: values (from the const in each branch)
• each type’s field names, with doc-comment descriptions, defaults and enum values
• required fields and unknown fields (e.g. a typo like shcema:), flagged as you type

Not covered by the schema:

• connection names referenced in from:, sink: and runtime.state.connection. The schema is static and can’t see names defined elsewhere in the config.
• entity names, which need a live database.

ferry validate catches the first, and ferry plan the second.

Rejected: a free-form options map checked when the connector is built. Simpler for connector authors, but it gives no autocomplete, catches typos only at run time, and lets every connector invent its own error format. The cost of fragments is one #[derive(Deserialize, JsonSchema)] per connector. A connector that really is free-form can declare Options = serde_json::Map<String, Value> and opt out.

4. Secrets: Secret, schema markers and one redaction pass

• ferry-registry provides Secret, a string newtype. Debug and Display print ***, the only way to read the value is expose(), and its JsonSchema is {type: string, writeOnly: true, x-ferry-secret: true}. Connectors must use it for every credential-bearing field. Postgres url and password, and Mongo url, are all Secret.
• ${ENV} is the way to supply secrets. Substitution happens before deserialization (B1), so a connector never knows whether a value came from the environment.
• Redaction set. On build, the registry returns the value of every x-ferry-secret field (whatever its origin, literal or ${ENV}) plus derived_secrets() (e.g. the password parsed out of a Postgres URL). The CLI’s renderer replaces each occurrence with *** in everything it prints: diagnostics, plan output, logs and run reports. This is a backstop for driver errors that echo a connection string. Secret values substituted into fields not marked secret are not redacted. That is documented, and it is why connectors must mark their fields.
• ferry plan and the run report show connections only through describe(). The state store records each connection’s name and type, never its fields.
• ferry validate warns (it doesn’t fail) when a secret field holds a literal instead of ${ENV}, because config directories get committed. Local runs against testcontainers legitimately use literals.
• Config hash (A4): the registry only ever sees substituted values, so it can’t compute the hash. The loader hashes the parsed model before substitution, by the rules in 0003 §10 (which also leave runtime tuning settings out). Rotating a password then doesn’t block resume, and secret values never reach the hash.
• ${NAME} from the environment is the only substitution form (0003 §9). Secrets mounted as files are exported to the environment by whatever starts ferry.

5. ferry-cli as lib + bin

The library exposes one builder and an entry point. That is the whole stable surface in v0.1:

// ferry-cli (lib)

pub use ferry_registry as registry;

pub struct Ferry { /* registry, binary name, version */ }

impl Ferry {
    /// No connectors registered.
    pub fn new() -> Self;
    /// Built-in postgres and mongo, each behind a default-on cargo feature
    /// (`postgres`, `mongo`), so a slim binary can drop the Mongo driver.
    pub fn with_builtins() -> Self;
    pub fn register<C: registry::Connector>(self, connector: C) -> Self;
    /// See §8. `with_builtins` pairs postgres with ferry-state's backend.
    pub fn register_state_capable<C, B>(self, connector: C, backend: B) -> Self
    where
        C: registry::StateCapable,
        B: registry::StateBackend<Pool = C::Pool, Conn = C::Conn>;
    /// Name used in help text and error messages; defaults to "ferry".
    pub fn name(self, name: &'static str) -> Self;
    /// Parses `std::env::args`, builds the Tokio runtime, runs the command
    /// and renders errors. Never panics on user error.
    pub fn main(self) -> ExitCode;
    /// Same, with explicit arguments and output, for tests.
    pub fn run_with(
        self,
        args: impl IntoIterator<Item = OsString>,
        out: &mut dyn Write,
        err: &mut dyn Write,
    ) -> ExitCode;
}

The ferry binary is fn main() -> ExitCode { Ferry::with_builtins().main() }. A third-party binary that adds a connector:

// acme-ferry/src/main.rs
use std::process::ExitCode;

use acme_ferry_clickhouse::ClickHouse;
use ferry_cli::Ferry;

fn main() -> ExitCode {
    Ferry::with_builtins()
        .register(ClickHouse)
        .name("acme-ferry")
        .main()
}

#![allow(unused)]
fn main() {
// acme-ferry-clickhouse/src/lib.rs (depends on ferry-core and ferry-registry only)
use std::sync::Arc;

use ferry_registry::{BuildContext, BuildError, Connector, Endpoints, Roles, Secret};

pub struct ClickHouse;

/// A ClickHouse database.
#[derive(serde::Deserialize, schemars::JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ClickHouseOptions {
    /// HTTP(S) endpoint, e.g. https://ch.example.com:8443
    url: String,
    user: String,
    password: Secret,
    database: String,
}

impl Connector for ClickHouse {
    const TYPE: &'static str = "clickhouse";
    const ROLES: Roles = Roles::SINK;
    type Options = ClickHouseOptions;

    fn build(&self, ctx: &BuildContext<'_>, o: ClickHouseOptions, _: Roles)
        -> Result<Endpoints, BuildError>
    {
        let sink = ClickHouseSink::new(&o.url, &o.user, o.password, &o.database)
            .map_err(BuildError::Other)?;
        Ok(Endpoints { sink: Some(Arc::new(sink)), ..Default::default() })
    }

    fn describe(&self, o: &ClickHouseOptions) -> String {
        format!("{} as {}, database {}", o.url, o.user, o.database)
    }
}
}

The third-party binary gets validate, plan, run, resume, retry and schema unchanged. Its schema output includes clickhouse, so its users get autocomplete for it.

Rust-level command functions (validate(&Project), plan(..), …) stay pub(crate) in v0.1. The subcommands are the reuse surface. Exposing the functions would freeze planner and runner types before S3 has run them on real data.

6. The CLI layer

Global flags: -c, --config <DIR> (default ./ferry, also read from FERRY_CONFIG), --color auto|always|never (honours NO_COLOR), -v/-q. If --config points at a file, the error suggests its parent directory. Command output (plan text, run id, summaries) goes to stdout. Diagnostics and logs go to stderr, so ferry plan > plan.txt captures only the plan.

The skeleton ticket ships every subcommand. Until its owner lands, a subcommand exits 1 with “not implemented yet (A5)” or similar, so S2 replaces a stub instead of adding a command.

Exit codes are part of the CLI’s contract, for scripts and CI:

Error rendering: every error the CLI prints is a diagnostic with

• a severity
• a stable code (e.g. F0301), so docs and CI can refer to it
• a message
• the original file and line after includes and merging (B1’s spans)
• the field path
• an optional help line

error[F0301]: connections[1] (legacy): unknown field `shcema`
  --> ferry/connections.yaml:14:5
   |
14 |     shcema: public
   |     ^^^^^^ did you mean `schema`?

--format json prints the same diagnostics as JSON lines for CI and editor integrations. The redaction pass (§4) runs last, on the final rendered text of every format. miette fits this design (named sources, labels, help, JSON); the ticket should confirm it or pick an equivalent.

7. Keeping a subprocess connector protocol possible, without building it

v0.1 registers connectors only at compile time. These choices keep a later language-agnostic protocol (a child process that speaks JSON for control and Arrow IPC for batches over stdio) an additive change:

• DynConnector is the real boundary, and it is data-only: options arrive as serde_json::Value, the schema is JSON Schema, and roles are two booleans. A SubprocessConnector would implement it by running <cmd> spec for its schema and returning proxy Source/Sink implementations. register_dyn and type_name(&self) -> &str (not &'static) exist so such a type can be registered from data, not only from a Rust const.
• build does no I/O and returns trait objects, so a proxy that starts its process lazily fits the same contract.
• Source/Sink already exchange Arrow RecordBatches and serializable Partitions, which cross a process boundary as IPC without loss.
• BuildContext is #[non_exhaustive] and carries only data, never handles. Nothing lets a connector reach into the runner. The one state store handle a connector ever sees is the Ledger a state-capable connector is given (§8), and it does one thing: record a partition done.

The protocol itself, how such a connector is declared in config, and what ferry schema does for it are all out of scope. One known limit: a subprocess sink cannot share the state store’s transaction, so it can only offer UpsertByKey or DeletePartitionThenInsert, never TxnWithState. SinkCaps already expresses that, and §8 explains why it stays true.

8. State-capable connectors: one transaction for the sink and the state store

TxnWithState is the only exactly-once strategy: a partition’s rows and its done-record (the checkpoint) commit in one transaction, and the commit is refused if the lease epoch is no longer current (0001). Everything else the state store does (create the run, store the plan, grant and renew leases, mark failures) needs the same database but not the sink’s transaction. The sharing goes through ferry-registry as a typed extension, so the built-in Postgres connector gets it the same way a third-party one would, and no connector crate has a private path into the state store.

Declaring it. A connector that can host the state store implements a second trait, and is registered with a backend that knows how to keep state in its database:

#![allow(unused)]
fn main() {
// ferry-registry::state

/// The state store as the runner sees it: runs, plans, leases, done-records.
/// A4 owns the methods. It lives here because connectors (through
/// `StateBackend`) and the runner both name it. `ferry-state` implements it
/// for the file backend and for each database backend.
#[async_trait]
pub trait StateStore: Send + Sync { /* A4 */ }

/// The one piece of state store work that runs inside a sink's transaction.
#[async_trait]
pub trait Ledger<Conn: Send + 'static>: Send + Sync {
    /// Fails with `FerryError::StaleLease` if a newer lease exists for the
    /// partition, otherwise writes its done-record. Runs on `conn` inside the
    /// sink's open transaction, so the rows and the record commit together
    /// or not at all.
    async fn record_done(
        &self,
        conn: &mut Conn,
        lease: &Lease,
        summary: &WriteSummary,
    ) -> Result<()>;
}

/// A connector whose sink can carry the state store's done-record.
pub trait StateCapable: Connector {
    /// Driver-level handles the backend works with. They are types from the
    /// driver crate, not from this connector, so a backend can name them
    /// without depending on the connector. Postgres: the pool type A1/A3
    /// settle on, and `tokio_postgres::Client`.
    type Pool: Clone + Send + Sync + 'static;
    type Conn: Send + 'static;

    /// The connection's pool. Connects lazily: no I/O, like `build`.
    fn pool(
        &self,
        ctx: &BuildContext<'_>,
        options: &Self::Options,
    ) -> Result<Self::Pool, BuildError>;

    /// `build` on `pool`, with the sink bound to `ledger`. Only a sink built
    /// here lists `TxnWithState` in `capabilities()`. Under that strategy its
    /// commit calls `ledger.record_done` on the transaction's own connection,
    /// after the last write and before COMMIT.
    fn build_with_ledger(
        &self,
        ctx: &BuildContext<'_>,
        options: Self::Options,
        wants: Roles,
        pool: Self::Pool,
        ledger: Arc<dyn Ledger<Self::Conn>>,
    ) -> Result<Endpoints, BuildError>;
}

/// A state store kept in a connector's database (A4, in `ferry-state`).
pub trait StateBackend: Send + Sync + 'static {
    type Pool: Clone + Send + Sync + 'static;
    type Conn: Send + 'static;

    /// No I/O. The store creates its tables on first use.
    fn open(&self, pool: Self::Pool) -> (Arc<dyn StateStore>, Arc<dyn Ledger<Self::Conn>>);
}
}

Ferry::with_builtins() calls register_state_capable(Postgres, ferry_state::PgStateBackend). A third-party database that wants exactly-once ships both halves and makes the same call.

Resolving runtime.state.connection.

1. ferry validate looks up the named connection, checks its type equals backend: and that the registry says the type is state-capable. Any failure is a config error (exit 3) at the runtime.yaml line. Under backend: file there is no state connection, and nothing below happens.
2. The CLI builds every connection once, with the roles its flows use (§2). The state connection goes through build_with_state instead of build, with Roles::NONE if no flow uses it. Inside, the registry deserializes the options once and calls pool, then backend.open(pool.clone()), then build_with_ledger(.., pool, ledger). The source half, the sink half and the state store share that one pool.
3. Each worker’s SinkTxn holds one connection from the pool for the whole partition. Under TxnWithState the runner doesn’t write the done-record itself: the sink’s commit calls ledger.record_done on that connection, then commits. A crash anywhere before COMMIT leaves neither the rows nor the record; a crash after it leaves both. Under every other strategy the runner calls the store after commit returns, and a crash in between re-runs a partition the strategy makes safe to re-run.

When the sink isn’t the state connection. This covers a flow whose sink is a different connection (even one with the same type and the same URL: connections are identified by name, and URLs are secrets the registry doesn’t compare), a sink whose connector isn’t state-capable, and every flow under backend: file. Such a sink was built with plain build, so it never lists TxnWithState, and the planner takes the next strategy SinkCaps::strategy_for returns: DeletePartitionThenInsert, UpsertByKey or StagingSwap, depending on the sink and the mode. Only when that is None is it a plan error (exit 4), as for any flow (0001, B7). ferry plan names the strategy per flow and, for these flows, why: DeletePartitionThenInsert (state store is on "warehouse", not "reporting").

Falling back, rather than refusing the plan, because:

• Every strategy in SinkCaps already makes a re-run safe; that is what 0001 requires of them. The difference is that a partition may be written twice and the second write repairs the first, instead of being written once. The data ends up the same.
• The config never names a strategy (0003 §5.5). Refusing would make the state store’s location decide whether a run can start at all, and a run with flows into two databases (only one of which can hold the state store) could never start.
• It is what strategy_for means: TxnWithState is the first preference, not a requirement. The fallback stays visible in ferry plan, and moving runtime.state.connection to the flow’s sink is a one-line fix.

Type safety, without unsafe. The connector’s handle types and the backend’s must be equal, and the compiler checks it once, in the where clause of register_state_capable. After that call, an internal StateTyped<C, B> implements DynConnector and calls pool, open and build_with_ledger with concrete types. The runner only ever sees Arc<dyn Sink> and Arc<dyn StateStore>. There is no Any, no downcast and no TypeId. If the connector and ferry-state end up on incompatible driver versions, their Pool types differ and the binary fails to compile rather than at run time. The contract suite checks that a sink lists TxnWithState if and only if it was built through build_with_ledger, so a connector can’t claim exactly-once without being bound to the run’s state store.

The subprocess protocol (§7). Pool and Conn are in-process Rust types, and a ledger call runs on a live driver connection, so neither can cross a process boundary. A SubprocessConnector is therefore never state-capable: it registers through register_dyn, keeps the default state_capable() == false, its sinks never list TxnWithState, and its flows fall back as above. Naming it in runtime.state.connection is a config error. DynConnector gains a boolean and one method with a default, so the boundary §7 keeps data-only is unchanged. If a later protocol wants exactly-once, the state backend would have to run inside the subprocess too; that is a new design, not an extension of this one.

Consequences

• No change to frozen ferry-core contracts. One new workspace member, ferry-registry, and the README’s crate table gains a row.

• ferry-state depends on ferry-registry, implements StateStore for the file backend, and provides PgStateBackend behind a postgres feature. The Postgres sink (A3) implements StateCapable; the state store (A4) implements the backend and Ledger. Both build on this record’s crate, so A3 and A4 depend on its ticket.

• A run gets exactly-once only for flows whose sink is the state connection. Other flows re-run partitions safely through their sink’s other strategy.

• Every connector crate (Track A) gains a dependency on ferry-registry and schemars, and a small impl Connector next to its Source/Sink. PostgresSource::new already has the no-I/O shape build requires.

• The JSON Schema is per-binary. Editor setup points at the output of ferry schema (via a yaml-language-server modeline or settings), not at a URL baked into the repo. The published schema is the stock binary’s.

• Connection field names (url, password, schema, database) are now part of the user-facing config. They are owned here, and renaming them later is a breaking change.

• ferry-registry gets a testkit contract suite, check_connector::<C>(), in the same style as 0.5’s source/sink suites. It checks that:

  • the schema generates
  • an unknown field is rejected with a path
  • a known secret never appears in Debug, describe() or build errors
  • build returns exactly the roles asked for
  • the Options schema uses no reserved key
  • for a state-capable connector: its sink lists TxnWithState only when built through build_with_ledger, and a crash between the data write and record_done commits neither

  Built-ins must pass it, and third parties can run the same suite.

• New ticket (replaces gap analysis items 3 and 6):

  Scope:

  • ferry-registry
  • the Connector impls for postgres and mongo
  • the ferry-cli lib/bin split with the Ferry builder
  • the state-capable extension (§8) and register_state_capable
  • argument parsing, --config, exit codes and diagnostic rendering
  • ferry validate and ferry schema
  • stub subcommands for plan, run, resume <run-id> and retry --failed <run-id>

  Done when:

  • validate reports every error in the bad-config set, each with file:line
  • a third-party example binary in tests/ registers a fake connector and validates a config that uses it
  • the generated schema autocompletes type: and per-type fields in VS Code YAML (checked by hand once and recorded)
  • a planted secret never appears in any output
  • a state connection that isn’t state-capable, or whose type differs from backend:, fails validate at the runtime.yaml line

  Blocks S2, A3 and A4. Depends on B1’s loader and spans.

Open questions

Decided 2026-10-16:

• Reserved connection keys: name, type, description, published in 0003 §3. register (and register_dyn, register_state_capable) panics if an Options schema uses one, and the contract suite checks it.
• State store connection: the sink↔state store transaction goes through ferry-registry as the typed state-capable extension (§8), not privately inside the Postgres crates. A sink that isn’t on the state connection falls back to its next strategy per SinkCaps::strategy_for.
• ferry run exits 0 when dead-lettered rows stay under max_errors, with the count in the summary.
• resume and retry require a run id.
• ${ENV} only; no ${file:...} in v0.1.
• url for both Postgres and Mongo.

Still open:

1. Connector versions in the run record. A4 pins runs to the engine version. In a third-party binary, a connector upgrade can change how partitions or keys are encoded. Proposed: record type@version per connection, and have resume warn (not refuse) on a mismatch.
2. --fail-on-dead-letters. Some CI users will want any dead-lettered row to fail the run. Not in v0.1 unless someone asks.
3. Requiring exactly-once. §8 falls back silently apart from the plan output. Should a flag (or a ferry plan warning that CI can fail on) refuse a run where any flow doesn’t get TxnWithState?
4. Per-connection throttles. Whether they replace or add to the run-wide source_max_connections and source_rows_per_sec (0003 §6, §12).

Changelog

Every release’s changes are in CHANGELOG.md in the repository, and each release, with its binaries and checksums, is on the GitHub Releases page.

0.1.0-beta.1 is the first release: MongoDB and PostgreSQL to PostgreSQL, with every shape of flow this guide describes, resumable runs and verification.