# transform/ — the transformation layer

The transform layer turns the raw lakehouse (`*_raw` databases owned by `ingestion/`, ADR 0016)
into staging and curated/service data products. It is a peer of `ingestion/` and holds peer
execution-engine units (ADR 0019):

- **`dbt/`** — the primary engine: dbt on Databend (`dbt-databend-cloud`).
- **`taskflow/`** — a parallel build of the same pipeline as native Databend Task Flow, kept only
  to compare the two engines while we feel things out. See `taskflow/README.md`.
- **`python/`** — standalone pandas jobs for models dbt can't run on this gateway (it has no
  python-model backend). Downstream of dbt's curated products, not a parity build — so it is an
  independent producer and owns a hand-authored contract for each product it writes
  (`python/contracts/`, ADR 0003/0015). See `python/README.md`.

Two pipelines are ported here from the BigQuery + dbt DWH (`digital-bi-dbt`):
`salesforce_org_monthly_revrec` and `salesforce_order_daily_view`, with their upstreams. Salesforce
CRM objects are read from the lakehouse `nexuscrm` instance; the historical Revenue__c-shaped
rev-rec input is read from the one-off `salesforce_raw` snapshot because NexusCRM has no populated
Revenue__c rows and the managed finance-hub mirrors lack the full legacy context; `cloud_tenants`
is snapshotted from BigQuery (`bq_warehouse_raw`). Pruning note (ADR 0019): branches not consumed
by a target are not re-ported — e.g. `salesforce_order_daily_view` needs only currency + exchange
rate from opportunities, so the full `salesforce_opportunities` enrichment is replaced by a
minimal projection; and the BigQuery nested-array + correlated-`UNNEST` accumulation is rewritten
as flat Databend aggregation.

## dbt/ layout
- `engine/` — generic, source-free config: `dbt_project.yml`, `profiles.yml`, `macros/`, `tests/`,
  and the helper scripts (`run_dbt.py`, `load_seeds.py`, `generate_contracts.py`).
- `sources/<source>/` — staging keyed on the raw source read FROM (`nexuscrm`, `metadb`,
  `googlesheets`, `bq_warehouse`): `models/staging/`, `sources.yml`, generated staging contracts,
  and (googlesheets) the seed.
- `products/<product>/` — warehouse models keyed on the product produced, with generated curated
  contracts beside them.
- `seeds/` — seeds with no single source home (e.g. `payment_type_mapping`).

Output databases: `transform_dbt_stg` (staging), `transform_dbt_dwh` (curated), `transform_seeds`
(neutral seed db shared with Task Flow).

## Run it
```bash
# install the transform extra (dbt + databend adapter)
uv sync --extra transform

# all dbt invocations go through the transform-env dbt entrypoint, with env from .env:
DBT="uv run --extra transform --env-file .env -- python transform/dbt/engine/run_dbt.py"

$DBT debug   --project-dir transform/dbt --profiles-dir transform/dbt
$DBT seed    --project-dir transform/dbt --profiles-dir transform/dbt
$DBT run     --project-dir transform/dbt --profiles-dir transform/dbt
$DBT test    --project-dir transform/dbt --profiles-dir transform/dbt   # includes dbt<->Task Flow parity

# regenerate the staging/curated contracts from the built relations:
uv run --extra transform --env-file .env -- python transform/dbt/engine/generate_contracts.py

# native Task Flow parallel build + parity:
$DBT compile --project-dir transform/dbt --profiles-dir transform/dbt
uv run --env-file .env -- python transform/taskflow/deploy.py

# BigQuery parity for the full emitted salesforce_order_daily_view output:
uv run --extra transform --env-file .env -- python transform/dbt/parity/salesforce_order_daily_view.py
```

## Adapter caveats (ADR 0019)
`dbt-databend-cloud` is pinned to the tested fork branch that carries TiDB Cloud Lake gateway
support while the upstream package catches up:
- the adapter accepts the `warehouse` profile field required by the gateway;
- `dbt seed` emits gateway-safe INSERT statements for committed seed CSVs.
- Connection vars (`DATABEND_HOST/PORT/USER/PASSWORD/WAREHOUSE`) are derived from `DATABEND_DSN` in
  `.env`. nexuscrm raw columns are case-sensitive (quoted with exact case in staging).

## Status: feel-things-out
Working end-to-end against live Databend on real `nexuscrm`/`metadb` data; the Google-Sheet rows and
`cloud_tenants` are fixtures (`dbt seed`, `engine/fixtures/cloud_tenants_fixture.sql`)
seeded with real ids. The `ingestion/.../cloud_tenants`, `metadb_tenants`, `metadb_users` units are
the real ingestion paths behind those fixtures (placeholders for buckets/creds).