fix(app): carry table and preview flag in staged-config stub

The staged-config stub now includes source_table_fqn, skip_history and
run_type, so the Jobs API failure feed can name a staged run's table and
exclude staged previews without a SQL lookup. This removes the last
warehouse query from the recent-failures endpoints.

Co-authored-by: Isaac <no-reply@databricks.com>

Author: OGordon100

app/src/databricks_labs_dqx_app/backend/routes/v1/dryrun.py

@@ -1,4 +1,3 @@
-import asyncio
 import json
 from collections.abc import Callable
 from typing import Annotated, Any
@@ -241,7 +240,6 @@ async def list_validation_runs(
 async def list_recent_validation_failures(
     job_svc: Annotated[JobService, Depends(get_job_service)],
     user_catalogs: Annotated[frozenset[str], Depends(get_user_catalog_names)],
-    sql: Annotated[SqlExecutor, Depends(get_sp_sql_executor)],
 ) -> list[RunFailureOut]:
     """Return recently-failed validation runs, bounded to the most recent *N*.
 
@@ -253,16 +251,11 @@ async def list_recent_validation_failures(
     """
     try:
         failed = [run for run in await job_svc.list_recent_failed_runs() if run.task_type in _VALIDATION_TASK_TYPES]
-        # Oversized configs are staged out of the job parameters, so only the
-        # run table can name their source table — a rare, failure-only lookup.
-        staged = [run.app_run_id for run in failed if not run.source_table_fqn]
-        staged_tables = (
-            await asyncio.to_thread(job_svc.lookup_source_tables, sql.fqn(_DRYRUN_TABLE), staged) if staged else {}
-        )
 
         results: list[RunFailureOut] = []
         for run in failed:
-            fqn = run.source_table_fqn or staged_tables.get(run.app_run_id) or ""
+            # A run with no table cannot pass the catalog filter, so it is skipped.
+            fqn = run.source_table_fqn or ""
             if not fqn or (not fqn.startswith(_SQL_CHECK_PREFIX) and _catalog_of(fqn) not in user_catalogs):
                 continue
             results.append(

app/src/databricks_labs_dqx_app/backend/routes/v1/profiler.py

@@ -1,4 +1,3 @@
-import asyncio
 import json
 from typing import Annotated
 from uuid import uuid4
@@ -262,19 +261,13 @@ async def list_recent_profile_failures(
     """
     try:
         failed = [run for run in await job_svc.list_recent_failed_runs() if run.task_type == "profile"]
-        # Oversized configs are staged out of the job parameters, so only the
-        # run table can name their source table — a rare, failure-only lookup.
-        staged = [run.app_run_id for run in failed if not run.source_table_fqn]
-        staged_tables = (
-            await asyncio.to_thread(job_svc.lookup_source_tables, _run_table_fqn(), staged) if staged else {}
-        )
 
         results: list[RunFailureOut] = []
         for run in failed:
             results.append(
                 RunFailureOut(
                     run_id=run.app_run_id,
-                    source_table_fqn=run.source_table_fqn or staged_tables.get(run.app_run_id) or "",
+                    source_table_fqn=run.source_table_fqn or "",
                     status="FAILED",
                     created_at=run.created_at,
                 )

app/src/databricks_labs_dqx_app/backend/run_config_store.py

@@ -7,6 +7,10 @@
 into the ``dq_run_configs`` Delta table keyed by ``run_id`` and passes a
 tiny stub ``{"__manifest__": true}`` instead. The task runner reads the config
 from the table using the ``run_id`` and deletes the row once the run finishes.
+
+The stub also carries the few small config fields the app reads back from
+the Jobs API (see *MANIFEST_SUMMARY_KEYS*), so a staged run can still be
+attributed to its table and recognised as a preview without a SQL lookup.
 """
 
 import base64
@@ -25,6 +29,12 @@
 # Marker key in the inline stub passed to the task runner.
 MANIFEST_CONFIG_KEY = "__manifest__"
 
+# Small config fields copied into the stub so Jobs API readers
+# (services.task_runner_runs) can name a staged run's table and tell previews
+# apart without reading the staged config. The runner ignores them: it loads
+# the full config from the table whenever the manifest key is set.
+MANIFEST_SUMMARY_KEYS = ("source_table_fqn", "skip_history", "run_type")
+
 # Delta table holding run configs that are too large to inline.
 RUN_CONFIGS_TABLE = "dq_run_configs"
 
@@ -78,9 +88,10 @@ def build_inline_config_payload(config: dict[str, Any]) -> str:
     return _compact_json(config)
 
 
-def build_manifest_config_payload() -> str:
-    """Serialize the stub the task runner uses to load a staged config from the table."""
-    return _compact_json({MANIFEST_CONFIG_KEY: True})
+def build_manifest_config_payload(config: dict[str, Any]) -> str:
+    """Serialize the stub the task runner uses to load *config* back from the table."""
+    summary = {key: config[key] for key in MANIFEST_SUMMARY_KEYS if key in config}
+    return _compact_json({MANIFEST_CONFIG_KEY: True, **summary})
 
 
 def stage_config_to_table(sql: SqlExecutor, run_id: str, config: dict[str, Any]) -> None:
@@ -145,7 +156,7 @@ def prepare_config_json(
         raise
     except Exception as exc:
         raise RunConfigStagingError(run_id, sql.fqn(RUN_CONFIGS_TABLE), exc) from exc
-    manifest = build_manifest_config_payload()
+    manifest = build_manifest_config_payload(config)
     manifest_params = {**job_parameters_without_config, "config_json": manifest}
     manifest_size = job_parameters_size(manifest_params)
     if manifest_size > JOB_PARAMETERS_CHAR_LIMIT:

app/src/databricks_labs_dqx_app/backend/services/job_service.py

@@ -79,7 +79,7 @@ def submit_run(
             config=config,
             job_parameters_without_config=base_params,
         )
-        staged = config_json == build_manifest_config_payload()
+        staged = config_json == build_manifest_config_payload(config)
 
         try:
             run = self._ws.jobs.run_now(
@@ -125,20 +125,6 @@ async def list_recent_failed_runs(self) -> list[TaskRunnerRun]:
         runs = await cached_recent_completed_runs(self._ws, self._job_id)
         return [run for run in runs if run.is_failed and not run.is_preview]
 
-    def lookup_source_tables(self, table: str, run_ids: list[str]) -> dict[str, str]:
-        """Map *run_ids* to their ``source_table_fqn`` from the run table *table*.
-
-        Only needed for runs whose config was staged out of the job parameters
-        (oversized configs), so the Jobs API alone cannot name their table.
-        """
-        if not run_ids:
-            return {}
-        in_list = ", ".join(f"'{escape_sql_string(run_id)}'" for run_id in run_ids)
-        rows = self._sql.query(
-            f"SELECT DISTINCT run_id, source_table_fqn FROM {table} WHERE run_id IN ({in_list})"  # noqa: S608
-        )
-        return {row[0]: row[1] for row in rows if row and row[0] and len(row) > 1 and row[1]}
-
     def get_run_creator(self, job_run_id: int) -> str | None:
         """Return the requesting end-user for a job run, or None if unavailable.
 

app/tests/test_job_service.py

@@ -168,22 +168,3 @@ async def test_recent_failed_runs_empty_without_a_job(sql_executor_mock: MagicMo
 
     assert await service.list_recent_failed_runs() == []
     ws.jobs.list_runs.assert_not_called()
-
-
-def test_lookup_source_tables_maps_run_ids(sql_executor_mock: MagicMock) -> None:
-    service, _ws = _job_service(sql_executor_mock)
-    sql_executor_mock.query.return_value = [("r1", "main.s.t1"), ("r2", None)]
-
-    result = service.lookup_source_tables("dqx.dqx_studio.dq_validation_runs", ["r1", "r2", "o'brien"])
-
-    assert result == {"r1": "main.s.t1"}
-    statement = sql_executor_mock.query.call_args.args[0]
-    assert "FROM dqx.dqx_studio.dq_validation_runs" in statement
-    assert "IN ('r1', 'r2', 'o''brien')" in statement
-
-
-def test_lookup_source_tables_skips_the_query_for_no_runs(sql_executor_mock: MagicMock) -> None:
-    service, _ws = _job_service(sql_executor_mock)
-
-    assert service.lookup_source_tables("dqx.dqx_studio.dq_validation_runs", []) == {}
-    sql_executor_mock.query.assert_not_called()

app/tests/test_recent_failures.py

@@ -8,8 +8,7 @@
 1. Only the endpoint's own task types are returned (validation vs. profile).
 2. The validation feed honours the caller's catalog access.
 3. The result is bounded at _RECENT_FAILURES_LIMIT with the minimal RunFailureOut shape.
-4. The run table is queried only for runs whose config was staged out of the
-   job parameters, and never otherwise.
+4. Neither feed ever queries the SQL warehouse.
 """
 
 from unittest.mock import MagicMock, create_autospec
@@ -27,7 +26,6 @@
 )
 from databricks_labs_dqx_app.backend.services.job_service import JobService
 from databricks_labs_dqx_app.backend.services.task_runner_runs import TaskRunnerRun
-from databricks_labs_dqx_app.backend.sql_executor import SqlExecutor
 
 
 def _failed_run(
@@ -50,67 +48,58 @@ def _failed_run(
 
 @pytest.fixture
 def job_service_mock() -> MagicMock:
-    svc = create_autospec(JobService, instance=True)
-    svc.lookup_source_tables.return_value = {}
-    return svc
+    return create_autospec(JobService, instance=True)
 
 
-@pytest.fixture
-def sql_executor() -> MagicMock:
-    sql = create_autospec(SqlExecutor, instance=True)
-    sql.fqn.side_effect = lambda name: f"dqx.dqx_studio.{name}"
-    return sql
-
-
-async def _validation(job_svc: MagicMock, sql: MagicMock, catalogs: frozenset[str] = frozenset({"main"})):
-    return await list_recent_validation_failures(job_svc=job_svc, user_catalogs=catalogs, sql=sql)
+async def _validation(job_svc: MagicMock, catalogs: frozenset[str] = frozenset({"main"})):
+    return await list_recent_validation_failures(job_svc=job_svc, user_catalogs=catalogs)
 
 
 class TestListRecentValidationFailures:
-    async def test_returns_only_validation_task_types(self, job_service_mock, sql_executor):
+    async def test_returns_only_validation_task_types(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = [
             _failed_run("run-dryrun"),
             _failed_run("run-scheduled", task_type="scheduled"),
             _failed_run("run-profile", task_type="profile"),
         ]
 
-        result = await _validation(job_service_mock, sql_executor)
+        result = await _validation(job_service_mock)
 
         assert [r.run_id for r in result] == ["run-dryrun", "run-scheduled"]
         assert all(r.status == "FAILED" for r in result)
 
-    async def test_excludes_runs_from_inaccessible_catalogs(self, job_service_mock, sql_executor):
+    async def test_excludes_runs_from_inaccessible_catalogs(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = [
             _failed_run("run-visible", fqn="main.public.orders"),
             _failed_run("run-hidden", fqn="restricted.public.orders"),
         ]
 
-        result = await _validation(job_service_mock, sql_executor)
+        result = await _validation(job_service_mock)
 
         assert [r.run_id for r in result] == ["run-visible"]
 
-    async def test_includes_sql_check_prefix_runs(self, job_service_mock, sql_executor):
+    async def test_includes_sql_check_prefix_runs(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = [
             _failed_run("run-sql", fqn="__sql_check__/my_check"),
         ]
 
-        result = await _validation(job_service_mock, sql_executor, catalogs=frozenset())
+        result = await _validation(job_service_mock, catalogs=frozenset())
 
         assert [r.source_table_fqn for r in result] == ["__sql_check__/my_check"]
 
-    async def test_result_bounded_at_limit(self, job_service_mock, sql_executor):
+    async def test_result_bounded_at_limit(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = [
             _failed_run(f"f-{i}") for i in range(DRYRUN_LIMIT + 10)
         ]
 
-        result = await _validation(job_service_mock, sql_executor)
+        result = await _validation(job_service_mock)
 
         assert len(result) == DRYRUN_LIMIT
 
-    async def test_returns_minimal_fields_only(self, job_service_mock, sql_executor):
+    async def test_returns_minimal_fields_only(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = [_failed_run("run-failed")]
 
-        result = await _validation(job_service_mock, sql_executor)
+        result = await _validation(job_service_mock)
 
         assert result == [
             RunFailureOut(
@@ -121,39 +110,22 @@ async def test_returns_minimal_fields_only(self, job_service_mock, sql_executor)
             )
         ]
 
-    async def test_never_queries_the_warehouse_when_configs_are_inline(self, job_service_mock, sql_executor):
-        job_service_mock.list_recent_failed_runs.return_value = [_failed_run("run-failed")]
-
-        await _validation(job_service_mock, sql_executor)
-
-        job_service_mock.lookup_source_tables.assert_not_called()
-        sql_executor.query.assert_not_called()
-        sql_executor.query_dicts.assert_not_called()
-
-    async def test_staged_config_runs_resolve_their_table_from_the_run_table(self, job_service_mock, sql_executor):
+    async def test_runs_without_a_source_table_are_skipped(self, job_service_mock):
+        # The catalog filter cannot be applied to a run with no table, so it
+        # is dropped rather than shown to everyone.
         job_service_mock.list_recent_failed_runs.return_value = [
-            _failed_run("run-inline"),
-            _failed_run("run-staged", fqn=None),
+            _failed_run("run-known"),
             _failed_run("run-unknown", fqn=None),
         ]
-        job_service_mock.lookup_source_tables.return_value = {"run-staged": "main.public.big"}
-
-        result = await _validation(job_service_mock, sql_executor)
-
-        job_service_mock.lookup_source_tables.assert_called_once_with(
-            "dqx.dqx_studio.dq_validation_runs", ["run-staged", "run-unknown"]
-        )
-        # A staged run whose table cannot be resolved is dropped (the catalog
-        # filter cannot be applied to it).
-        assert [(r.run_id, r.source_table_fqn) for r in result] == [
-            ("run-inline", "main.public.orders"),
-            ("run-staged", "main.public.big"),
-        ]
 
-    async def test_empty_list_when_no_failures(self, job_service_mock, sql_executor):
+        result = await _validation(job_service_mock)
+
+        assert [r.run_id for r in result] == ["run-known"]
+
+    async def test_empty_list_when_no_failures(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = []
 
-        assert await _validation(job_service_mock, sql_executor) == []
+        assert await _validation(job_service_mock) == []
 
 
 class TestListRecentProfileFailures:
@@ -189,7 +161,6 @@ async def test_returns_minimal_fields_only(self, job_service_mock):
                 created_at="2026-09-21T14:13:20+00:00",
             )
         ]
-        job_service_mock.lookup_source_tables.assert_not_called()
 
     async def test_empty_list_when_no_failures(self, job_service_mock):
         job_service_mock.list_recent_failed_runs.return_value = []

app/tests/test_run_config_store.py

@@ -95,6 +95,25 @@ def test_stages_when_over_limit(self) -> None:
         assert kwargs["key_cols"] == {"run_id": "run123"}
         assert _staged_config(kwargs["value_cols"]["config"]) == config
 
+    def test_stub_carries_the_fields_jobs_api_readers_need(self) -> None:
+        # The failure toast reads a staged run's table and preview flag from
+        # the job parameters, so they must survive staging (no SQL lookup).
+        config = {**_big_config(), "source_table_fqn": "main.sales.orders", "skip_history": True, "run_type": "preview"}
+
+        result = prepare_config_json(
+            _sql_mock(),
+            run_id="run123",
+            config=config,
+            job_parameters_without_config=_base_params(),
+        )
+
+        assert json.loads(result) == {
+            MANIFEST_CONFIG_KEY: True,
+            "source_table_fqn": "main.sales.orders",
+            "skip_history": True,
+            "run_type": "preview",
+        }
+
     def test_stub_stays_within_the_job_parameter_limit(self) -> None:
         # Regression guard: the stub plus base params must always fit, so a
         # staged config never re-trips the limit it was meant to dodge.

app/tests/test_task_runner_runs.py

@@ -63,6 +63,13 @@ def test_staged_config_has_no_source_table(self):
 
         assert run is not None and run.source_table_fqn is None
 
+    def test_staged_config_stub_still_names_its_table_and_preview_flag(self):
+        run = parse_task_runner_run(
+            make_run(config={"__manifest__": True, "source_table_fqn": "main.sales.big", "skip_history": True})
+        )
+
+        assert run is not None and run.source_table_fqn == "main.sales.big" and run.is_preview
+
     def test_unparseable_config_has_no_source_table(self):
         run = parse_task_runner_run(make_run(config="not json"))