feat(e2e): add end-to-end test job on Databricks serverless

.github/workflows/e2e.yml

@@ -0,0 +1,101 @@
+name: e2e
+
+# End-to-end tests on Databricks serverless (see databricks.yml). On PRs into main, nightly
+# and on demand.
+on:
+  pull_request:
+    branches: [main]
+    # `closed` removes the PR's bundle.
+    types: [opened, synchronize, reopened, ready_for_review, closed]
+  schedule:
+    # 04:00 UTC daily.
+    - cron: "0 4 * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+# Each PR deploys its own copy of the bundle and job (the suffix names its bundle root), so
+# PRs run in parallel. A new push replaces the PR's running check; nightly and manual runs
+# share the default bundle and queue.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || 'default' }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+env:
+  BUNDLE_VAR_suffix: ${{ github.event.pull_request.number && format('-pr-{0}', github.event.pull_request.number) || '' }}
+
+jobs:
+  e2e:
+    name: E2E on Databricks (serverless)
+    # Fork PRs get no secrets; mirror them with `make fork-sync PR=<n>` (see acceptance.yml).
+    if: >-
+      github.event.action != 'closed'
+      && !github.event.pull_request.draft
+      && !github.event.pull_request.head.repo.fork
+    runs-on:
+      group: databrickslabs-protected-runner-group
+      labels: linux-ubuntu-latest
+    environment: e2e
+    permissions:
+      # For JFrog OIDC.
+      id-token: write
+    env:
+      DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
+      DATABRICKS_CLIENT_ID: ${{ secrets.DATABRICKS_CLIENT_ID }}
+      DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_CLIENT_SECRET }}
+      DATABRICKS_AUTH_TYPE: oauth-m2m
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Install UV
+        uses: astral-sh/setup-uv@5a095e7a2014a4212f075830d4f7277575a9d098 # v7.3.1
+        with:
+          enable-cache: false
+          version: "0.11.2"
+          checksum: "7ac2ca0449c8d68dae9b99e635cd3bc9b22a4cb1de64b7c43716398447d42981"
+
+      - name: Setup for JFrog
+        uses: ./.github/actions/jfrog-auth
+
+      - name: Install dependencies
+        run: make dev
+
+      - name: Install Databricks CLI
+        uses: databricks/setup-cli@8821c27b1fe57472264eeb6e50c1f31d69552f28 # v1.10.0
+
+      # Also builds the wheel.
+      - name: Deploy bundle
+        run: databricks bundle deploy -t e2e
+
+      - name: Run e2e job
+        run: databricks bundle run e2e_tests -t e2e
+
+      # A cancelled workflow leaves the job run going; stop it too.
+      - name: Cancel e2e job run
+        if: cancelled()
+        run: |
+          job_id=$(databricks bundle summary -t e2e -o json | jq -r '.resources.jobs.e2e_tests.id')
+          databricks jobs cancel-all-runs --job-id "$job_id"
+
+  # The PR's bundle is kept while the PR is open, so failed runs can be inspected.
+  destroy:
+    name: Remove the PR's e2e bundle
+    if: github.event.action == 'closed' && !github.event.pull_request.head.repo.fork
+    runs-on:
+      group: databrickslabs-protected-runner-group
+      labels: linux-ubuntu-latest
+    environment: e2e
+    env:
+      DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
+      DATABRICKS_CLIENT_ID: ${{ secrets.DATABRICKS_CLIENT_ID }}
+      DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_CLIENT_SECRET }}
+      DATABRICKS_AUTH_TYPE: oauth-m2m
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Install Databricks CLI
+        uses: databricks/setup-cli@8821c27b1fe57472264eeb6e50c1f31d69552f28 # v1.10.0
+
+      - name: Destroy bundle
+        run: databricks bundle destroy -t e2e --auto-approve

.gitignore

@@ -51,6 +51,7 @@ logs/
 # Local development settings
 .env
 .env.local
+scripts/local/
 
 # macOS specific
 .DS_Store

Makefile

@@ -45,6 +45,11 @@ coverage:
 	$(UV_RUN) pytest tests/ --cov=src --cov-branch --cov-report=html
 	open htmlcov/index.html
 
+# Run the end-to-end suite on Databricks. Needs the Databricks CLI and workspace auth.
+test-e2e:
+	databricks bundle deploy -t e2e
+	databricks bundle run e2e_tests -t e2e
+
 build:
 	uv build --require-hashes --build-constraints=.build-constraints.txt
 
@@ -67,4 +72,4 @@ fork-sync:
 	./.github/scripts/fork-sync-pr.sh $(PR)
 
 .DEFAULT: all
-.PHONY: all clean dev lint fmt test coverage build update-api-docs lock-dependencies fork-sync
+.PHONY: all clean dev lint fmt test test-e2e coverage build update-api-docs lock-dependencies fork-sync

