"""Deploy + run the salesforce_org_monthly_revrec pipeline as native Databend Task Flow tasks. This is the parallel implementation to the dbt build, for comparing the two execution engines (ADR 0019). Each dbt model becomes one CREATE TASK that materializes the same logic into transform_taskflow_{stg,dwh}. The SQL bodies are taken from the dbt-compiled output, so both engines run identical, dialect-verified SQL — the comparison is about the orchestration and materialization experience, not SQL divergence. LIMITATION: this Databend build (TiDB Cloud Lake) does not support the task `AFTER` clause, so a native task-DAG cannot be declared. Instead each model is a standalone scheduled task (effectively never auto-fires; schedule is 1 year and tasks are suspended after running), and this script drives them in topological order, polling system.task_history between steps. The generated, committed artifact is flows/salesforce_org_monthly_revrec.flow.sql. Prereqs: dbt compiled (transform/dbt/target/compiled/...), seeds loaded (transform_seeds), and the cloud_tenants fixture present. Usage: uv run --env-file .env -- python transform/taskflow/deploy.py """ from __future__ import annotations import os import sys import time from pathlib import Path from databend_driver import BlockingDatabendClient REPO = Path(__file__).resolve().parents[2] COMPILED = REPO / "transform" / "dbt" / "target" / "compiled" / "pingcap_lakehouse" FLOW_OUT = REPO / "transform" / "taskflow" / "flows" / "transform.flow.sql" WAREHOUSE = os.environ.get("DATABEND_WAREHOUSE", "default") # (task_name, compiled_path, target_db, materialization) — in dependency (topological) order. MODELS = [ ("revrec__stg_salesforce_org", "sources/nexuscrm/models/staging/stg_salesforce_org.sql", "transform_taskflow_stg", "VIEW"), ("revrec__stg_salesforce_account", "sources/nexuscrm/models/staging/stg_salesforce_account.sql", "transform_taskflow_stg", "VIEW"), ("revrec__stg_metadb_billing2_invoices", "sources/metadb/models/staging/stg_metadb_billing2_invoices.sql", "transform_taskflow_stg", "VIEW"), ("revrec__stg_cloud_tenants", "sources/bq_warehouse/models/staging/stg_cloud_tenants.sql", "transform_taskflow_stg", "VIEW"), ("revrec__stg_data_foundation_finance_salesforce_account", "sources/data_foundation_finance/models/staging/stg_data_foundation_finance_salesforce_account.sql", "transform_taskflow_stg", "VIEW"), ("revrec__stg_google_sheets_org_monthly_bill_confirmation_for_revrec", "sources/googlesheets/models/staging/stg_google_sheets_org_monthly_bill_confirmation_for_revrec.sql", "transform_taskflow_stg", "VIEW"), ("revrec__salesforce_orgs", "products/salesforce_orgs/models/salesforce_orgs.sql", "transform_taskflow_dwh", "VIEW"), ("revrec__salesforce_accounts", "products/salesforce_accounts/models/salesforce_accounts.sql", "transform_taskflow_dwh", "VIEW"), ("revrec__customer_surrogate_keys", "products/customer_surrogate_keys/models/customer_surrogate_keys.sql", "transform_taskflow_dwh", "TABLE"), ("revrec__customers", "products/customers/models/customers.sql", "transform_taskflow_dwh", "VIEW"), ("revrec__salesforce_org_monthly_revrec", "products/salesforce_org_monthly_revrec/models/salesforce_org_monthly_revrec.sql", "transform_taskflow_dwh", "TABLE"), # --- salesforce_order_daily_view chain --- ("odv__stg_dates", "sources/core/models/staging/stg_dates.sql", "transform_taskflow_stg", "TABLE"), ("odv__stg_salesforce_order", "sources/nexuscrm/models/staging/stg_salesforce_order.sql", "transform_taskflow_stg", "VIEW"), ("odv__stg_salesforce_opportunity", "sources/nexuscrm/models/staging/stg_salesforce_opportunity.sql", "transform_taskflow_stg", "VIEW"), ("odv__stg_salesforce_order_item", "sources/nexuscrm/models/staging/stg_salesforce_order_item.sql", "transform_taskflow_stg", "VIEW"), ("odv__stg_salesforce_product", "sources/nexuscrm/models/staging/stg_salesforce_product.sql", "transform_taskflow_stg", "VIEW"), ("odv__stg_salesforce_revenue", "sources/salesforce/models/staging/stg_salesforce_revenue.sql", "transform_taskflow_stg", "VIEW"), ("odv__stg_cloud_revenue", "sources/data_foundation_finance/models/staging/stg_cloud_revenue.sql", "transform_taskflow_stg", "VIEW"), ("odv__date_spine", "products/date_spine/models/date_spine.sql", "transform_taskflow_dwh", "TABLE"), ("odv__salesforce_opportunities", "products/salesforce_opportunities/models/salesforce_opportunities.sql", "transform_taskflow_dwh", "VIEW"), ("odv__cloud_revenue", "products/cloud_revenue/models/cloud_revenue.sql", "transform_taskflow_dwh", "VIEW"), ("odv__salesforce_orders", "products/salesforce_orders/models/salesforce_orders.sql", "transform_taskflow_dwh", "VIEW"), ("odv__salesforce_revrec", "products/salesforce_revrec/models/salesforce_revrec.sql", "transform_taskflow_dwh", "VIEW"), ("odv__salesforce_order_daily_view", "products/salesforce_order_daily_view/models/salesforce_order_daily_view.sql", "transform_taskflow_dwh", "TABLE"), ] def model_name(task: str) -> str: # task names are "__"; strip the pipeline prefix. return task.split("__", 1)[1] if "__" in task else task def body_for(path: str) -> str: sql = (COMPILED / path).read_text().strip().rstrip(";") # Re-point the dbt schemas at the Task Flow schemas; raw dbs and transform_seeds are shared. return sql.replace("transform_dbt_stg.", "transform_taskflow_stg.").replace("transform_dbt_dwh.", "transform_taskflow_dwh.") def task_sql(task: str, path: str, db: str, mat: str) -> str: obj = f"{db}.{model_name(task)}" return ( f"CREATE TASK {task}\n" f" WAREHOUSE = '{WAREHOUSE}'\n" f" SCHEDULE = 525600 MINUTE\n" f"AS\n" f"CREATE OR REPLACE {mat} {obj} AS\n" f"{body_for(path)}" ) def _dsn() -> str: d = os.environ["DATABEND_DSN"] if d.startswith("lake://"): d = "databend://" + d[len("lake://"):] if "warehouse=" not in d: d += ("&" if "?" in d else "?") + f"warehouse={WAREHOUSE}" return d def write_flow_artifact() -> None: header = [ "-- GENERATED by transform/taskflow/deploy.py — do not edit by hand.", "-- Native Databend Task Flow expression of the salesforce_org_monthly_revrec pipeline,", "-- parallel to the dbt build (ADR 0019). The task SQL bodies are the dbt-compiled output.", "-- NOTE: this Databend build does not support the task AFTER clause, so these are standalone", "-- tasks executed in topological order by deploy.py (system.task_history gives observability).", "", "CREATE DATABASE IF NOT EXISTS transform_taskflow_stg;", "CREATE DATABASE IF NOT EXISTS transform_taskflow_dwh;", "", ] blocks = [task_sql(t, p, db, mat) + ";\n" for t, p, db, mat in MODELS] FLOW_OUT.parent.mkdir(parents=True, exist_ok=True) FLOW_OUT.write_text("\n".join(header) + "\n" + "\n".join(blocks)) print(f"wrote {FLOW_OUT.relative_to(REPO)}") def main() -> None: cur = BlockingDatabendClient(_dsn()).cursor() cur.execute("CREATE DATABASE IF NOT EXISTS transform_taskflow_stg") cur.execute("CREATE DATABASE IF NOT EXISTS transform_taskflow_dwh") write_flow_artifact() for task, path, db, mat in MODELS: cur.execute(f"DROP TASK IF EXISTS {task}") cur.execute(task_sql(task, path, db, mat)) cur.execute(f"ALTER TASK {task} RESUME") cur.execute(f"EXECUTE TASK {task}") state, err = None, "" for _ in range(90): cur.execute( f"SELECT state, exception_text FROM system.task_history " f"WHERE name = '{task}' ORDER BY scheduled_time DESC LIMIT 1" ) rows = [tuple(r.values()) if hasattr(r, "values") else tuple(r) for r in cur.fetchall()] if rows: state, err = rows[0][0], rows[0][1] if state in ("SUCCEEDED", "FAILED"): break time.sleep(2) cur.execute(f"ALTER TASK {task} SUSPEND") print(f" {str(state):10s} {task}") if state != "SUCCEEDED": print(" EXCEPTION:", err) sys.exit(f"task {task} did not succeed (state={state})") print("all Task Flow tasks succeeded") if __name__ == "__main__": main()