Batch Monitored Tables N+1 fix + hide Review&Approve when approvals off (B2-141/B2-142)

- B2-141: the Monitored Tables list computed draft check-counts with a
  per-never-approved-binding render loop (~3N + 2*sum(R) sequential OLTP
  round-trips). Batched to 3 constant grouped queries (list_applied_rules_many,
  get_rules_many, get_versions_many) with count parity preserved.
- B2-142: hide the Review & Approve sidebar item and its trailing divider when
  the app-wide approvals mode is 'disabled' (no review queue).

Co-authored-by: Isaac

Author: OGordon100

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

@@ -156,26 +156,42 @@ def _apply_snapshot_check_counts(
       frozen snapshot, resolved for ALL such bindings in one batched
       :meth:`MonitoredTableVersionService.snapshot_counts_many` call;
     * never-approved binding (``version == 0``, no snapshot) -> the live render
-      count (exactly what a draft run would execute).
-
-    Mirrors :class:`DataProductService`'s pinned-vs-live member-count split.
+      count (exactly what a draft run would execute), resolved for ALL such
+      bindings in one batched
+      :meth:`Materializer.render_binding_checks_counts_many` call.
+
+    Both branches are query-bounded regardless of how many bindings the list
+    holds (B2-141): the never-approved branch previously called
+    ``render_binding_checks`` once per binding, fanning out to ~``3N + 2·ΣR``
+    sequential OLTP round-trips on a fresh (all-draft) install; it now costs a
+    constant handful of grouped queries. Mirrors
+    :class:`DataProductService`'s pinned-vs-live member-count split.
     """
     pins = [(s.table.binding_id, s.table.version) for s in summaries if s.table.version > 0]
     snapshot_counts = version_svc.snapshot_counts_many(pins) if pins else {}
+
+    # Bindings that must fall back to the live draft-render count: every
+    # never-approved binding, plus any approved one whose frozen snapshot row
+    # is missing (so the overview still reflects what a draft run would run).
+    live_bindings: list[tuple[str, str]] = []
+    for summary in summaries:
+        version = summary.table.version
+        if version > 0 and snapshot_counts.get((summary.table.binding_id, version)) is not None:
+            continue
+        live_bindings.append((summary.table.binding_id, summary.table.table_fqn))
+    try:
+        live_counts = materializer.render_binding_checks_counts_many(live_bindings)
+    except MaterializationError:
+        live_counts = {}
+
     for summary in summaries:
         version = summary.table.version
         if version > 0:
             snapshot = snapshot_counts.get((summary.table.binding_id, version))
             if snapshot is not None:
                 summary.check_count = snapshot[1]
                 continue
-        # Never approved (or an approved binding whose snapshot row is missing):
-        # fall back to the live render count so the overview still reflects
-        # what a draft run would execute.
-        try:
-            summary.check_count = len(materializer.render_binding_checks(summary.table.binding_id))
-        except MaterializationError:
-            summary.check_count = 0
+        summary.check_count = live_counts.get(summary.table.binding_id, 0)
 
 
 @router.get(

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

@@ -58,6 +58,7 @@
 
 import json
 import logging
+from collections.abc import Mapping
 from typing import Any
 
 from databricks.labs.dqx.errors import UnsafeSqlQueryError
@@ -373,8 +374,55 @@ def materialize_binding(self, binding_id: str) -> list[str]:
         self._cleanup_orphans(applied_ids=applied_ids, written_ids=written_ids)
         return sorted(written_ids)
 
+    def _resolve_registry(
+        self,
+        applied: AppliedRule,
+        *,
+        rules: Mapping[str, RegistryRule] | None = None,
+        versions: Mapping[tuple[str, int], RuleVersion] | None = None,
+    ) -> tuple[RegistryRule, int, RuleVersion] | None:
+        """Resolve an applied rule to its ``(registry_rule, version_number, snapshot)``.
+
+        This is the ONLY place the ``get_rule`` + ``get_version`` OLTP lookups
+        happen for a render. When *rules* / *versions* preloaded maps are
+        supplied (batch path), they are consulted instead of the per-id
+        ``RegistryService`` calls — so a batch caller pays two grouped queries
+        for N applications rather than ``2N`` sequential round-trips. The
+        resolution logic (missing rule, unpublished version, missing snapshot)
+        is identical either way, so single- and batch-path renders stay
+        byte-identical.
+
+        Returns ``None`` (with the same warning logs as before) when the applied
+        rule can't be resolved at all.
+        """
+        registry_rule = rules.get(applied.rule_id) if rules is not None else self._registry.get_rule(applied.rule_id)
+        if registry_rule is None or not applied.id:
+            logger.warning("Skipping applied rule %s: registry rule %s not found", applied.id, applied.rule_id)
+            return None
+
+        version_number = applied.pinned_version or registry_rule.version
+        if version_number <= 0:
+            logger.warning("Skipping applied rule %s: rule %s has no published version", applied.id, applied.rule_id)
+            return None
+
+        if versions is not None:
+            version_snapshot = versions.get((applied.rule_id, version_number))
+        else:
+            version_snapshot = self._registry.get_version(applied.rule_id, version_number)
+        if version_snapshot is None:
+            logger.warning(
+                "Skipping applied rule %s: version %d of rule %s not found", applied.id, version_number, applied.rule_id
+            )
+            return None
+        return registry_rule, version_number, version_snapshot
+
     def _iter_rendered_checks(
-        self, table_fqn: str, applied: AppliedRule
+        self,
+        table_fqn: str,
+        applied: AppliedRule,
+        *,
+        rules: Mapping[str, RegistryRule] | None = None,
+        versions: Mapping[tuple[str, int], RuleVersion] | None = None,
     ) -> list[tuple[str, str, dict[str, Any]]] | None:
         """Resolve + render an applied rule's mapping groups WITHOUT writing anything.
 
@@ -393,24 +441,20 @@ def _iter_rendered_checks(
         order — possibly EMPTY when every group failed to render, which the
         materializer treats as "this application now renders no rows" and
         cleans up accordingly.
-        """
-        registry_rule = self._registry.get_rule(applied.rule_id)
-        if registry_rule is None or not applied.id:
-            logger.warning("Skipping applied rule %s: registry rule %s not found", applied.id, applied.rule_id)
-            return None
-
-        version_number = applied.pinned_version or registry_rule.version
-        if version_number <= 0:
-            logger.warning("Skipping applied rule %s: rule %s has no published version", applied.id, applied.rule_id)
-            return None
 
-        version_snapshot = self._registry.get_version(applied.rule_id, version_number)
-        if version_snapshot is None:
-            logger.warning(
-                "Skipping applied rule %s: version %d of rule %s not found", applied.id, version_number, applied.rule_id
-            )
+        Registry resolution (the ``get_rule`` + ``get_version`` round-trips) is
+        delegated to :meth:`_resolve_registry`; passing a preloaded
+        *rules*/*versions* map lets a batch caller resolve every application's
+        registry rows in two grouped queries and share them here without any
+        per-application round-trip (see :meth:`render_binding_checks_many`).
+        """
+        resolved = self._resolve_registry(applied, rules=rules, versions=versions)
+        if resolved is None:
             return None
-
+        registry_rule, version_number, version_snapshot = resolved
+        # ``_resolve_registry`` already rejected a falsy ``applied.id`` (returning
+        # None), so it is a real str here — narrow it for the type checker.
+        applied_id = applied.id or ""
         effective_severity = (
             applied.severity_override or get_rule_severity(version_snapshot.user_metadata) or _DEFAULT_SEVERITY
         )
@@ -424,7 +468,7 @@ def _iter_rendered_checks(
         rendered_mode = version_snapshot.mode or registry_rule.mode
         rendered: list[tuple[str, str, dict[str, Any]]] = []
         for idx, group in enumerate(applied.column_mapping):
-            row_id = f"{applied.id}-{idx}"
+            row_id = f"{applied_id}-{idx}"
             try:
                 check, is_tableless = render_check(
                     mode=rendered_mode,
@@ -434,7 +478,7 @@ def _iter_rendered_checks(
                     per_application_tags=applied.user_metadata,
                     registry_rule_id=applied.rule_id,
                     registry_version=version_number,
-                    applied_rule_id=applied.id,
+                    applied_rule_id=applied_id,
                     app_settings=self._app_settings,
                 )
             except (ValueError, UnsafeSqlQueryError):
@@ -689,6 +733,68 @@ def render_binding_checks(self, binding_id: str) -> list[dict[str, Any]]:
             checks.extend(check for _row_id, _row_fqn, check in rendered)
         return checks
 
+    def render_binding_checks_counts_many(self, bindings: list[tuple[str, str]]) -> dict[str, int]:
+        """Batched draft-render CHECK COUNT for many *(binding_id, table_fqn)* pairs.
+
+        The bounded-query counterpart of calling :meth:`render_binding_checks`
+        in a loop: it produces the SAME per-binding count (the number of checks
+        a draft run would render) but resolves the registry rows for EVERY
+        binding's applications in a constant handful of grouped queries instead
+        of ``~3N + 2·ΣR`` sequential round-trips (N bindings, R applied rules
+        each):
+
+        1. ONE grouped ``dq_applied_rules`` query for all bindings' applications
+           (:meth:`MonitoredTableService.list_applied_rules_many`);
+        2. ONE ``dq_rules`` ``IN (...)`` query for the distinct rule ids
+           (:meth:`RegistryService.get_rules_many`);
+        3. ONE ``dq_rule_versions`` predicate-OR query for the distinct
+           ``(rule_id, version)`` pairs (:meth:`RegistryService.get_versions_many`).
+
+        Rendering then runs purely in-memory per binding through the SAME
+        :meth:`_iter_rendered_checks` path — so each count is byte-identical to
+        what the per-binding :meth:`render_binding_checks` would return. A
+        binding whose applications all fail to resolve counts 0. Bindings not
+        present in *bindings* are absent from the result; an empty input issues
+        no query and returns ``{{}}``.
+        """
+        if not bindings:
+            return {}
+        binding_fqns = {binding_id: table_fqn for binding_id, table_fqn in bindings}
+        applied_by_binding = self._monitored_tables.list_applied_rules_many(list(binding_fqns))
+
+        rule_ids: set[str] = set()
+        for applied_rules in applied_by_binding.values():
+            for applied in applied_rules:
+                if applied.id:
+                    rule_ids.add(applied.rule_id)
+        rules = self._registry.get_rules_many(rule_ids)
+
+        version_pairs: set[tuple[str, int]] = set()
+        for applied_rules in applied_by_binding.values():
+            for applied in applied_rules:
+                if not applied.id:
+                    continue
+                registry_rule = rules.get(applied.rule_id)
+                if registry_rule is None:
+                    continue
+                version_number = applied.pinned_version or registry_rule.version
+                if version_number > 0:
+                    version_pairs.add((applied.rule_id, version_number))
+        versions = self._registry.get_versions_many(version_pairs)
+
+        counts: dict[str, int] = {}
+        for binding_id, table_fqn in binding_fqns.items():
+            count = 0
+            for applied in applied_by_binding.get(binding_id, []):
+                if not applied.id:
+                    continue
+                rendered = self._iter_rendered_checks(table_fqn, applied, rules=rules, versions=versions)
+                if rendered is None:
+                    continue
+                count += len(rendered)
+            counts[binding_id] = count
+        return counts
+
     def rematerialize_for_rule(self, rule_id: str) -> list[str]:
         """Re-materialize every binding with a FOLLOWING (unpinned) application of *rule_id*.
 

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

@@ -629,6 +629,35 @@ def _list_applied_rules(self, binding_id: str) -> list[AppliedRuleSummary]:
             )
         return summaries
 
+    def list_applied_rules_many(self, binding_ids: list[str]) -> dict[str, list[AppliedRule]]:
+        """Applied rules for all *binding_ids* in ONE grouped query (no per-binding round-trip).
+
+        Unlike :meth:`_list_applied_rules`, this returns bare
+        :class:`AppliedRule` rows WITHOUT the per-rule descriptive-tag join
+        (name/dimension/severity): callers that only need the applications
+        themselves — e.g. the draft check-count render — must not pay one
+        ``dq_rules`` lookup per applied rule. Each binding's list preserves
+        ``created_at`` order to match :meth:`_list_applied_rules`; a binding
+        with no applied rows is simply absent from the result. An empty input
+        issues no query.
+        """
+        if not binding_ids:
+            return {}
+        distinct = sorted({b for b in binding_ids if b})
+        if not distinct:
+            return {}
+        in_list = ", ".join(f"'{escape_sql_string(b)}'" for b in distinct)
+        sql = (
+            f"SELECT {self._applied_select_cols} FROM {self._applied_table} "  # noqa: S608
+            f"WHERE binding_id IN ({in_list}) ORDER BY binding_id, created_at"
+        )
+        rows = self._sql.query(sql)
+        grouped: dict[str, list[AppliedRule]] = {}
+        for row in rows:
+            applied = self._row_to_applied_rule(row)
+            grouped.setdefault(applied.binding_id, []).append(applied)
+        return grouped
+
     def _rule_tags(self, rule_id: str) -> tuple[str | None, str | None, str | None]:
         """Look up name/dimension/severity tags for *rule_id* from ``dq_rules``.
 

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

@@ -20,6 +20,7 @@
 
 import json
 import logging
+from collections.abc import Collection
 from datetime import datetime, timezone
 from typing import Any, cast, get_args
 from uuid import uuid4
@@ -316,6 +317,62 @@ def get_version(self, rule_id: str, version: int) -> RuleVersion | None:
         """
         return self._get_version(rule_id, version)
 
+    def get_rules_many(self, rule_ids: Collection[str]) -> dict[str, RegistryRule]:
+        """Resolve many rules by id in ONE ``IN (...)`` query (no per-id round-trip).
+
+        Mirrors :meth:`get_rule` for a batch of ids — the value at each key is
+        exactly what :meth:`get_rule` would return for that id. Ids with no
+        ``dq_rules`` row are simply absent from the result, so callers can treat
+        a missing key the same as a ``None`` from :meth:`get_rule`. Duplicate
+        ids collapse to one predicate. An empty input issues no query.
+
+        Backs list-view batching (e.g. the monitored-tables draft check-count
+        pass) that must resolve the rules of many applications at once instead
+        of looping :meth:`get_rule` per application.
+        """
+        distinct = {rid for rid in rule_ids if rid}
+        if not distinct:
+            return {}
+        in_list = ", ".join(f"'{escape_sql_string(rid)}'" for rid in sorted(distinct))
+        sql = f"SELECT {self._select_cols} FROM {self._table} WHERE rule_id IN ({in_list})"  # noqa: S608
+        rows = self._sql.query(sql)
+        result: dict[str, RegistryRule] = {}
+        for row in rows:
+            rule = self._row_to_rule(row)
+            result[rule.rule_id] = rule
+        return result
+
+    def get_versions_many(self, pairs: Collection[tuple[str, int]]) -> dict[tuple[str, int], RuleVersion]:
+        """Resolve many frozen version snapshots by ``(rule_id, version)`` in ONE query.
+
+        The batch counterpart of :meth:`get_version`: the value at each
+        ``(rule_id, version)`` key equals what :meth:`get_version` would return
+        for that pair. Pairs with no ``dq_rule_versions`` row are absent from
+        the result. Duplicate pairs collapse to one predicate; an empty input
+        issues no query. Uses the same predicate-OR shape as
+        :meth:`_attach_modified` / :meth:`MonitoredTableVersionService.snapshot_counts_many`.
+        """
+        distinct = {(rid, int(ver)) for rid, ver in pairs if rid}
+        if not distinct:
+            return {}
+        definition = self._sql.select_json_text("definition")
+        user_metadata = self._sql.select_json_text("user_metadata")
+        created_at = self._sql.ts_text("created_at")
+        predicates = " OR ".join(
+            f"(rule_id = '{escape_sql_string(rid)}' AND version = {int(ver)})" for rid, ver in sorted(distinct)
+        )
+        sql = (
+            f"SELECT rule_id, version, {definition} AS definition_json, polarity, "  # noqa: S608
+            f"{user_metadata} AS user_metadata_json, created_by, {created_at}, mode "
+            f"FROM {self._versions_table} WHERE {predicates}"
+        )
+        rows = self._sql.query(sql)
+        result: dict[tuple[str, int], RuleVersion] = {}
+        for row in rows:
+            version = self._row_to_version(row)
+            result[(version.rule_id, version.version)] = version
+        return result
+
     def get_rule_with_version(self, rule_id: str) -> tuple[RegistryRule, RuleVersion | None] | None:
         """Get a registry rule plus its current published snapshot, if any.