implement P5 Databricks: DAB bundle, Unity deploy, config targets, parity (fix #18, fix #19, fix #20, fix #21)

- databricks.yml: DAB bundle with all 13 pipelines as Databricks jobs
  in dependency order, parameterized catalog via bundle variables
- stack lake deploy --target databricks: wire setup_catalog_from_schemas
  to CLI for Unity Catalog schema/table creation
- stack.toml: add [lake.databricks] and [default] sections for
  config-driven target switching (duckdb/databricks/trino/lake)
- stack run: default target from stack.toml [default.target]
- stack lake parity: compare DuckDB table schemas against SQLTable models

databricks.yml

@@ -0,0 +1,173 @@
+# Databricks Asset Bundle — deploys all pipelines as Databricks jobs.
+#
+# Usage:
+#   databricks bundle validate
+#   databricks bundle deploy --target dev
+#   databricks bundle run stack_pipelines --target dev
+
+bundle:
+  name: stack
+
+variables:
+  catalog:
+    description: Unity Catalog name for table references
+    default: aco
+
+workspace:
+  host: ${DATABRICKS_HOST}
+
+targets:
+  dev:
+    mode: development
+    default: true
+    variables:
+      catalog: aco_dev
+  staging:
+    variables:
+      catalog: aco_staging
+  prod:
+    mode: production
+    variables:
+      catalog: aco
+
+resources:
+  jobs:
+    stack_pipelines:
+      name: "stack-pipelines"
+      description: "Run all 13 ACO pipelines in dependency order"
+      schedule:
+        quartz_cron_expression: "0 0 6 * * ?"
+        timezone_id: America/New_York
+      job_clusters:
+        - job_cluster_key: pipeline_cluster
+          new_cluster:
+            spark_version: "15.4.x-scala2.12"
+            node_type_id: "i3.xlarge"
+            num_workers: 1
+            spark_conf:
+              spark.sql.catalog.${var.catalog}: "org.apache.iceberg.spark.SparkCatalog"
+
+      tasks:
+        - task_key: input_layer
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "input_layer", "--target", "databricks", "--save"]
+
+        - task_key: cclf
+          depends_on:
+            - task_key: input_layer
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "cclf", "--target", "databricks", "--save"]
+
+        - task_key: claims_preprocessing
+          depends_on:
+            - task_key: input_layer
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "claims_preprocessing", "--target", "databricks", "--save"]
+
+        - task_key: core
+          depends_on:
+            - task_key: claims_preprocessing
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "core", "--target", "databricks", "--save"]
+
+        - task_key: readmissions
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "readmissions", "--target", "databricks", "--save"]
+
+        - task_key: pharmacy
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "pharmacy", "--target", "databricks", "--save"]
+
+        - task_key: hcc_suspecting
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "hcc_suspecting", "--target", "databricks", "--save"]
+
+        - task_key: provider_attribution
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "provider_attribution", "--target", "databricks", "--save"]
+
+        - task_key: quality_measures
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "quality_measures", "--target", "databricks", "--save"]
+
+        - task_key: cms_quality_measures
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "cms_quality_measures", "--target", "databricks", "--save"]
+
+        - task_key: ahrq_measures
+          depends_on:
+            - task_key: core
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "ahrq_measures", "--target", "databricks", "--save"]
+
+        - task_key: data_quality
+          depends_on:
+            - task_key: core
+            - task_key: readmissions
+            - task_key: pharmacy
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "data_quality", "--target", "databricks", "--save"]
+
+        - task_key: main
+          depends_on:
+            - task_key: readmissions
+            - task_key: pharmacy
+            - task_key: hcc_suspecting
+            - task_key: provider_attribution
+            - task_key: quality_measures
+            - task_key: cms_quality_measures
+            - task_key: ahrq_measures
+            - task_key: data_quality
+          job_cluster_key: pipeline_cluster
+          python_wheel_task:
+            package_name: stack
+            entry_point: cli
+            parameters: ["run", "main", "--target", "databricks", "--save"]

src/cli/lake.py

@@ -11,27 +11,63 @@ app = typer.Typer(no_args_is_help=True)
 
 @app.command()
 def deploy(
+    target: str = typer.Option(
+        "iceberg", help="Target: iceberg (nessie/polaris) or databricks."
+    ),
     catalog_type: str = typer.Option("nessie", help="Catalog type: nessie or polaris."),
     schema: Optional[list[str]] = typer.Option(None, help="Only deploy these schemas."),
     dry_run: bool = typer.Option(False, "--dry-run", help="Report without creating."),
 ) -> None:
-    """Create Iceberg schemas and tables from SQLTable models."""
+    """Create schemas and tables from SQLTable models."""
+    if target == "databricks":
+        _deploy_databricks(schemas=schema, dry_run=dry_run)
+    else:
+        _deploy_iceberg(catalog_type=catalog_type, schemas=schema, dry_run=dry_run)
+
+
+def _deploy_iceberg(
+    catalog_type: str, schemas: list[str] | None, dry_run: bool
+) -> None:
     from aco.lake.deploy import deploy_schemas
 
     results = deploy_schemas(
         catalog_type=catalog_type,
-        schemas=schema,
+        schemas=schemas,
         dry_run=dry_run,
     )
 
     total = sum(len(tables) for tables in results.values())
     verb = "Would create" if dry_run else "Deployed"
-    for schema_name, tables in sorted(results.items()):
+    for tables in results.values():
         for t in tables:
             typer.echo(f"  {t}")
     typer.echo(f"{verb} {total} tables across {len(results)} schemas.")
 
 
