# Orchestration Local-first orchestration work lives here. The current unit is an Airflow 3.2 dev stack under `orchestration/airflow/`, created for #21. This is not a production deployment target. It is the local control-plane surface for proving contract-compiled assets, then later wrapping dbt, transform/python, ingestion load, and ingestion extract work as data product assets. ## Layout - `airflow/engine/` - shared orchestration code, including the contract-to-Asset compiler that projects authored contracts into Airflow Asset specs. - `airflow/dags/` - thin DAG registration modules. Business dependencies should come from contracts, not hand-authored DAG logic. - `airflow/sources/` - orchestration-only config such as local schedules, pools, and connection declarations. - `airflow/setup/` - local Docker Compose stack for Airflow 3.2. ## Start Local Airflow ```bash cd orchestration/airflow/setup cp .env.example .env docker compose build docker compose up airflow-init docker compose up -d --wait ``` The API/UI is available at `http://localhost:8080` with the local admin user from `.env` (`airflow` / `airflow` by default). Useful checks: ```bash docker compose run --rm airflow-cli airflow version docker compose ps docker compose down --volumes --remove-orphans ``` ## Hello Asset Smoke The stack includes `orchestration_hello_asset_producer` and `orchestration_hello_asset_consumer`. Trigger `orchestration_hello_asset_producer` in the UI or CLI. When its task emits `x-datalake://orchestration/hello_raw`, Airflow's asset-aware scheduler should create a run of `orchestration_hello_asset_consumer`, which emits `x-datalake://orchestration/hello_consumed`. The hello DAG remains a local smoke test. Product DAGs should be registered from compiled contract assets rather than hand-authored dependencies. ## dbt Revrec Assets `orchestration_dbt_revrec_assets` registers the first dbt/Cosmos slice for #23. It loads `orchestration/airflow/sources/dbt-revrec-manifest.json` with `LoadMode.DBT_MANIFEST`, selects the revrec products from `orchestration/airflow/sources/dbt-revrec.yaml`, and uses contract-compiled `x-datalake://...` Assets for task inlets and outlets. Cosmos calls `orchestration/airflow/bin/dbt-databend-shim` instead of the stock `dbt` executable. The shim runs the dbt transform extra through `uv` in a separate project environment under the persisted Airflow venv volume, then delegates to `transform/dbt/engine/run_dbt.py`. The transform extra pins the tested Databend adapter fork branch that carries TiDB Cloud Lake warehouse and seed support. The local Airflow image needs Cosmos and `uv` installed. Copy the updated `.env.example`; its `_PIP_ADDITIONAL_REQUIREMENTS` value installs the local dev dependencies into the Airflow containers. dbt stays in the transform extra because Airflow 3.2 and dbt Core 1.8 have incompatible dependency pins. Prewarm the shim once before a full DAG trigger when the Docker volumes are cold; after that, dbt tasks reuse the persisted transform venv and `uv` cache. Refresh the manifest fixture after dbt graph changes by running dbt parse and sanitizing the generated manifest into `orchestration/airflow/sources/`. Do not commit `transform/dbt/target/`; it contains run-local dbt artifacts. ## Python Revrec Assets `orchestration_python_revrec_assets` registers the `transform/python` `monthly_cloud_billing_revrec` job for #24. It is scheduled by the two upstream dbt-curated Assets declared in the hand-authored Python product contracts and emits `finance.monthly_revrec_collections` plus `finance.monthly_revrec_orders`. The task attaches row-count metadata to each output asset event in contract order and returns the same summary for task-level inspection. The task calls `orchestration/airflow/bin/python-asset-job-shim`, which runs the job through `uv --extra pyjobs` so pandas, numpy, and the Databend driver stay in the transform/python runtime rather than the Airflow service environment. The compose stack persists the `uv` cache plus shim venvs in Docker volumes, and the Airflow task streams shim output so the first dependency sync is visible in task logs. ## Ingestion Load Assets `ingestion_load_assets.py` registers the first load tasks for #25 from `orchestration/airflow/sources/load-assets.yaml`. `orchestration_load_data_foundation_finance` is the manifest-driven S3 load. It is scheduled by the extract manifest handoff Asset and emits the source's raw-snapshot product Assets from the generated load contracts. At runtime the task resolves the manifest from triggering asset event metadata (`manifest_path` or `manifest_uri`) or from the configured environment variable. `orchestration_load_nexuscrm` is the NexusCRM manifest-driven S3 load. It follows the same extract-manifest handoff shape and emits the NexusCRM raw-snapshot product Assets from the generated load contracts. `orchestration_load_metadb_external_incremental` is the scheduled external-incremental poll. It calls the same shared loader CLI path; the loader lists new object-store increments, self-watermarks against the raw table, and fails the task on schema drift after writing the proposed contract plus an `open_pr` action request for the contract-update PR handoff. All load tasks call `orchestration/airflow/bin/load-runner`, which runs the loader through `uv` in `/tmp/pingcap-datalake-load-venv`. Before execution it syncs the read-only repo mount into `DATALAKE_LOAD_WORKSPACE` so runtime logs and proposed drift contracts are written to a local artifact volume rather than the mounted Git checkout. Schema-drift PR action requests are written under the load source's `runs/action-requests/` there. This keeps the load runtime separate from the dbt transform and Python job environments. The local Airflow image is built from `orchestration/airflow/setup/Dockerfile`; it adds `aws`, `gcloud`, `ticloud`, and `rsync`, which the manifest download, object-store listing, TiDB Cloud export, and writable workspace sync paths need. ## Ingestion Extract Assets `ingestion_extract_assets.py` registers the first extract tasks for #26 from `orchestration/airflow/sources/extract-assets.yaml`. `orchestration_extract_cloud_tenants_bigquery` is the synchronous BigQuery path: `generate-contracts -> export -> write-manifest`. The final task emits a manifest Asset with `manifest_path`, `manifest_uri`, `run_date`, and `run_id` event metadata. `orchestration_extract_data_foundation_finance_tidbcloud` is the asynchronous TiDB Cloud path. It starts the export, defers to the Airflow triggerer while polling `export-status`, then writes and emits the manifest Asset after TiDB Cloud reports success. That manifest Asset is the same `ingestion.extract.tidbcloud.data_foundation_finance.manifest` identity consumed by the manifest-driven load DAG. Extract tasks call `orchestration/airflow/bin/extract-runner`, which runs in `/tmp/pingcap-datalake-extract-venv` and syncs the repo into `DATALAKE_EXTRACT_WORKSPACE` before writing generated contracts and run manifests. ## Secrets Do not commit source secrets here. Local credentials should be provided as Airflow Connections, Variables, or a secret backend entry. The `sources/` config may declare which connection IDs a unit expects, but actual values belong in local `.env` files or the configured secret backend.