Data Movement Catalog

The Data Movement Catalog is faucet’s first-party, persistent record of everything your pipelines touch. Where a run’s logs and metrics describe one run, the catalog accumulates across runs:

• Datasets — every source and sink a pipeline has read or written, keyed by a canonical, credential-redacted dataset URI.
• Schema timelines — the observed record schema of each dataset, stored as a deduplicated timeline: a new version is appended only when the schema actually changes, together with a computed diff (added / widened / incompatible / removed columns).
• Volume & freshness — per-run record counts and the last-success timestamp for each dataset.
• Lineage edges — which dataset feeds which, with per-edge column lineage whenever the transform chain is expressible (the same derivation the OpenLineage emitter uses).
• Provenance — every catalog row is linked to the run that produced it (the serve run id under faucet serve, the invocation run id otherwise).

After a few weeks of runs the catalog answers the operational questions that otherwise require spelunking logs: what’s the schema history of this table?, what feeds it?, when did this pipeline last land data, and how much?

Recording is observational only: a catalog write never fails or slows a run — a broken store logs a warning and the pipeline continues.

Requires a build with the catalog Cargo feature (included in --features full), plus serve-history-sqlite / serve-history-postgres for persistent stores.

Recording from faucet run / schedule / replicate

Add a top-level catalog: block naming the store:

# cli/examples/csv_to_jsonl_with_catalog.yaml
version: 1
name: csv_to_jsonl_with_catalog

catalog:
  url: sqlite:./faucet-catalog.db
  sample_records: 100        # schema-inference sample per side (default 100)

pipeline:
  source: { type: csv,   config: { path: ./data/input.csv } }
  sink:   { type: jsonl, config: { path: ./out/records.jsonl } }

url accepts sqlite:<path>, a postgres://… URL, or memory (process-lifetime only — for tests). Every successful root invocation then folds its observations into the store: dry runs, --limit runs, shard executions, and cancelled runs are excluded so partial or synthetic volumes never pollute the history.

Recording from faucet serve

faucet serve needs no config block: every run is recorded into the server’s --history backend automatically, attributed to its serve run id. Use a persistent history for a persistent catalog:

faucet serve --history sqlite:./faucet-catalog.db --auth-token "$TOKEN"

The run-record retention window does not purge the catalog — the accumulated history is the point. Only per-dataset volume points are capped (newest 500 kept).

Browsing: CLI

faucet catalog datasets --config pipeline.yaml            # list (newest first)
faucet catalog datasets --config pipeline.yaml --kind csv --q users
faucet catalog show 3f2a9c1e0b7d4a55 --config pipeline.yaml
faucet catalog lineage --config pipeline.yaml --root 3f2a9c1e0b7d4a55 --depth 3

show accepts a unique prefix of the dataset id. Every subcommand takes --json for machine-readable output. faucet schema catalog prints the catalog: block’s JSON Schema.

show renders the schema timeline with diff markers:

schema timeline (2 versions):
  v1  2026-07-01T02:00:04Z  2 column(s)  run 0197e6…
  v2  2026-07-06T02:00:03Z  3 column(s)  run 0197f1…  [+email]

Browsing: HTTP API + web console

Three read-only endpoints (viewer-readable under RBAC):

The embedded web console adds a Datasets browser (filterable list → per-dataset detail with the schema timeline and volume bars) and a Lineage graph view (layered SVG; click a node for its detail).

Dataset identity & cardinality

The catalog key is the connector’s dataset URI after two normalizations:

1. Credentials are redacted (postgres://user:***@host/db/table).
2. ${now.*}-derived path segments are folded back to their tokens — a sink writing ./out/dt=${now.date}/part.jsonl catalogues as one dataset (…/dt=${now.date}/part.jsonl), not one per day.

Matrix rows that resolve to the same URI converge on one dataset with one provenance trail per run.

Schema observation

Schemas are inferred from a bounded sample of the records actually read (source side, pre-transform) and written (sink side, post-transform) — the same samplers the lineage emitter uses, capped by sample_records. The timeline dedupes by a content hash, so re-running an unchanged pipeline never grows it; a real change appends one version whose diff is computed with the same engine as schema-drift handling.

Relationship to lineage emission

OpenLineage emission exports run events to an external backend (Marquez, DataHub, …); the catalog is the first-party store faucet keeps for itself. They compose — the catalog’s per-edge column lineage matches the OpenLineage column-lineage facet, and both can be active at once.