+def _deploy_databricks(schemas: list[str] | None, dry_run: bool) -> None:
+    from aco.lake.unity import UnityClient, setup_catalog_from_schemas
+    from conf import cfg
+
+    client = UnityClient.from_env()
+    catalog_name = cfg.lake.databricks.catalog
+    warehouse_id = cfg.lake.databricks.get("warehouse_id", "")
+
+    report = setup_catalog_from_schemas(
+        client=client,
+        catalog_name=catalog_name,
+        warehouse_id=warehouse_id,
+        dry_run=dry_run,
+    )
+
+    created = len(report.get("tables_created", []))
+    errors = len(report.get("errors", []))
+    typer.echo(f"Databricks: {created} tables created, {errors} errors.")
+    if errors:
+        for err in report["errors"]:
+            typer.echo(f"  ERROR: {err}", err=True)
+        raise typer.Exit(code=1)
+
+
 @app.command()
 def load(
     catalog_type: str = typer.Option("nessie", help="Catalog type: nessie or polaris."),
@@ -101,3 +137,51 @@ def validate(
     else:
         typer.echo(f"{issues} issues found.")
         raise typer.Exit(code=1)
+
+
+@app.command()
+def parity(
+    schema: Optional[list[str]] = typer.Option(None, help="Only check these schemas."),
+) -> None:
+    """Compare pipeline outputs between DuckDB and Databricks."""
+    from aco.lake.catalog import Catalog
+    from aco.lake.context import DuckDBContext
+    from conf import path
+
+    cat = Catalog()
+    duckdb_ctx = DuckDBContext(database=str(path("db.aco")), read_only=True)
+
+    target_schemas = schema or cat.schemas()
+    mismatches = 0
+
+    for schema_name in target_schemas:
+        tables = cat.tables(schema_name)
+        for table_ref in tables:
+            model = cat.model(table_ref)
+            expected_cols = sorted(model.model_fields.keys())
+
+            try:
+                df = duckdb_ctx.load(table_ref)
+                actual_cols = sorted(df.columns)
+                row_count = len(df)
+
+                if actual_cols != expected_cols:
+                    missing = set(expected_cols) - set(actual_cols)
+                    extra = set(actual_cols) - set(expected_cols)
+                    typer.echo(f"  {table_ref}: column mismatch")
+                    if missing:
+                        typer.echo(f"    missing: {sorted(missing)}")
+                    if extra:
+                        typer.echo(f"    extra: {sorted(extra)}")
+                    mismatches += 1
+                else:
+                    typer.echo(f"  {table_ref}: {row_count:,} rows, schema OK")
+            except Exception as e:
+                typer.echo(f"  {table_ref}: ERROR {e}")
+                mismatches += 1
+
+    if mismatches == 0:
+        typer.echo("All tables have matching schemas.")
+    else:
+        typer.echo(f"{mismatches} mismatches found.")
+        raise typer.Exit(code=1)

src/cli/run.py

@@ -8,7 +8,8 @@ import typer
 def run(
     name: str = typer.Argument(help="Pipeline name (e.g. readmissions)."),
     target: str = typer.Option(
-        "local", help="Target context: local, lake, trino, databricks."
+        "",
+        help="Target context: local, lake, trino, databricks. Defaults to stack.toml.",
     ),
     save: bool = typer.Option(
         False, "--save", help="Save pipeline outputs back to the database."
@@ -22,6 +23,12 @@ def run(
         typer.echo(f"Unknown pipeline '{name}'. Available: {names}", err=True)
         raise typer.Exit(1)
 
+    if not target:
+        from conf import cfg
+
+        target = getattr(cfg, "default", None)
+        target = getattr(target, "target", "local") if target else "local"
+
     pipeline = registry[name]
     ctx = _make_context(target, read_only=not save)
 
@@ -69,9 +76,10 @@ def _make_context(target: str, *, read_only: bool = True):  # noqa: ANN202
     if target == "databricks":
         from aco.lake.context import EnterpriseContext
 
+        db_cfg = cfg.lake.databricks
         return EnterpriseContext(
-            catalog_uri=cfg.lake.get("databricks_uri", ""),
+            catalog_uri=getattr(db_cfg, "catalog_uri", ""),
-            warehouse=cfg.lake.get("databricks_warehouse", "main"),
+            warehouse=getattr(db_cfg, "warehouse", "main"),
             dialect="databricks",
         )
 

stack.toml

@@ -68,11 +68,19 @@ catalog_uri = "http://nessie:19120/iceberg/"
 [lake.polaris]
 catalog_uri = "http://polaris:8181/api/catalog"
 
+[lake.databricks]
+catalog_uri = ""   # set via DATABRICKS_HOST env var
+warehouse = "main" # Unity Catalog name
+catalog = "aco"    # default catalog for table references
+
 [lake.trino]
 host = "trino"
 port = 8080
 catalog = "iceberg"
 
+[default]
+target = "local" # duckdb | databricks | trino | lake
+
 [api]
 host = "0.0.0.0"
 port = 8000

tests/cli/test_cli.py

@@ -100,6 +100,11 @@ class TestLake:
         assert result.exit_code == 0
         assert "catalog-type" in result.output
 
+    def test_lake_parity_help(self) -> None:
+        result = runner.invoke(app, ["lake", "parity", "--help"])
+        assert result.exit_code == 0
+        assert "schema" in result.output
+
 
 class TestDb:
     def test_db_comment(self) -> None: