How a source change becomes a denormalized document.
A VentStream pipeline has two halves: capture (read the source’s
change stream) and project (turn each change into the documents it
affects). This page explains both and how they meet.
Logical replication via the in-tree pgoutput plugin. You create a
publication listing the tables to stream and the engine consumes a
replication slot. Every insert/update/delete on a published table
arrives as a typed event with a WAL position (LSN) as its cursor.
CREATE PUBLICATION ventstream_shop FOR TABLE shop.orders, shop.customers, shop.order_items;
A replication slot is single-consumer. Exactly one agent reads a
given slot. Pointing two agents at one slot is undefined behavior —
give each agent a unique VS_PG_SLOT.
Neo4j CDC via the db.cdc.query procedures. The engine polls the change feed
on an interval and persists an opaque cursor string. Use a Neo4j Enterprise
release or AuraDB tier that provides db.cdc.*; the
Neo4j source guide covers the current deployment
requirements.
ALTER DATABASE neo4j SET OPTION txLogEnrichment 'DIFF';
DIFF is recommended for projections (lighter transaction log, same result —
the engine re-queries the graph to build each document). FULL is only needed
for a raw CDC tail that ships full per-event state. See
Neo4j source → DIFF vs FULL.
A projection spec is optional. With no joins or denormalize spec at all,
every change flows through as a flat per-row document with a deterministic id
({schema.table}:["pk",…] — see
deterministic document IDs):
updates overwrite the document in place, deletes remove it, and a
primary-key-changing update removes the old document and writes the new one.
Reach for a spec when you want composed documents — a parent with embedded
children — not for correctness.A projection spec (YAML) declares the target document. The engine
uses it three ways:
Bootstrap — derive the initial scan that seeds every existing
primary into the sink.
Fan-out — on each change, find the affected primaries and
recompute their documents.
Delete — when a primary disappears, emit a sink delete for its
document.
The Postgres spec embeds related rows into the primary’s document:
joins: - name: orders primary: table: shop.orders pk: order_id related: - id: customer table: shop.customers pk: customer_id join_on: { from: customer_id, to: customer_id } embed_as: customer # one customer object on each order cardinality: one - id: items table: shop.order_items pk: item_id join_on: { from: order_id, to: order_id } embed_as: items # array of line items on each order cardinality: many
A change to orders, customers, or order_items recomputes the
affected order documents. A cardinality: one related row embeds as an
object; many embeds as an array.
The Neo4j spec is Cypher. You write the body that, given a primary node
p, returns the document; the engine wraps it with the fan-out anchor:
denormalize: - primary_label: Product output_table: products_denormalized fan_out_max_hops: 2 cypher: | OPTIONAL MATCH (p)-[:IN_CATEGORY]->(cat:Category) OPTIONAL MATCH (p)-[:SUPPLIED_BY]->(sup:Supplier)-[:LOCATED_IN]->(reg:Region) RETURN elementId(p) AS primaryEid, { elementId: elementId(p), name: p.name, category: cat.name, region: reg.name } AS doc
fan_out_max_hops bounds how far a change can be from a Product and
still recompute its document. A change to a node 3 hops away when the
cap is 2 simply isn’t reached.
The join engine sits between the source and the sink. On each event:For Postgres the “which primaries” step uses an in-memory reverse index
of foreign keys. For Neo4j it’s a Cypher query anchored on the changed
element’s ID. Either way, only affected primaries are recomputed —
see Fan-out for how that stays bounded even when a
change touches a shared lookup node.
An event the join engine can never process — a payload or row that isn’t a
JSON object, a subject that isn’t CDC-shaped, a row with no usable key — is
written to the dead-letter file (VS_DLQ_PATH) with the reason prefixed
join engine:, fsynced, logged as metric=join.poison, and skipped; the
cursor advances past it so a restart never replays it. Only
failures that are a property of the event itself are handled this way. A
failure of the environment — the related-row fetcher cannot reach the
source, state cannot be persisted — stays fatal and is retried by the
supervisor, because the same row would compose once the source is back and
dead-lettering it would silently drop data.
A change propagates to the sink only if it lies on a path the spec
actually declares, within the hop limit, and (for temporally-gated
Neo4j edges) satisfies the spec’s WHERE. A node whose label the spec
never traverses produces zero recomputations — the event is read and
discarded.This is the contract: the spec is the boundary. Nothing outside it can
cause a write, which is exactly what keeps the index clean and the
fan-out bounded. (In a flat pipeline with no spec, the source scope — the
Postgres publication, the collection or table list, the topic set — is the
boundary instead.)
