Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

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.