databricks.yml

@@ -0,0 +1,37 @@
+# Bundle for the end-to-end tests: one serverless job that runs tests/e2e.
+bundle:
+  name: impulse-e2e
+
+artifacts:
+  impulse_wheel:
+    type: whl
+    # Clean first so the dist/*.whl glob in resources/e2e_job.yml matches one wheel.
+    build: rm -rf dist && uv build --wheel
+    path: .
+    # Version the wheel by build time, so serverless installs every deploy's code.
+    dynamic_version: true
+
+include:
+  - resources/*.yml
+
+variables:
+  catalog:
+    description: Catalog for the ephemeral test schemas.
+    default: maxim_catalog
+  source_catalog:
+    description: Catalog with the silver test data (read-only).
+    default: maxim_catalog
+  source_schema:
+    description: Schema under `source_catalog` with the silver test data (read-only).
+    default: default
+  suffix:
+    description: Suffix of the bundle root and job name, so each PR deploys its own copy.
+    default: ""
+
+targets:
+  # Workspace host and auth come from the environment.
+  e2e:
+    default: true
+    mode: development
+    workspace:
+      root_path: ~/.bundle/${bundle.name}/${bundle.target}${var.suffix}

resources/e2e_job.yml

@@ -0,0 +1,81 @@
+resources:
+  jobs:
+    e2e_tests:
+      name: "impulse e2e tests${var.suffix}"
+      timeout_seconds: 3600
+      max_concurrent_runs: 1
+
+      tasks:
+        - task_key: setup
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          notebook_task:
+            notebook_path: ../tests/e2e/setup_schema.py
+            base_parameters: &params
+              catalog: ${var.catalog}
+              schema: e2e_{{job.run_id}}
+              source_catalog: ${var.source_catalog}
+              source_schema: ${var.source_schema}
+              # Each test writes its tables with this prefix and reads expected/<prefix>/.
+              prefix: "{{task.name}}"
+              expected_dir: ${workspace.file_path}/tests/e2e/expected
+              actual_dir: ${workspace.root_path}/actual
+
+        # The tests run in parallel.
+        - task_key: rpm_histogram
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_rpm_histogram.py, base_parameters: *params}
+
+        - task_key: incremental
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_incremental.py, base_parameters: *params}
+
+        - task_key: pipeline
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_pipeline.py, base_parameters: *params}
+
+        - task_key: container_filter
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_container_filter.py, base_parameters: *params}
+
+        - task_key: incremental_updates
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_incremental_updates.py, base_parameters: *params}
+
+        - task_key: events
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on: [{task_key: setup}]
+          notebook_task: {notebook_path: ../tests/e2e/e2e_events.py, base_parameters: *params}
+
+        # Runs only when all tests passed. Otherwise it ends as "upstream failed", which fails
+        # the job run, and the schema is kept for debugging.
+        - task_key: teardown
+          environment_key: e2e_env
+          disable_auto_optimization: true
+          depends_on:
+            - {task_key: rpm_histogram}
+            - {task_key: incremental}
+            - {task_key: pipeline}
+            - {task_key: container_filter}
+            - {task_key: incremental_updates}
+            - {task_key: events}
+          notebook_task: {notebook_path: ../tests/e2e/teardown.py, base_parameters: *params}
+
+      environments:
+        - environment_key: e2e_env
+          spec:
+            # pyspark and delta come with the serverless environment.
+            environment_version: "5"
+            dependencies:
+              - ../dist/*.whl

tests/e2e/README.md

@@ -0,0 +1,46 @@
+# End-to-end tests
+
+Runs Impulse reports on Databricks serverless compute and compares the results with the expected files in `expected/`.
+
+## Job
+
+The job is defined in `resources/e2e_job.yml`. Each task is a notebook in this folder. After `setup`, the tests run in parallel in one temporary schema `e2e_<run_id>`, each with its own table prefix (the task name).
+
+| Task | Notebook | Checks |
+|---|---|---|
+| `setup` | `setup_schema` | Creates the schema, drops this job's schemas older than one day, checks the source data against `expected/containers.csv` |
+| `rpm_histogram` | `e2e_rpm_histogram` | Baseline report (event `Engine RPM > 2000`, duration histogram of `Engine RPM`): a compute-only run writes nothing and gives the expected histogram and events; two persisted runs update the gold tables in place (same Delta table, new version); the histogram adds up to the event durations |
+| `incremental` | `e2e_incremental` | Baseline report on a silver copy that receives the containers in two loads, run incrementally in batches of 15; the second load leaves the first load's rows untouched and commits one MERGE per batch; must equal `rpm_histogram` |
+| `incremental_updates` | `e2e_incremental_updates` | A rerun without changes rewrites no fact table; a re-ingested container (half its samples) is recomputed, its histogram matching its new events, while the other containers' rows stay untouched; changed histogram bins recompute only the histogram, which still adds up to the event durations; the result equals a full run |
+| `pipeline` | `e2e_pipeline` | Container event, per-container statistics, an RPM x speed heatmap and a calculated channel; the calculated channel's samples must give the same statistics as the StatsAggregator |
+| `container_filter` | `e2e_container_filter` | Baseline report on (`experiment_1` or `experiment_2`) and longer than 30 minutes; must equal those containers' `rpm_histogram` rows, with the containers selected independently in SQL |
+| `events` | `e2e_events` | Points-in-time, sequence and 5-minute window events with point values and window statistics; point events have zero length; windows tile each container without gaps or overlaps; each slow_down has one value per channel, with speed at most 100 |
+| `teardown` | `teardown` | Drops the schema; runs only if all tests passed, otherwise it ends as upstream failed, which fails the job run |
+
+Shared code (report config, baseline report, comparison) is in `e2e_utils.py`. When a test fails, its schema is kept for debugging.
+
+## Running
+
+```bash
+make test-e2e
+```
+
+This deploys the bundle (`databricks.yml`) and runs the job. It needs the Databricks CLI and workspace authentication.
+
+CI runs it via `.github/workflows/e2e.yml` on every push to a PR into main, nightly and on demand. Each PR deploys its own copy of the bundle and job (bundle variable `suffix`, e.g. `-pr-123`), which is removed when the PR is closed. Fork PRs are skipped; mirror them with `make fork-sync PR=<n>`.
+
+## Input data
+
+The reports read the silver tables in `<source_catalog>.<source_schema>` (bundle variables in `databricks.yml`) and write to temporary schemas in `<catalog>`. They were created from MF4 files with the [Impulse MF4 ingest solution accelerator](https://github.com/databricks-industry-solutions/impulse-mf4-ingest).
+
+`expected/containers.csv` lists the MF4 file (`experiment_id` folder and `filename`) behind each container id. Use it to rebuild the silver layer from scratch with the accelerator. If the rebuild assigns different container ids, refresh the expected files.
+
+## Expected results
+
+`expected/<test>/<table>.csv` holds the expected gold rows of a test (without `_` bookkeeping columns). `incremental` and `container_filter` reuse `expected/rpm_histogram/`; `incremental_updates` compares with full reference runs instead of expected files.
+
+To refresh them after an intended change, run the job. On a mismatch, the test uploads its actual rows as CSV to `<bundle root>/actual/<test>/` and logs the path. Review them, then copy them into `expected/`.
+
+## Adding a test
+
+Add a notebook `e2e_<name>.py` that builds a report with `config(p)` and `new_report(...)` from `e2e_utils`, runs it, and calls `check(p, table, gold(p, table))`. Add a task `<name>` to the job and to the `depends_on` of `teardown`. The first run fails and uploads the actual rows; review them and save them as `expected/<name>/`.

tests/e2e/e2e_container_filter.py

@@ -0,0 +1,41 @@
+# Databricks notebook source
+# MAGIC %md
+# MAGIC # e2e: container filter
+# MAGIC Runs the baseline report on the containers of `experiment_1` or `experiment_2` that are
+# MAGIC longer than 30 minutes (tag filter groups OR-combined, AND-combined with a metric filter).
+# MAGIC The result must equal those containers' rows in `expected/rpm_histogram`.
+
+# COMMAND ----------
+
+import pyspark.sql.functions as F
+from e2e_utils import add_rpm_histogram, check, config, gold, new_report, params, spark
+
+p = params()
+experiments = ["experiment_1", "experiment_2"]
+filters = {
+    "tag_filters": [
+        [{"tag_name": "experiment_id", "comparator": "==", "value": e}] for e in experiments
+    ],
+    "metric_filters": [[{"column_name": "duration_s", "comparator": ">", "value": 1800.0}]],
+}
+report = new_report(config(p, container_filters=filters))
+add_rpm_histogram(report)
+report.run()
+
+# COMMAND ----------
+
+# The same selection in plain SQL.
+ids = [
+    r.container_id
+    for r in spark.sql(
+        f"SELECT m.container_id FROM {p.source}container_metrics m "
+        f"JOIN {p.source}container_tags t ON t.container_id = m.container_id "
+        f"WHERE t.key = 'experiment_id' AND t.value IN ('experiment_1', 'experiment_2') "
+        "AND m.duration_s > 1800"
+    ).collect()
+]
+all_ids = spark.read.table(f"{p.source}container_tags").select("container_id").distinct().count()
+assert 0 < len(ids) < all_ids, f"filter selects {len(ids)} of {all_ids} containers"
+
+for table in ["histogram_fact", "event_instance_fact"]:
+    check(p, table, gold(p, table), test="rpm_histogram", where=F.col("container_id").isin(ids))